1: <?php
  2: 
  3: namespace Guzzle\Batch;
  4: 
  5: use Guzzle\Batch\Exception\BatchTransferException;
  6: 
  7: /**
  8:  * Default batch implementation used to convert queued items into smaller chunks of batches using a
  9:  * {@see BatchDivisorIterface} and transfers each batch using a {@see BatchTransferInterface}.
 10:  *
 11:  * Any exception encountered during a flush operation will throw a {@see BatchTransferException} object containing the
 12:  * batch that failed. After an exception is encountered, you can flush the batch again to attempt to finish transferring
 13:  * any previously created batches or queued items.
 14:  */
 15: class Batch implements BatchInterface
 16: {
 17:     /**
 18:      * @var \SplQueue Queue of items in the queue
 19:      */
 20:     protected $queue;
 21: 
 22:     /**
 23:      * @var array Divided batches to be transferred
 24:      */
 25:     protected $dividedBatches;
 26: 
 27:     /**
 28:      * @var BatchTransferInterface
 29:      */
 30:     protected $transferStrategy;
 31: 
 32:     /**
 33:      * @var BatchDivisorInterface
 34:      */
 35:     protected $divisionStrategy;
 36: 
 37:     /**
 38:      * @param BatchTransferInterface $transferStrategy Strategy used to transfer items
 39:      * @param BatchDivisorInterface  $divisionStrategy Divisor used to create batches
 40:      */
 41:     public function __construct(BatchTransferInterface $transferStrategy, BatchDivisorInterface $divisionStrategy)
 42:     {
 43:         $this->transferStrategy = $transferStrategy;
 44:         $this->divisionStrategy = $divisionStrategy;
 45:         $this->queue = new \SplQueue();
 46:         $this->queue->setIteratorMode(\SplQueue::IT_MODE_DELETE);
 47:         $this->dividedBatches = array();
 48:     }
 49: 
 50:     /**
 51:      * {@inheritdoc}
 52:      */
 53:     public function add($item)
 54:     {
 55:         $this->queue->enqueue($item);
 56: 
 57:         return $this;
 58:     }
 59: 
 60:     /**
 61:      * {@inheritdoc}
 62:      */
 63:     public function flush()
 64:     {
 65:         $this->createBatches();
 66: 
 67:         $items = array();
 68:         foreach ($this->dividedBatches as $batchIndex => $dividedBatch) {
 69:             while ($dividedBatch->valid()) {
 70:                 $batch = $dividedBatch->current();
 71:                 $dividedBatch->next();
 72:                 try {
 73:                     $this->transferStrategy->transfer($batch);
 74:                     $items = array_merge($items, $batch);
 75:                 } catch (\Exception $e) {
 76:                     throw new BatchTransferException($batch, $items, $e, $this->transferStrategy, $this->divisionStrategy);
 77:                 }
 78:             }
 79:             // Keep the divided batch down to a minimum in case of a later exception
 80:             unset($this->dividedBatches[$batchIndex]);
 81:         }
 82: 
 83:         return $items;
 84:     }
 85: 
 86:     /**
 87:      * {@inheritdoc}
 88:      */
 89:     public function isEmpty()
 90:     {
 91:         return count($this->queue) == 0 && count($this->dividedBatches) == 0;
 92:     }
 93: 
 94:     /**
 95:      * Create batches for any queued items
 96:      */
 97:     protected function createBatches()
 98:     {
 99:         if (count($this->queue)) {
100:             if ($batches = $this->divisionStrategy->createBatches($this->queue)) {
101:                 // Convert arrays into iterators
102:                 if (is_array($batches)) {
103:                     $batches = new \ArrayIterator($batches);
104:                 }
105:                 $this->dividedBatches[] = $batches;
106:             }
107:         }
108:     }
109: }
110: