[RFC] Batch

.travis.yml

@@ -6,11 +6,16 @@ php:
     - 5.3
     - 5.4
     - 5.5
+    - 5.6
+    - hhvm
+
+matrix:
+    allow_failures:
+        - php: hhvm
 
 before_script:
-    - "pyrus install pecl/redis && pyrus build pecl/redis"
-    - "echo \"extension=redis.so\" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini"
-    - "echo '{\"config\":{\"github-oauth\": {\"github.com\":\"fa354b2f030836334eac842f8fd69a03e353d247\"}}}' > ~/.composer/config.json"
-    - "composer install --dev --no-progress --no-custom-installers --no-scripts"
+    - sh -c "if [ \"$TRAVIS_PHP_VERSION\" != \"hhvm\" ]; then pyrus install pecl/redis && pyrus build pecl/redis; fi"
+    - sh -c "if [ \"$TRAVIS_PHP_VERSION\" != \"hhvm\" ]; then echo \"extension=redis.so\" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini; fi"
+    - "composer install --no-progress --no-plugins"
 
 script: phpunit

src/Bernard/Batch.php

@@ -0,0 +1,70 @@
+<?php
+
+namespace Bernard;
+
+use Bernard\Batch\Storage;
+use Bernard\Batch\Status;
+
+final class Batch implements Message
+{
+    private $name;
+    private $description;
+    private $status;
+    private $envelopes = array();
+
+    public function __construct($name = null, $description = '', Status $status = null)
+    {
+        $this->name        = $name ?: uniqid('bernard_batch', true);
+        $this->description = $description;
+        $this->status      = $status ?: new Status(0, 0, 0);
+    }
+
+    public function assign(Message $message)
+    {
+        $this->envelopes[] = new Envelope($message, array('batch' => $this->name));
+
+        $this->status = new Status($this->status->total + 1, $this->status->failed, $this->status->successful);
+    }
+
+    /**
+     * Internal method used by the producer to produce the envelopes
+     * through its mnamedlewares.
+     *
+     * @return array
+     */
+    public function flush()
+    {
+        $envelopes = $this->envelopes;
+        $this->envelopes = array();
+
+        return $envelopes;
+    }
+
+    public function isRunning()
+    {
+        return $this->status->isRunning();
+    }
+
+    public function isComplete()
+    {
+        return $this->status->isComplete();
+    }
+
+    public function getStatus()
+    {
+        return $this->status;
+    }
+
+    public function getDescription()
+    {
+        return $this->description;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getName()
+    {
+        return $this->name;
+    }
+}

src/Bernard/Batch/AbstractStorage.php

@@ -0,0 +1,16 @@
+<?php
+
+namespace Bernard\Batch;
+
+use Bernard\Batch;
+
+abstract class AbstractStorage implements Storage
+{
+    /**
+     * {@inheritDoc}
+     */
+    public function reload(Batch $batch)
+    {
+        return $this->find($batch->getName());
+    }
+}

src/Bernard/Batch/RedisStorage.php

@@ -0,0 +1,93 @@
+<?php
+
+namespace Bernard\Batch;
+
+use Bernard\Batch;
+use Redis;
+
+/**
+ * Storage specific to php redis extension.
+ *
+ * @package Bernard
+ */
+class RedisStorage extends AbstractStorage
+{
+    protected $redis;
+    protected $ttl;
+
+    /**
+     * @param Redis $redis
+     * @param integer $ttl
+     */
+    public function __construct(Redis $redis, $ttl = null)
+    {
+        $this->redis = $redis;
+        $this->ttl = $ttl ?: 3600 * 36;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function find($name)
+    {
+        $result = $this->redis->hmget($this->resolveKey($name), array('total', 'failed', 'successful', 'description'));
+
+        return new Batch($name, $result['description'], new Status($result['total'], $result['failed'], $result['successful']));
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function register($name)
+    {
+        $redis = $this->redis->multi();
+
+        // insert the batch name into the set if dosent exists all ready.
+        // we use sorted sets to cheat into not really expiring them.
+        // we update this on last insert into the batch, to make sure we
+        // dont add message and expire the batch instantanious.
+        $redis->zadd('batches', $this->ttl + time(), $name);
+        $redis->incr($this->resolveKey($name, 'total'));
+
+        // create sets with expire at
+        $redis->expire($this->resolveKey($name, 'total'), $this->ttl);
+        $redis->expire($this->resolveKey($name, 'failed'), $this->ttl);
+        $redis->expire($this->resolveKey($name, 'successful'), $this->ttl);
+
+        // fire
+        $redis->exec();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function increment($name, $type)
+    {
+        if (!in_array($type, array('failed', 'successful'))) {
+            // should be more precise
+            throw new \InvalidArgumentException();
+        }
+
+        // creates a new key if dosent exists, creates a new field with a
+        // 0 as default if not exists.
+        $this->redis->incr($this->resolveKey($name, $type));
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function all()
+    {
+        $batches = $this->redis->zrangebyscore('batches', time(), '+inf');
+
+        return array_map($batches, array($this, 'find'));
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    protected function resolveKey($name, $type = null)
+    {
+        return rtrim('batch:' . $name . ':' . $type, ':');
+    }
+}

src/Bernard/Batch/Status.php

@@ -0,0 +1,38 @@
+<?php
+
+namespace Bernard\Batch;
+
+/**
+ * ValueObject that holds different information about the current
+ * status of a Batch
+ *
+ * @package Bernard
+ */
+final class Status
+{
+    private $total;
+    private $failed;
+    private $successful;
+
+    public function __construct($total, $failed, $successful)
+    {
+        $this->total      = $total;
+        $this->failed     = $failed;
+        $this->successful = $successful;
+    }
+
+    public function isComplete()
+    {
+        return $this->total == $this->failed + $this->successful;
+    }
+
+    public function isRunning()
+    {
+        return false == $this->isComplete();
+    }
+
+    public function __get($property)
+    {
+        return $this->$property;
+    }
+}

src/Bernard/Batch/Storage.php

@@ -0,0 +1,50 @@
+<?php
+
+namespace Bernard\Batch;
+
+use Bernard\Batch;
+
+/**
+ * Handles storage support for Batch. Implement this interface
+ * to use another storage backend.
+ *
+ * @package Bernard
+ */
+interface Storage
+{
+    /**
+     * @param integer $id
+     * @return Batch
+     */
+    public function find($id);
+
+    /**
+     * Returns a new batch based on the given instance indentifier.
+     *
+     * @param Batch $batch
+     * @return Batch
+     */
+    public function reload(Batch $batch);
+
+    /**
+     * @return Batch[]
+     */
+    public function all();
+
+    /**
+     * Register a batch with the storage, this may
+     * be called multiple times. As it is called by the
+     * middleware everytime a batch have been added
+     *
+     * @param string $id
+     */
+    public function register($id);
+
+    /**
+     * Increments a counter for a specific batch.
+     *
+     * @param string $id
+     * @param string $type
+     */
+    public function increment($id, $type);
+}

src/Bernard/Envelope.php

@@ -13,15 +13,17 @@ class Envelope
     protected $message;
     protected $class;
     protected $timestamp;
+    protected $stamps;
 
     /**
      * @param Message $message
      */
-    public function __construct(Message $message)
+    public function __construct(Message $message, array $stamps = array())
     {
         $this->message   = $message;
         $this->class     = get_class($message);
         $this->timestamp = time();
+        $this->stamps    = array_filter($stamps, 'is_scalar');
     }
 
     /**
@@ -55,4 +57,26 @@ public function getTimestamp()
     {
         return $this->timestamp;
     }
+
+    /**
+     * @return array 
+     */
+    public function getStamps()
+    {
+        return $this->stamps;
+    }
+
+    /**
+     * @param string $name
+     * @param mixed $default
+     * @return mixed
+     */
+    public function getStamp($name, $default = null)
+    {
+        if (!isset($this->stamps[$name])) {
+            return $default;
+        }
+
+        return $this->stamps[$name];
+    }
 }

src/Bernard/EventListener/BatchSubscriber.php

@@ -0,0 +1,48 @@
+<?php
+
+namespace Bernard\EventListener;
+
+use Bernard\Event\EnvelopeEvent;
+use Bernard\Event\RejectEnvelopeEvent;
+use Bernard\Batch\Storage;
+use Bernard\Envelope;
+
+class BatchSubscriber implements \Symfony\Component\EventDispatcher\EventSubscriberInterface
+{
+    protected $storage;
+
+    public function __construct(Storage $storage)
+    {
+        $this->storage = $storage;
+    }
+
+    public function onProduce(EnvelopeEvent $event)
+    {
+        $envelope = $event->getEnvelope();
+
+        if ($batch = $envelope->getStamp('batch')) {
+            $this->storage->register($envelope->getStamp('batch'));
+        }
+    }
+
+    public function onAcknowledgeReject(EnvelopeEvent $event)
+    {
+        $envelope = $event->getEnvelope();
+
+        if (false == $batch = $envelope->getStamp('batch')) {
+            return;
+        }
+
+        $type = $event instanceof RejectEnvelopeEvent ? 'failed' : 'successful';
+        $this->storage->increment($batch, $type);
+    }
+
+    public static function getSubscribedEvents()
+    {
+        return array(
+            'bernard.produce'     => 'onProduce',
+            'bernard.acknowledge' => 'onAcknowledgeReject',
+            'bernard.reject'      => 'onAcknowledgeReject',
+        );
+    }
+}