Update to MediaWiki 1.26.0

29 files changed, 5614 insertions, 0 deletions

diff --git a/vendor/nmred/kafka-php/src/Kafka/Client.php b/vendor/nmred/kafka-php/src/Kafka/Client.php
new file mode 100644
index 00000000..a38e705b
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Client.php
@@ -0,0 +1,290 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Client
+{
+    // {{{ consts
+    // }}}
+    // {{{ members
+
+    /**
+     * cluster metadata
+     *
+     * @var \Kafka\ClusterMetaData
+     * @access private
+     */
+    private $metadata = null;
+
+    /**
+     * broker host list
+     *
+     * @var array
+     * @access private
+     */
+    private $hostList = array();
+
+    /**
+     * save broker connection
+     *
+     * @var array
+     * @access private
+     */
+    private static $stream = array();
+
+    /**
+     * default stream options
+     *
+     * @var array
+     * @access private
+     */
+    private $streamOptions = array(
+        'RecvTimeoutSec' => 0,
+        'RecvTimeoutUsec' => 750000,
+        'SendTimeoutSec' => 0,
+        'SendTimeoutUsec' => 100000,
+    );
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @access public
+     * @return void
+     */
+    public function __construct(ClusterMetaData $metadata)
+    {
+        $this->metadata = $metadata;
+        if (method_exists($metadata, 'setClient')) {
+            $this->metadata->setClient($this);
+        }
+    }
+
+    /**
+     * update stream options
+     *
+     * @param array $options
+     */
+    public function setStreamOptions($options = array())
+    {
+        // Merge the arrays
+        $this->streamOptions = array_merge($this->streamOptions, $options);
+        $this->updateStreamOptions();
+    }
+
+    /**
+     * @access public
+     * @param $name - name of stream option
+     * @param $value - value for option
+     */
+    public function setStreamOption($name, $value)
+    {
+        $this->streamOptions[$name] = $value;
+        $this->updateStreamOptions();
+    }
+
+    /**
+     * @access public
+     * @param $name - name of option
+     * @return mixed
+     */
+    public function getStreamOption($name)
+    {
+        if (array_key_exists($name, $this->streamOptions)) {
+            return $this->streamOptions[$name];
+        }
+        return null;
+    }
+
+    /**
+     * @access private
+     */
+    private function updateStreamOptions()
+    {
+        // Loop thru each stream
+        foreach (self::$stream as $host => $streams) {
+            foreach ($streams as $key => $info) {
+                // Update options
+                if (isset($info['stream'])) {
+                    /** @var \Kafka\Socket $stream */
+                    $stream = $info['stream'];
+                    $stream->setRecvTimeoutSec($this->streamOptions['RecvTimeoutSec']);
+                    $stream->setRecvTimeoutUsec($this->streamOptions['SendTimeoutUsec']);
+                    $stream->setSendTimeoutSec($this->streamOptions['SendTimeoutSec']);
+                    $stream->setSendTimeoutUsec($this->streamOptions['SendTimeoutUsec']);
+                }
+            }
+        }
+    }
+
+    // }}}
+    // {{{ public function getBrokers()
+
+    /**
+     * get broker server
+     *
+     * @access public
+     * @return void
+     */
+    public function getBrokers()
+    {
+        if (empty($this->hostList)) {
+            $brokerList = $this->metadata->listBrokers();
+            foreach ($brokerList as $brokerId => $info) {
+                if (!isset($info['host']) || !isset($info['port'])) {
+                    continue;
+                }
+                $this->hostList[$brokerId] = $info['host'] . ':' . $info['port'];
+            }
+        }
+
+        return $this->hostList;
+    }
+
+    // }}}
+    // {{{ public function getHostByPartition()
+
+    /**
+     * get broker host by topic partition
+     *
+     * @param string $topicName
+     * @param int $partitionId
+     * @access public
+     * @return string
+     */
+    public function getHostByPartition($topicName, $partitionId = 0)
+    {
+        $partitionInfo = $this->metadata->getPartitionState($topicName, $partitionId);
+        if (!$partitionInfo) {
+            throw new \Kafka\Exception('topic:' . $topicName . ', partition id: ' . $partitionId . ' is not exists.');
+        }
+
+        $hostList = $this->getBrokers();
+        if (isset($partitionInfo['leader']) && isset($hostList[$partitionInfo['leader']])) {
+            return $hostList[$partitionInfo['leader']];
+        } else {
+            throw new \Kafka\Exception('can\'t find broker host.');
+        }
+    }
+
+    // }}}
+    // {{{ public function getZooKeeper()
+
+    /**
+     * get kafka zookeeper object
+     *
+     * @access public
+     * @return \Kafka\ZooKeeper
+     */
+    public function getZooKeeper()
+    {
+        if ($this->metadata instanceof \Kafka\ZooKeeper) {
+                return $this->metadata;
+        } else {
+                throw new \Kafka\Exception( 'ZooKeeper was not provided' );
+        }
+    }
+
+    // }}}
+    // {{{ public function getStream()
+
+    /**
+     * get broker broker connect
+     *
+     * @param string $host
+     * @access private
+     * @return void
+     */
+    public function getStream($host, $lockKey = null)
+    {
+        if (!$lockKey) {
+            $lockKey = uniqid($host);
+        }
+
+        list($hostname, $port) = explode(':', $host);
+        // find unlock stream
+        if (isset(self::$stream[$host])) {
+            foreach (self::$stream[$host] as $key => $info) {
+                if ($info['locked']) {
+                    continue;
+                } else {
+                    self::$stream[$host][$key]['locked'] = true;
+                    $info['stream']->connect();
+                    return array('key' => $key, 'stream' => $info['stream']);
+                }
+            }
+        }
+
+        // no idle stream
+        $stream = new \Kafka\Socket($hostname, $port, $this->getStreamOption('RecvTimeoutSec'), $this->getStreamOption('RecvTimeoutUsec'), $this->getStreamOption('SendTimeoutSec'), $this->getStreamOption('SendTimeoutUsec'));
+        $stream->connect();
+        self::$stream[$host][$lockKey] = array(
+            'locked' => true,
+            'stream' => $stream,
+        );
+        return array('key' => $lockKey, 'stream' => $stream);
+    }
+
+    // }}}
+    // {{{ public function freeStream()
+
+    /**
+     * free stream pool
+     *
+     * @param string $key
+     * @access public
+     * @return void
+     */
+    public function freeStream($key)
+    {
+        foreach (self::$stream as $host => $values) {
+            if (isset($values[$key])) {
+                self::$stream[$host][$key]['locked'] = false;
+            }
+        }
+    }
+
+    // }}}
+    // {{{ public function getTopicDetail()
+
+    /**
+     * get topic detail info
+     *
+     * @param  string $topicName
+     * @return array
+     */
+    public function getTopicDetail($topicName)
+    {
+        return $this->metadata->getTopicDetail($topicName);
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/ClusterMetaData.php b/vendor/nmred/kafka-php/src/Kafka/ClusterMetaData.php
new file mode 100644
index 00000000..e9b3d064
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/ClusterMetaData.php
@@ -0,0 +1,53 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+/**
++------------------------------------------------------------------------------
+* Metadata about the kafka cluster
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author ebernhardson@wikimedia.org
++------------------------------------------------------------------------------
+*/
+
+interface ClusterMetaData
+{
+    /**
+     * get broker list from kafka metadata
+     *
+     * @access public
+     * @return array
+     */
+    public function listBrokers();
+
+    /**
+     * @param string $topicName
+     * @param integer $partitionId
+     * @access public
+     * @return array
+     */
+    public function getPartitionState($topicName, $partitionId = 0);
+
+    /**
+     * @param string $topicName
+     * @access public
+     * @return array
+     */
+    public function getTopicDetail($topicName);
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Consumer.php b/vendor/nmred/kafka-php/src/Kafka/Consumer.php
new file mode 100644
index 00000000..5ff2d43f
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Consumer.php
@@ -0,0 +1,378 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Consumer
+{
+    // {{{ consts
+    // }}}
+    // {{{ members
+
+    /**
+     * client
+     *
+     * @var mixed
+     * @access private
+     */
+    private $client = null;
+
+    /**
+     * send message options cache
+     *
+     * @var array
+     * @access private
+     */
+    private $payload = array();
+
+    /**
+     * consumer group
+     *
+     * @var string
+     * @access private
+     */
+    private $group = '';
+
+    /**
+     * from offset
+     *
+     * @var mixed
+     * @access private
+     */
+    private $fromOffset = true;
+
+    /**
+     * produce instance
+     *
+     * @var \Kafka\Produce
+     * @access private
+     */
+    private static $instance = null;
+
+    /**
+     * broker host list
+     *
+     * @var array
+     * @access private
+     */
+    private $hostList = array();
+
+    /**
+     * save broker connection
+     *
+     * @var array
+     * @access private
+     */
+    private $stream = array();
+
+    /**
+     * maxSize
+     *
+     * @var integer
+     */
+    private $maxSize = 1048576;
+
+    /**
+     * offsetStrategy
+     * @var integer
+     */
+    private $offsetStrategy = \Kafka\Offset::DEFAULT_EARLY;
+
+    // }}}
+    // {{{ functions
+    // {{{ public function static getInstance()
+
+    /**
+     * set send messages
+     *
+     * @access public
+     * @return void
+     */
+    public static function getInstance($hostList, $timeout = null)
+    {
+        if (is_null(self::$instance)) {
+            self::$instance = new self($hostList, $timeout);
+        }
+
+        return self::$instance;
+    }
+
+    // }}}
+    // {{{ private function __construct()
+
+    /**
+     * __construct
+     *
+     * @access public
+     * @return void
+     */
+    private function __construct($hostList, $timeout = null)
+    {
+        $zookeeper = new \Kafka\ZooKeeper($hostList, $timeout);
+        $this->client = new \Kafka\Client($zookeeper);
+    }
+
+    // }}}
+    // {{{ public function clearPayload()
+
+    /**
+     * clearPayload
+     *
+     * @access public
+     * @return void
+     */
+    public function clearPayload()
+    {
+        $this->payload = array();
+    }
+
+    // }}}
+    // {{{ public function setTopic()
+
+    /**
+     * set topic name
+     *
+     * @access public
+     * @return void
+     */
+    public function setTopic($topicName, $defaultOffset = null)
+    {
+        $parts = $this->client->getTopicDetail($topicName);
+        if (!isset($parts['partitions']) || empty($parts['partitions'])) {
+            // set topic fail.
+            return $this;
+        }
+
+        foreach ($parts['partitions'] as $partId => $info) {
+            $this->setPartition($topicName, $partId, $defaultOffset);
+        }
+
+        return $this;
+    }
+
+    // }}}
+    // {{{ public function setPartition()
+
+    /**
+     * set topic partition
+     *
+     * @access public
+     * @return void
+     */
+    public function setPartition($topicName, $partitionId = 0, $offset = null)
+    {
+        if (is_null($offset)) {
+            if ($this->fromOffset) {
+                $offsetObject = new \Kafka\Offset($this->client, $this->group, $topicName, $partitionId);
+                $offset = $offsetObject->getOffset($this->offsetStrategy);
+                \Kafka\Log::log('topic name:' . $topicName . ', part:' . $partitionId . 'get offset from kafka server, offet:' . $offset, LOG_DEBUG);
+            } else {
+                $offset = 0;
+            }
+        }
+        $this->payload[$topicName][$partitionId] = $offset;
+
+        return $this;
+    }
+
+    // }}}
+    // {{{ public function setFromOffset()
+
+    /**
+     * set whether starting offset fetch
+     *
+     * @param boolean $fromOffset
+     * @access public
+     * @return void
+     */
+    public function setFromOffset($fromOffset)
+    {
+        $this->fromOffset = (boolean) $fromOffset;
+    }
+
+    // }}}
+    // {{{ public function setMaxBytes()
+
+    /**
+     * set fetch message max bytes
+     *
+     * @param int $maxSize
+     * @access public
+     * @return void
+     */
+    public function setMaxBytes($maxSize)
+    {
+        $this->maxSize = $maxSize;
+    }
+
+    // }}}
+    // {{{ public function setGroup()
+
+    /**
+     * set consumer group
+     *
+     * @param string $group
+     * @access public
+     * @return void
+     */
+    public function setGroup($group)
+    {
+        $this->group = (string) $group;
+        return $this;
+    }
+
+    // }}}
+    // {{{ public function fetch()
+
+    /**
+     * fetch message to broker
+     *
+     * @access public
+     * @return void
+     */
+    public function fetch()
+    {
+        $data = $this->_formatPayload();
+        if (empty($data)) {
+            return false;
+        }
+
+        $responseData = array();
+        $streams = array();
+        foreach ($data as $host => $requestData) {
+            $connArr = $this->client->getStream($host);
+            $conn    = $connArr['stream'];
+            $encoder = new \Kafka\Protocol\Encoder($conn);
+            $encoder->fetchRequest($requestData);
+            $streams[$connArr['key']] = $conn;
+        }
+
+        $fetch = new \Kafka\Protocol\Fetch\Topic($streams, $data);
+
+        // register fetch helper
+        $freeStream = new \Kafka\Protocol\Fetch\Helper\FreeStream($this->client);
+        $freeStream->setStreams($streams);
+        \Kafka\Protocol\Fetch\Helper\Helper::registerHelper('freeStream', $freeStream);
+
+        // register partition commit offset
+        $commitOffset = new \Kafka\Protocol\Fetch\Helper\CommitOffset($this->client);
+        $commitOffset->setGroup($this->group);
+        \Kafka\Protocol\Fetch\Helper\Helper::registerHelper('commitOffset', $commitOffset);
+
+        $updateConsumer = new \Kafka\Protocol\Fetch\Helper\Consumer($this);
+        \Kafka\Protocol\Fetch\Helper\Helper::registerHelper('updateConsumer', $updateConsumer);
+
+        return $fetch;
+    }
+
+    // }}}
+    // {{{ public function getClient()
+
+    /**
+     * get client object
+     *
+     * @access public
+     * @return void
+     */
+    public function getClient()
+    {
+        return $this->client;
+    }
+
+    /**
+     * passthru method to client for setting stream options
+     *
+     * @param array $options
+     */
+    public function setStreamOptions($options = array())
+    {
+        $this->client->setStreamOptions($options);
+    }
+
+    // }}}
+    // {{{ private function _formatPayload()
+
+    /**
+     * format payload array
+     *
+     * @access private
+     * @return array
+     */
+    private function _formatPayload()
+    {
+        if (empty($this->payload)) {
+            return array();
+        }
+
+        $data = array();
+        foreach ($this->payload as $topicName => $partitions) {
+            foreach ($partitions as $partitionId => $offset) {
+                $host = $this->client->getHostByPartition($topicName, $partitionId);
+                $data[$host][$topicName][$partitionId] = $offset;
+            }
+        }
+
+        $requestData = array();
+        foreach ($data as $host => $info) {
+            $topicData = array();
+            foreach ($info as $topicName => $partitions) {
+                $partitionData = array();
+                foreach ($partitions as $partitionId => $offset) {
+                    $partitionData[] = array(
+                        'partition_id' => $partitionId,
+                        'offset'       => $offset,
+                        'max_bytes'    => $this->maxSize,
+                    );
+                }
+                $topicData[] = array(
+                    'topic_name' => $topicName,
+                    'partitions' => $partitionData,
+                );
+            }
+
+            $requestData[$host] = array(
+                'data' => $topicData,
+            );
+        }
+
+       return $requestData;
+    }
+
+    /**
+     * const LAST_OFFSET = -1;
+     * const EARLIEST_OFFSET = -2;
+     * const DEFAULT_LAST  = -2;
+     * const DEFAULT_EARLY = -1;
+     * @param type $offsetStrategy
+     */
+    public function setOffsetStrategy($offsetStrategy)
+    {
+        $this->offsetStrategy = $offsetStrategy;
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Exception.php b/vendor/nmred/kafka-php/src/Kafka/Exception.php
new file mode 100644
index 00000000..f336f1c3
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Exception.php
@@ -0,0 +1,31 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Exception extends \RuntimeException
+{
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Exception/NotSupported.php b/vendor/nmred/kafka-php/src/Kafka/Exception/NotSupported.php
new file mode 100644
index 00000000..011129a2
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Exception/NotSupported.php
@@ -0,0 +1,33 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Exception;
+
+use \Kafka\Exception;
+
+/**
++------------------------------------------------------------------------------
+* Kafka php client exception
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class NotSupported extends \Exception
+{
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Exception/OutOfRange.php b/vendor/nmred/kafka-php/src/Kafka/Exception/OutOfRange.php
new file mode 100644
index 00000000..374d1538
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Exception/OutOfRange.php
@@ -0,0 +1,33 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Exception;
+
+use \Kafka\Exception;
+
+/**
++------------------------------------------------------------------------------
+* Kafka php client exception
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class OutOfRange extends Exception
+{
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Exception/Protocol.php b/vendor/nmred/kafka-php/src/Kafka/Exception/Protocol.php
new file mode 100644
index 00000000..6e213f05
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Exception/Protocol.php
@@ -0,0 +1,33 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Exception;
+
+use \Kafka\Exception;
+
+/**
++------------------------------------------------------------------------------
+* Kafka php client exception
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Protocol extends \Exception
+{
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Exception/Socket.php b/vendor/nmred/kafka-php/src/Kafka/Exception/Socket.php
new file mode 100644
index 00000000..aca93e25
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Exception/Socket.php
@@ -0,0 +1,33 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Exception;
+
+use \Kafka\Exception;
+
+/**
++------------------------------------------------------------------------------
+* Kafka php client exception
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Socket extends \Exception
+{
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Exception/SocketConnect.php b/vendor/nmred/kafka-php/src/Kafka/Exception/SocketConnect.php
new file mode 100644
index 00000000..476b2a48
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Exception/SocketConnect.php
@@ -0,0 +1,33 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Exception;
+
+use \Kafka\Exception;
+
+/**
++------------------------------------------------------------------------------
+ * Kafka php client exception
++------------------------------------------------------------------------------
+ *
+ * @package
+ * @version $_SWANBR_VERSION_$
+ * @copyright Copyleft
+ * @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+ */
+
+class SocketConnect extends Socket
+{
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Exception/SocketEOF.php b/vendor/nmred/kafka-php/src/Kafka/Exception/SocketEOF.php
new file mode 100644
index 00000000..35e34e07
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Exception/SocketEOF.php
@@ -0,0 +1,33 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Exception;
+
+use \Kafka\Exception;
+
+/**
++------------------------------------------------------------------------------
+* Kafka php client exception
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class SocketEOF extends Exception
+{
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Exception/SocketTimeout.php b/vendor/nmred/kafka-php/src/Kafka/Exception/SocketTimeout.php
new file mode 100644
index 00000000..b0d38c2c
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Exception/SocketTimeout.php
@@ -0,0 +1,33 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Exception;
+
+use \Kafka\Exception;
+
+/**
++------------------------------------------------------------------------------
+* Kafka php client exception
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class SocketTimeout extends Exception
+{
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Log.php b/vendor/nmred/kafka-php/src/Kafka/Log.php
new file mode 100644
index 00000000..481bfc3d
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Log.php
@@ -0,0 +1,78 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Log
+{
+    // {{{ consts
+    // }}}
+    // {{{ members
+
+    /**
+     * log
+     *
+     * @var mixed
+     * @access private
+     */
+    private static $log = null;
+
+    // }}}
+    // {{{ functions
+    // {{{ public static function setLog()
+
+    /**
+     * setLog
+     *
+     * @access public
+     * @return void
+     */
+    public static function setLog($log)
+    {
+        if ($log) {
+            self::$log = $log;
+        }
+    }
+
+    // }}}
+    // {{{ public static function log()
+
+    /**
+     * log
+     *
+     * @access public
+     * @return void
+     */
+    public static function log($message, $level = LOG_DEBUG)
+    {
+        if (self::$log && method_exists(self::$log, 'log')) {
+            self::$log->log($message, $level);
+        }
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/MetaDataFromKafka.php b/vendor/nmred/kafka-php/src/Kafka/MetaDataFromKafka.php
new file mode 100644
index 00000000..9d2c613e
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/MetaDataFromKafka.php
@@ -0,0 +1,200 @@
+<?php
+
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+/**
++------------------------------------------------------------------------------
+* Cluster metadata provided by kafka
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author ebernhardson@wikimedia.org
++------------------------------------------------------------------------------
+*/
+
+class MetaDataFromKafka implements ClusterMetaData
+{
+    // {{{ consts
+    // }}}
+    // {{{ members
+
+    /**
+     * client
+     *
+     * @var \Kafka\Client
+     * @access private
+     */
+    private $client;
+
+    /**
+     * list of kafka brokers to get metadata from
+     *
+     * @var array
+     * @access private
+     */
+    private $hostList;
+
+    /**
+     * List of all kafka brokers
+     *
+     * @var array
+     * @access private
+     */
+    private $brokers = array();
+
+    /**
+     * List of all loaded topic metadata
+     *
+     * @var array
+     * @access private
+     */
+    private $topics = array();
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * @var string|array $hostList List of kafka brokers to get metadata from
+     * @access public
+     */
+    public function __construct($hostList)
+    {
+        if (is_string($hostList)) { // support host list 127.0.0.1:9092,192.168.2.11:9092 form
+            $this->hostList = explode(',', $hostList);
+        } else {
+            $this->hostList = (array)$hostList;
+        }
+        // randomize the order of servers we collect metadata from
+        shuffle($this->hostList);
+    }
+
+    // }}}
+    // {{{ public function setClient()
+
+    /**
+     * @var \Kafka\Client $client
+     * @access public
+     * @return void
+     */
+    public function setClient(\Kafka\Client $client)
+    {
+        $this->client = $client;
+    }
+
+    // }}}
+    // {{{ public function listBrokers()
+
+    /**
+     * get broker list from kafka metadata
+     *
+     * @access public
+     * @return array
+     */
+    public function listBrokers()
+    {
+        if ($this->brokers === null) {
+            $this->loadBrokers();
+        }
+        return $this->brokers;
+    }
+
+    // }}}
+    // {{{ public function getPartitionState()
+
+    public function getPartitionState($topicName, $partitionId = 0)
+    {
+        if (!isset( $this->topics[$topicName] ) ) {
+            $this->loadTopicDetail(array($topicName));
+        }
+        if ( isset( $this->topics[$topicName]['partitions'][$partitionId] ) ) {
+            return $this->topics[$topicName]['partitions'][$partitionId];
+        } else {
+            return null;
+        }
+    }
+
+    // }}}
+    // {{{ public function getTopicDetail()
+
+    /**
+     *
+     * @param string $topicName
+     * @access public
+     * @return array
+     */
+    public function getTopicDetail($topicName)
+    {
+        if (!isset( $this->topics[$topicName] ) ) {
+            $this->loadTopicDetail(array($topicName));
+        }
+        if (isset( $this->topics[$topicName] ) ) {
+            return $this->topics[$topicName];
+        } else {
+            return array();
+        }
+    }
+
+    // }}}
+    // {{{ private function loadBrokers()
+
+    private function loadBrokers()
+    {
+        $this->brokers = array();
+        // not sure how to ask for only the brokers without a topic...
+        // just ask for a topic we don't care about
+        $this->loadTopicDetail(array('test'));
+    }
+
+    // }}}
+    // {{{ private function loadTopicDetail()
+
+    private function loadTopicDetail(array $topics)
+    {
+        if ($this->client === null) {
+            throw new \Kafka\Exception('client was not provided');
+        }
+        $response = null;
+        foreach ($this->hostList as $host) {
+            try {
+                $response = null;
+                $stream = $this->client->getStream($host);
+                $conn = $stream['stream'];
+                $encoder = new \Kafka\Protocol\Encoder($conn);
+                $encoder->metadataRequest($topics);
+                $decoder = new \Kafka\Protocol\Decoder($conn);
+                $response = $decoder->metadataResponse();
+                $this->client->freeStream($stream['key']);
+                break;
+            } catch (\Kafka\Exception $e) {
+                // keep trying
+            }
+        }
+        if ($response) {
+            // Merge arrays using "+" operator to preserve key (which are broker IDs)
+            // instead of array_merge (which reindex numeric keys)
+            $this->brokers = $response['brokers'] + $this->brokers;
+            $this->topics = array_merge($response['topics'], $this->topics);
+        } else {
+            throw new \Kafka\Exception('Could not connect to any kafka brokers');
+        }
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Offset.php b/vendor/nmred/kafka-php/src/Kafka/Offset.php
new file mode 100644
index 00000000..7ad3f9d8
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Offset.php
@@ -0,0 +1,305 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+use \Kafka\Log;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Offset
+{
+    // {{{ consts
+
+    /**
+     * receive the latest offset
+     */
+    const LAST_OFFSET = -1;
+
+    /**
+     *   receive the earliest available offset.
+     */
+    const EARLIEST_OFFSET = -2;
+
+    /**
+     * function getOffset if read invalid value use latest offset instead of
+     */
+    const DEFAULT_LAST  = -2;
+
+    /**
+     * function getOffset if read invalid value use earliest offset instead of
+     */
+    const DEFAULT_EARLY = -1;
+
+    // }}}
+    // {{{ members
+
+    /**
+     * client
+     *
+     * @var mixed
+     * @access private
+     */
+    private $client = null;
+
+    /**
+     * consumer group
+     *
+     * @var string
+     * @access private
+     */
+    private $groupId = '';
+
+    /**
+     * topic name
+     *
+     * @var string
+     * @access private
+     */
+    private $topicName = '';
+
+    /**
+     * topic partition id, default 0
+     *
+     * @var float
+     * @access private
+     */
+    private $partitionId = 0;
+
+    /**
+     * encoder
+     *
+     * @var mixed
+     * @access private
+     */
+    private $encoder = null;
+
+    /**
+     * decoder
+     *
+     * @var mixed
+     * @access private
+     */
+    private $decoder = null;
+
+    /**
+     * streamKey
+     *
+     * @var string
+     * @access private
+     */
+    private $streamKey = '';
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @access public
+     * @return void
+     */
+    public function __construct($client, $groupId, $topicName, $partitionId = 0)
+    {
+        $this->client      = $client;
+        $this->groupId     = $groupId;
+        $this->topicName   = $topicName;
+        $this->partitionId = $partitionId;
+
+        $host   = $this->client->getHostByPartition($topicName, $partitionId);
+        $stream = $this->client->getStream($host);
+        $conn   = $stream['stream'];
+        $this->streamKey = $stream['key'];
+        $this->encoder = new \Kafka\Protocol\Encoder($conn);
+        $this->decoder = new \Kafka\Protocol\Decoder($conn);
+    }
+
+    // }}}
+    // {{{ public function setOffset()
+
+    /**
+     * set consumer offset
+     *
+     * @param integer $offset
+     * @access public
+     * @return void
+     */
+    public function setOffset($offset)
+    {
+        $maxOffset = $this->getProduceOffset();
+        if ($offset > $maxOffset) {
+            throw new \Kafka\Exception('this offset is invalid. must less than max offset:' . $maxOffset);
+        }
+
+        $data = array(
+            'group_id' => $this->groupId,
+            'data' => array(
+                array(
+                    'topic_name' => $this->topicName,
+                    'partitions' => array(
+                        array(
+                            'partition_id' => $this->partitionId,
+                            'offset' => $offset,
+                            ),
+                        ),
+                ),
+            ),
+        );
+
+        $topicName = $this->topicName;
+        $partitionId = $this->partitionId;
+
+        $this->encoder->commitOffsetRequest($data);
+        $result = $this->decoder->commitOffsetResponse();
+        $this->client->freeStream($this->streamKey);
+        if (!isset($result[$topicName][$partitionId]['errCode'])) {
+            throw new \Kafka\Exception('commit topic offset failed.');
+        }
+        if ($result[$topicName][$partitionId]['errCode'] != 0) {
+            throw new \Kafka\Exception(\Kafka\Protocol\Decoder::getError($result[$topicName][$partitionId]['errCode']));
+        }
+    }
+
+    // }}}
+    // {{{ public function getOffset()
+
+    /**
+     * get consumer offset
+     *
+     * @param integer $defaultOffset
+     *   if defaultOffset -1 instead of early offset
+     *   if defaultOffset -2 instead of last offset
+     * @access public
+     * @return void
+     */
+    public function getOffset($defaultOffset = self::DEFAULT_LAST)
+    {
+        $maxOffset = $this->getProduceOffset(self::LAST_OFFSET);
+        $minOffset = $this->getProduceOffset(self::EARLIEST_OFFSET);
+        $data = array(
+            'group_id' => $this->groupId,
+            'data' => array(
+                array(
+                    'topic_name' => $this->topicName,
+                    'partitions' => array(
+                        array(
+                            'partition_id' => $this->partitionId,
+                        ),
+                    ),
+                ),
+            ),
+        );
+
+        $this->encoder->fetchOffsetRequest($data);
+        $result = $this->decoder->fetchOffsetResponse();
+        $this->client->freeStream($this->streamKey);
+
+        $topicName = $this->topicName;
+        $partitionId = $this->partitionId;
+        if (!isset($result[$topicName][$partitionId]['errCode'])) {
+            throw new \Kafka\Exception('fetch topic offset failed.');
+        }
+        if ($result[$topicName][$partitionId]['errCode'] == 3) {
+            switch ($defaultOffset) {
+                case self::DEFAULT_LAST:
+                    return $maxOffset;
+                    Log::log("topic name: $topicName, partitionId: $partitionId, get offset value is default last.", LOG_INFO);
+                case self::DEFAULT_EARLY:
+                    Log::log("topic name: $topicName, partitionId: $partitionId, get offset value is default early.", LOG_INFO);
+                    return $minOffset;
+                default:
+                    $this->setOffset($defaultOffset);
+                    Log::log("topic name: $topicName, partitionId: $partitionId, get offset value is default $defaultOffset.", LOG_INFO);
+                    return $defaultOffset;
+            }
+            if ($defaultOffset) {
+                $this->setOffset($defaultOffset);
+                return $defaultOffset;
+            }
+        } elseif ($result[$topicName][$partitionId]['errCode'] == 0) {
+            $offset = $result[$topicName][$partitionId]['offset'];
+            if ($offset > $maxOffset || $offset < $minOffset) {
+                if ($defaultOffset == self::DEFAULT_EARLY) {
+                    $offset = $minOffset;
+                } else {
+                    $offset = $maxOffset;
+                }
+            }
+            Log::log("topic name: $topicName, partitionId: $partitionId, get offset value is $offset.", LOG_INFO);
+
+            return $offset;
+        } else {
+            throw new \Kafka\Exception(\Kafka\Protocol\Decoder::getError($result[$topicName][$partitionId]['errCode']));
+        }
+    }
+
+    // }}}
+    // {{{ public function getProduceOffset()
+
+    /**
+     * get produce server offset
+     *
+     * @param string $topicName
+     * @param integer $partitionId
+     * @access public
+     * @return int
+     */
+    public function getProduceOffset($timeLine = self::LAST_OFFSET)
+    {
+        $topicName = $this->topicName;
+        $partitionId = $this->partitionId;
+
+        $requestData = array(
+            'data' => array(
+                array(
+                    'topic_name' => $this->topicName,
+                    'partitions' => array(
+                        array(
+                            'partition_id' => $this->partitionId,
+                            'time' => $timeLine,
+                            'max_offset' => 1,
+                        ),
+                    ),
+                ),
+            ),
+        );
+        $this->encoder->offsetRequest($requestData);
+        $result = $this->decoder->offsetResponse();
+        $this->client->freeStream($this->streamKey);
+
+        if (!isset($result[$topicName][$partitionId]['offset'])) {
+            if (isset($result[$topicName][$partitionId]['errCode'])) {
+                throw new \Kafka\Exception(\Kafka\Protocol\Decoder::getError($result[$topicName][$partitionId]['errCode']));
+            } else {
+                throw new \Kafka\Exception('get offset failed. topic name:' . $this->topicName . ' partitionId: ' . $this->partitionId);
+            }
+        }
+
+        return array_shift($result[$topicName][$partitionId]['offset']);
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Produce.php b/vendor/nmred/kafka-php/src/Kafka/Produce.php
new file mode 100644
index 00000000..2b9e6cd3
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Produce.php
@@ -0,0 +1,337 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Produce
+{
+    // {{{ consts
+    // }}}
+    // {{{ members
+
+    /**
+     * client
+     *
+     * @var mixed
+     * @access private
+     */
+    private $client = null;
+
+    /**
+     * send message options cache
+     *
+     * @var array
+     * @access private
+     */
+    private $payload = array();
+
+    /**
+     * default the server will not send any response
+     *
+     * @var float
+     * @access private
+     */
+    private $requiredAck = 0;
+
+    /**
+     * default timeout is 100ms
+     *
+     * @var float
+     * @access private
+     */
+    private $timeout = 100;
+
+    /**
+     * produce instance
+     *
+     * @var \Kafka\Produce
+     * @access private
+     */
+    private static $instance = null;
+
+    /**
+     * broker host list
+     *
+     * @var array
+     * @access private
+     */
+    private $hostList = array();
+
+    /**
+     * save broker connection
+     *
+     * @var array
+     * @access private
+     */
+    private $stream = array();
+
+    // }}}
+    // {{{ functions
+    // {{{ public function static getInstance()
+
+    /**
+     * set send messages
+     *
+     * @access public
+     * @return void
+     */
+    public static function getInstance($hostList, $timeout, $kafkaHostList = null)
+    {
+        if (is_null(self::$instance)) {
+            self::$instance = new self($hostList, $timeout, $kafkaHostList);
+        }
+
+        return self::$instance;
+    }
+
+    // }}}
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @access public
+     * @return void
+     */
+    public function __construct($hostList, $timeout = null, $kafkaHostList = null)
+    {
+        if ($hostList instanceof \Kafka\ClusterMetaData) {
+            $metadata = $hostList;
+        } elseif ( $kafkaHostList !== null ) {
+            $metadata = new \Kafka\MetaDataFromKafka($kafkaHostList);
+        } else {
+            $metadata = new \Kafka\ZooKeeper($hostList, $timeout);
+        }
+        $this->client = new \Kafka\Client($metadata);
+    }
+
+    // }}}
+    // {{{ public function setMessages()
+
+    /**
+     * set send messages
+     *
+     * @access public
+     * @return void
+     */
+    public function setMessages($topicName, $partitionId = 0, $messages = array())
+    {
+        if (isset($this->payload[$topicName][$partitionId])) {
+            $this->payload[$topicName][$partitionId] =
+                    array_merge($this->payload[$topicName][$partitionId], $messages);
+        } else {
+            $this->payload[$topicName][$partitionId] = $messages;
+        }
+
+        return $this;
+    }
+
+    // }}}
+    // {{{ public function setRequireAck()
+
+    /**
+     * set request mode
+     * This field indicates how many acknowledgements the servers should receive
+     * before responding to the request. If it is 0 the server will not send any
+     * response (this is the only case where the server will not reply to a
+     * request). If it is 1, the server will wait the data is written to the
+     * local log before sending a response. If it is -1 the server will block
+     * until the message is committed by all in sync replicas before sending a
+     * response. For any number > 1 the server will block waiting for this
+     * number of acknowledgements to occur (but the server will never wait for
+     * more acknowledgements than there are in-sync replicas).
+     *
+     * @param int $ack
+     * @access public
+     * @return void
+     */
+    public function setRequireAck($ack = 0)
+    {
+        if ($ack >= -1) {
+            $this->requiredAck = (int) $ack;
+        }
+
+        return $this;
+    }
+
+    // }}}
+    // {{{ public function setTimeOut()
+
+    /**
+     * set request timeout
+     *
+     * @param int $timeout
+     * @access public
+     * @return void
+     */
+    public function setTimeOut($timeout = 100)
+    {
+        if ((int) $timeout) {
+            $this->timeout = (int) $timeout;
+        }
+        return $this;
+    }
+
+    // }}}
+    // {{{ public function send()
+
+    /**
+     * send message to broker
+     *
+     * @access public
+     * @return void
+     */
+    public function send()
+    {
+        $data = $this->_formatPayload();
+        if (empty($data)) {
+            return false;
+        }
+
+        $responseData = array();
+        foreach ($data as $host => $requestData) {
+            $stream = $this->client->getStream($host);
+            $conn   = $stream['stream'];
+            $encoder = new \Kafka\Protocol\Encoder($conn);
+            $encoder->produceRequest($requestData);
+            if ((int) $this->requiredAck !== 0) { // get broker response
+                $decoder = new \Kafka\Protocol\Decoder($conn);
+                $response = $decoder->produceResponse();
+                foreach ($response as $topicName => $info) {
+                    if (!isset($responseData[$topicName])) {
+                        $responseData[$topicName] = $info;
+                    } else {
+                        $responseData[$topicName] = array_merge($info, $responseData[$topicName]);
+                    }
+                }
+            }
+
+            $this->client->freeStream($stream['key']);
+        }
+
+        $this->payload = array();
+        return $responseData;
+    }
+
+    // }}}
+    // {{{ public function getClient()
+
+    /**
+     * get client object
+     *
+     * @access public
+     * @return void
+     */
+    public function getClient()
+    {
+        return $this->client;
+    }
+
+    /**
+     * passthru method to client for setting stream options
+     *
+     * @access public
+     * @param array $options
+     */
+    public function setStreamOptions($options = array())
+    {
+        $this->client->setStreamOptions($options);
+    }
+
+    // }}}
+    // {{{ public function getAvailablePartitions()
+
+    /**
+     * get available partition
+     *
+     * @access public
+     * @return array
+     */
+    public function getAvailablePartitions($topicName)
+    {
+        $topicDetail = $this->client->getTopicDetail($topicName);
+        if (is_array($topicDetail) && isset($topicDetail['partitions'])) {
+            $topicPartitiions = array_keys($topicDetail['partitions']);
+        } else {
+            $topicPartitiions = array();
+        }
+
+        return $topicPartitiions;
+    }
+
+    // }}}
+    // {{{ private function _formatPayload()
+
+    /**
+     * format payload array
+     *
+     * @access private
+     * @return array
+     */
+    private function _formatPayload()
+    {
+        if (empty($this->payload)) {
+            return array();
+        }
+
+        $data = array();
+        foreach ($this->payload as $topicName => $partitions) {
+            foreach ($partitions as $partitionId => $messages) {
+                $host = $this->client->getHostByPartition($topicName, $partitionId);
+                $data[$host][$topicName][$partitionId] = $messages;
+            }
+        }
+
+        $requestData = array();
+        foreach ($data as $host => $info) {
+            $topicData = array();
+            foreach ($info as $topicName => $partitions) {
+                $partitionData = array();
+                foreach ($partitions as $partitionId => $messages) {
+                    $partitionData[] = array(
+                        'partition_id' => $partitionId,
+                        'messages'     => $messages,
+                    );
+                }
+                $topicData[] = array(
+                    'topic_name' => $topicName,
+                    'partitions' => $partitionData,
+                );
+            }
+
+            $requestData[$host] = array(
+                'required_ack' => $this->requiredAck,
+                'timeout'      => $this->timeout,
+                'data' => $topicData,
+            );
+        }
+
+       return $requestData;
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Decoder.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Decoder.php
new file mode 100644
index 00000000..f1e4b496
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Decoder.php
@@ -0,0 +1,430 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Decoder extends Protocol
+{
+    // {{{ functions
+    // {{{ public function produceResponse()
+
+    /**
+     * decode produce response
+     *
+     * @param string $data
+     * @access public
+     * @return array
+     */
+    public function produceResponse()
+    {
+        $result = array();
+        $dataLen = self::unpack(self::BIT_B32, $this->stream->read(4, true));
+        $dataLen = array_shift($dataLen);
+        if (!$dataLen) {
+            throw new \Kafka\Exception\Protocol('produce response invalid.');
+        }
+        $data = $this->stream->read($dataLen, true);
+
+        // parse data struct
+        $offset = 4;
+        $topicCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+        $topicCount = array_shift($topicCount);
+        $offset += 4;
+        for ($i = 0; $i < $topicCount; $i++) {
+            $topicLen = self::unpack(self::BIT_B16, substr($data, $offset, 2)); // int16 topic name length
+            $topicLen = isset($topicLen[1]) ? $topicLen[1] : 0;
+            $offset += 2;
+            $topicName = substr($data, $offset, $topicLen);
+            $offset += $topicLen;
+            $partitionCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+            $partitionCount = isset($partitionCount[1]) ? $partitionCount[1] : 0;
+            $offset += 4;
+            $result[$topicName] = array();
+            for ($j = 0; $j < $partitionCount; $j++) {
+                $partitionId = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                $offset += 4;
+                $errCode = self::unpack(self::BIT_B16, substr($data, $offset, 2));
+                $offset += 2;
+                $partitionOffset = self::unpack(self::BIT_B64, substr($data, $offset, 8));
+                $offset += 8;
+                $result[$topicName][$partitionId[1]] = array(
+                    'errCode' => $errCode[1],
+                    'offset'  => $partitionOffset
+                );
+            }
+        }
+
+        return $result;
+    }
+
+    // }}}
+    // {{{ public function fetchResponse()
+
+    /**
+     * decode fetch response
+     *
+     * @param string $data
+     * @access public
+     * @return Iterator
+     */
+    public function fetchResponse()
+    {
+        return new \Kafka\Protocol\Fetch\Topic($this->stream);
+    }
+
+    // }}}
+    // {{{ public function metadataResponse()
+
+    /**
+     * decode metadata response
+     *
+     * @param string $data
+     * @access public
+     * @return array
+     */
+    public function metadataResponse()
+    {
+        $result = array();
+        $broker = array();
+        $topic = array();
+        $dataLen = self::unpack(self::BIT_B32, $this->stream->read(4, true));
+        $dataLen = array_shift($dataLen);
+        if (!$dataLen) {
+            throw new \Kafka\Exception\Protocol('metaData response invalid.');
+        }
+        $data = $this->stream->read($dataLen, true);
+        $offset = 4;
+        $brokerCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+        $offset += 4;
+        $brokerCount = isset($brokerCount[1]) ? $brokerCount[1] : 0;
+        for ($i = 0; $i < $brokerCount; $i++) {
+            $nodeId = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+            $nodeId = $nodeId[1];
+            $offset += 4;
+            $hostNameLen = self::unpack(self::BIT_B16, substr($data, $offset, 2)); // int16 host name length
+            $hostNameLen = isset($hostNameLen[1]) ? $hostNameLen[1] : 0;
+            $offset += 2;
+            $hostName = substr($data, $offset, $hostNameLen);
+            $offset += $hostNameLen;
+            $port = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+            $offset += 4;
+            $broker[$nodeId] = array(
+                'host' => $hostName,
+                'port' => $port[1],
+            );
+        }
+
+        $topicMetaCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+        $offset += 4;
+        $topicMetaCount = isset($topicMetaCount[1]) ? $topicMetaCount[1] : 0;
+        for ($i = 0; $i < $topicMetaCount; $i++) {
+            $topicErrCode = self::unpack(self::BIT_B16, substr($data, $offset, 2));
+            $offset += 2;
+            $topicLen = self::unpack(self::BIT_B16, substr($data, $offset, 2));
+            $offset += 2;
+            $topicName = substr($data, $offset, $topicLen[1]);
+            $offset += $topicLen[1];
+            $partitionCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+            $offset += 4;
+            $partitionCount = isset($partitionCount[1]) ? $partitionCount[1] : 0;
+            $topic[$topicName]['errCode'] = $topicErrCode[1];
+            $partitions = array();
+            for ($j = 0; $j < $partitionCount; $j++) {
+                $partitionErrCode = self::unpack(self::BIT_B16, substr($data, $offset, 2));
+                $offset += 2;
+                $partitionId = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                $partitionId = isset($partitionId[1]) ? $partitionId[1] : 0;
+                $offset += 4;
+                $leaderId = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                $offset += 4;
+                $repliasCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                $offset += 4;
+                $repliasCount = isset($repliasCount[1]) ? $repliasCount[1] : 0;
+                $replias = array();
+                for ($z = 0; $z < $repliasCount; $z++) {
+                    $repliaId = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                    $offset += 4;
+                    $replias[] = $repliaId[1];
+                }
+                $isrCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                $offset += 4;
+                $isrCount = isset($isrCount[1]) ? $isrCount[1] : 0;
+                $isrs = array();
+                for ($z = 0; $z < $isrCount; $z++) {
+                    $isrId = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                    $offset += 4;
+                    $isrs[] = $isrId[1];
+                }
+
+                $partitions[$partitionId] = array(
+                    'errCode'  => $partitionErrCode[1],
+                    'leader'   => $leaderId[1],
+                    'replicas' => $replias,
+                    'isr'      => $isrs,
+                );
+            }
+            $topic[$topicName]['partitions'] = $partitions;
+        }
+
+        $result = array(
+            'brokers' => $broker,
+            'topics'  => $topic,
+        );
+        return $result;
+    }
+
+    // }}}
+    // {{{ public function offsetResponse()
+
+    /**
+     * decode offset response
+     *
+     * @param string $data
+     * @access public
+     * @return array
+     */
+    public function offsetResponse()
+    {
+        $result = array();
+        $dataLen = self::unpack(self::BIT_B32, $this->stream->read(4, true));
+        $dataLen = array_shift($dataLen);
+        if (!$dataLen) {
+            throw new \Kafka\Exception\Protocol('offset response invalid.');
+        }
+        $data = $this->stream->read($dataLen, true);
+        $offset = 4;
+        $topicCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+        $offset += 4;
+        $topicCount = array_shift($topicCount);
+        for ($i = 0; $i < $topicCount; $i++) {
+            $topicLen = self::unpack(self::BIT_B16, substr($data, $offset, 2)); // int16 topic name length
+            $topicLen = isset($topicLen[1]) ? $topicLen[1] : 0;
+            $offset += 2;
+            $topicName = substr($data, $offset, $topicLen);
+            $offset += $topicLen;
+            $partitionCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+            $partitionCount = isset($partitionCount[1]) ? $partitionCount[1] : 0;
+            $offset += 4;
+            $result[$topicName] = array();
+            for ($j = 0; $j < $partitionCount; $j++) {
+                $partitionId = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                $offset += 4;
+                $errCode     = self::unpack(self::BIT_B16, substr($data, $offset, 2));
+                $offset += 2;
+                $offsetCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                $offset += 4;
+                $offsetCount = array_shift($offsetCount);
+                $offsetArr = array();
+                for ($z = 0; $z < $offsetCount; $z++) {
+                    $offsetArr[] = self::unpack(self::BIT_B64, substr($data, $offset, 8));
+                    $offset += 8;
+                }
+                $result[$topicName][$partitionId[1]] = array(
+                    'errCode' => $errCode[1],
+                    'offset'  => $offsetArr
+                );
+            }
+        }
+        return $result;
+    }
+
+    // }}}
+    // {{{ public function commitOffsetResponse()
+
+    /**
+     * decode commit offset response
+     *
+     * @param string $data
+     * @access public
+     * @return array
+     */
+    public function commitOffsetResponse()
+    {
+        $result = array();
+        $dataLen = self::unpack(self::BIT_B32, $this->stream->read(4, true));
+        $dataLen = array_shift($dataLen);
+        if (!$dataLen) {
+            throw new \Kafka\Exception\Protocol('commit offset response invalid.');
+        }
+        $data = $this->stream->read($dataLen, true);
+        $offset = 4;
+        $topicCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+        $offset += 4;
+        $topicCount = array_shift($topicCount);
+        for ($i = 0; $i < $topicCount; $i++) {
+            $topicLen = self::unpack(self::BIT_B16, substr($data, $offset, 2)); // int16 topic name length
+            $topicLen = isset($topicLen[1]) ? $topicLen[1] : 0;
+            $offset += 2;
+            $topicName = substr($data, $offset, $topicLen);
+            $offset += $topicLen;
+            $partitionCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+            $partitionCount = isset($partitionCount[1]) ? $partitionCount[1] : 0;
+            $offset += 4;
+            $result[$topicName] = array();
+            for ($j = 0; $j < $partitionCount; $j++) {
+                $partitionId = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                $offset += 4;
+                $errCode     = self::unpack(self::BIT_B16, substr($data, $offset, 2));
+                $offset += 2;
+                $result[$topicName][$partitionId[1]] = array(
+                    'errCode' => $errCode[1],
+                );
+            }
+        }
+        return $result;
+    }
+
+    // }}}
+    // {{{ public function fetchOffsetResponse()
+
+    /**
+     * decode fetch offset response
+     *
+     * @param string $data
+     * @access public
+     * @return array
+     */
+    public function fetchOffsetResponse()
+    {
+        $result = array();
+        $dataLen = self::unpack(self::BIT_B32, $this->stream->read(4, true));
+        $dataLen = array_shift($dataLen);
+        if (!$dataLen) {
+            throw new \Kafka\Exception\Protocol('fetch offset response invalid.');
+        }
+        $data = $this->stream->read($dataLen, true);
+        $offset = 4;
+        $topicCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+        $offset += 4;
+        $topicCount = array_shift($topicCount);
+        for ($i = 0; $i < $topicCount; $i++) {
+            $topicLen = self::unpack(self::BIT_B16, substr($data, $offset, 2)); // int16 topic name length
+            $topicLen = isset($topicLen[1]) ? $topicLen[1] : 0;
+            $offset += 2;
+            $topicName = substr($data, $offset, $topicLen);
+            $offset += $topicLen;
+            $partitionCount = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+            $partitionCount = isset($partitionCount[1]) ? $partitionCount[1] : 0;
+            $offset += 4;
+            $result[$topicName] = array();
+            for ($j = 0; $j < $partitionCount; $j++) {
+                $partitionId = self::unpack(self::BIT_B32, substr($data, $offset, 4));
+                $offset += 4;
+                $partitionOffset = self::unpack(self::BIT_B64, substr($data, $offset, 8));
+                $offset += 8;
+                $metaLen = self::unpack(self::BIT_B16, substr($data, $offset, 2));
+                $metaLen = array_shift($metaLen);
+                $offset += 2;
+                $metaData = '';
+                if ($metaLen) {
+                    $metaData = substr($data, $offset, $metaLen);
+                    $offset += $metaLen;
+                }
+                $errCode = self::unpack(self::BIT_B16, substr($data, $offset, 2));
+                $offset += 2;
+                $result[$topicName][$partitionId[1]] = array(
+                    'offset'   => $partitionOffset,
+                    'metadata' => $metaData,
+                    'errCode'  => $errCode[1],
+                );
+            }
+        }
+        return $result;
+    }
+
+    // }}}
+    // {{{ public static function getError()
+
+    /**
+     * get error
+     *
+     * @param integer $errCode
+     * @static
+     * @access public
+     * @return string
+     */
+    public static function getError($errCode)
+    {
+        $error = '';
+        switch($errCode) {
+            case 0:
+                $error = 'No error--it worked!';
+                break;
+            case -1:
+                $error = 'An unexpected server error';
+                break;
+            case 1:
+                $error = 'The requested offset is outside the range of offsets maintained by the server for the given topic/partition.';
+                break;
+            case 2:
+                $error = 'This indicates that a message contents does not match its CRC';
+                break;
+            case 3:
+                $error = 'This request is for a topic or partition that does not exist on this broker.';
+                break;
+            case 4:
+                $error = 'The message has a negative size';
+                break;
+            case 5:
+                $error = 'This error is thrown if we are in the middle of a leadership election and there is currently no leader for this partition and hence it is unavailable for writes';
+                break;
+            case 6:
+                $error = 'This error is thrown if the client attempts to send messages to a replica that is not the leader for some partition. It indicates that the clients metadata is out of date.';
+                break;
+            case 7:
+                $error = 'This error is thrown if the request exceeds the user-specified time limit in the request.';
+                break;
+            case 8:
+                $error = 'This is not a client facing error and is used only internally by intra-cluster broker communication.';
+                break;
+            case 10:
+                $error = 'The server has a configurable maximum message size to avoid unbounded memory allocation. This error is thrown if the client attempt to produce a message larger than this maximum.';
+                break;
+            case 11:
+                $error = 'Internal error code for broker-to-broker communication.';
+                break;
+            case 12:
+                $error = 'If you specify a string larger than configured maximum for offset metadata';
+                break;
+            case 14:
+                $error = 'The broker returns this error code for an offset fetch request if it is still loading offsets (after a leader change for that offsets topic partition).';
+                break;
+            case 15:
+                $error = 'The broker returns this error code for consumer metadata requests or offset commit requests if the offsets topic has not yet been created.';
+                break;
+            case 16:
+                $error = 'The broker returns this error code if it receives an offset fetch or commit request for a consumer group that it is not a coordinator for.';
+                break;
+            default:
+                $error = 'Unknown error';
+        }
+
+        return $error;
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Encoder.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Encoder.php
new file mode 100644
index 00000000..7d36e10f
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Encoder.php
@@ -0,0 +1,652 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Encoder extends Protocol
+{
+    // {{{ functions
+    // {{{ public function produceRequest()
+
+    /**
+     * produce request
+     *
+     * @param array $payloads
+     * @static
+     * @access public
+     * @return void
+     */
+    public function produceRequest($payloads, $compression = self::COMPRESSION_NONE)
+    {
+        if (!isset($payloads['data'])) {
+            throw new \Kafka\Exception\Protocol('given procude data invalid. `data` is undefined.');
+        }
+
+        if (!isset($payloads['required_ack'])) {
+            // default server will not send any response
+            // (this is the only case where the server will not reply to a request)
+            $payloads['required_ack'] = 0;
+        }
+
+        if (!isset($payloads['timeout'])) {
+            $payloads['timeout'] = 100; // default timeout 100ms
+        }
+
+        $header = self::requestHeader('kafka-php', 0, self::PRODUCE_REQUEST);
+        $data   = self::pack(self::BIT_B16, $payloads['required_ack']);
+        $data  .= self::pack(self::BIT_B32, $payloads['timeout']);
+        $data  .= self::encodeArray($payloads['data'], array(__CLASS__, '_encodeProcudeTopic'), $compression);
+        $data   = self::encodeString($header . $data, self::PACK_INT32);
+
+        return $this->stream->write($data);
+    }
+
+    // }}}
+    // {{{ public function metadataRequest()
+
+    /**
+     * build metadata request protocol
+     *
+     * @param array $topics
+     * @access public
+     * @return string
+     */
+    public function metadataRequest($topics)
+    {
+        if (!is_array($topics)) {
+            $topics = array($topics);
+        }
+
+        foreach ($topics as $topic) {
+            if (!is_string($topic)) {
+                throw new \Kafka\Exception\Protocol('request metadata topic array have invalid value. ');
+            }
+        }
+
+        $header = self::requestHeader('kafka-php', 0, self::METADATA_REQUEST);
+        $data   = self::encodeArray($topics, array(__CLASS__, 'encodeString'), self::PACK_INT16);
+        $data   = self::encodeString($header . $data, self::PACK_INT32);
+
+        return $this->stream->write($data);
+    }
+
+    // }}}
+    // {{{ public function fetchRequest()
+
+    /**
+     * build fetch request
+     *
+     * @param array $payloads
+     * @access public
+     * @return string
+     */
+    public function fetchRequest($payloads)
+    {
+        if (!isset($payloads['data'])) {
+            throw new \Kafka\Exception\Protocol('given fetch kafka data invalid. `data` is undefined.');
+        }
+
+        if (!isset($payloads['replica_id'])) {
+            $payloads['replica_id'] = -1;
+        }
+
+        if (!isset($payloads['max_wait_time'])) {
+            $payloads['max_wait_time'] = 100; // default timeout 100ms
+        }
+
+        if (!isset($payloads['min_bytes'])) {
+            $payloads['min_bytes'] = 64 * 1024; // 64k
+        }
+
+        $header = self::requestHeader('kafka-php', 0, self::FETCH_REQUEST);
+        $data   = self::pack(self::BIT_B32, $payloads['replica_id']);
+        $data  .= self::pack(self::BIT_B32, $payloads['max_wait_time']);
+        $data  .= self::pack(self::BIT_B32, $payloads['min_bytes']);
+        $data  .= self::encodeArray($payloads['data'], array(__CLASS__, '_encodeFetchTopic'));
+        $data   = self::encodeString($header . $data, self::PACK_INT32);
+
+        return $this->stream->write($data);
+    }
+
+    // }}}
+    // {{{ public function offsetRequest()
+
+    /**
+     * build offset request
+     *
+     * @param array $payloads
+     * @access public
+     * @return string
+     */
+    public function offsetRequest($payloads)
+    {
+        if (!isset($payloads['data'])) {
+            throw new \Kafka\Exception\Protocol('given offset data invalid. `data` is undefined.');
+        }
+
+        if (!isset($payloads['replica_id'])) {
+            $payloads['replica_id'] = -1;
+        }
+
+        $header = self::requestHeader('kafka-php', 0, self::OFFSET_REQUEST);
+        $data   = self::pack(self::BIT_B32, $payloads['replica_id']);
+        $data  .= self::encodeArray($payloads['data'], array(__CLASS__, '_encodeOffsetTopic'));
+        $data   = self::encodeString($header . $data, self::PACK_INT32);
+
+        return $this->stream->write($data);
+    }
+
+    // }}}
+    // {{{ public function commitOffsetRequest()
+
+    /**
+     * build consumer commit offset request
+     *
+     * @param array $payloads
+     * @access public
+     * @return string
+     */
+    public function commitOffsetRequest($payloads)
+    {
+        if (!isset($payloads['data'])) {
+            throw new \Kafka\Exception\Protocol('given commit offset data invalid. `data` is undefined.');
+        }
+
+        if (!isset($payloads['group_id'])) {
+            throw new \Kafka\Exception\Protocol('given commit offset data invalid. `group_id` is undefined.');
+        }
+
+        $header = self::requestHeader('kafka-php', 0, self::OFFSET_COMMIT_REQUEST);
+        $data   = self::encodeString($payloads['group_id'], self::PACK_INT16);
+        $data  .= self::encodeArray($payloads['data'], array(__CLASS__, '_encodeCommitOffset'));
+        $data   = self::encodeString($header . $data, self::PACK_INT32);
+
+        return $this->stream->write($data);
+    }
+
+    // }}}
+    // {{{ public function fetchOffsetRequest()
+
+    /**
+     * build consumer fetch offset request
+     *
+     * @param array $payloads
+     * @access public
+     * @return string
+     */
+    public function fetchOffsetRequest($payloads)
+    {
+        if (!isset($payloads['data'])) {
+            throw new \Kafka\Exception\Protocol('given fetch offset data invalid. `data` is undefined.');
+        }
+
+        if (!isset($payloads['group_id'])) {
+            throw new \Kafka\Exception\Protocol('given fetch offset data invalid. `group_id` is undefined.');
+        }
+
+        $header = self::requestHeader('kafka-php', 0, self::OFFSET_FETCH_REQUEST);
+        $data   = self::encodeString($payloads['group_id'], self::PACK_INT16);
+        $data  .= self::encodeArray($payloads['data'], array(__CLASS__, '_encodeFetchOffset'));
+        $data   = self::encodeString($header . $data, self::PACK_INT32);
+
+        return $this->stream->write($data);
+    }
+
+    // }}}
+    // {{{ public static function encodeString()
+
+    /**
+     * encode pack string type
+     *
+     * @param string $string
+     * @param int $bytes   self::PACK_INT32: int32 big endian order. self::PACK_INT16: int16 big endian order.
+     * @static
+     * @access public
+     * @return string
+     */
+    public static function encodeString($string, $bytes, $compression = self::COMPRESSION_NONE)
+    {
+        $packLen = ($bytes == self::PACK_INT32) ? self::BIT_B32 : self::BIT_B16;
+        switch ($compression) {
+            case self::COMPRESSION_NONE:
+                break;
+            case self::COMPRESSION_GZIP:
+                $string = gzencode($string);
+                break;
+            case self::COMPRESSION_SNAPPY:
+                throw new \Kafka\Exception\NotSupported('SNAPPY compression not yet implemented');
+            default:
+                throw new \Kafka\Exception\NotSupported('Unknown compression flag: ' . $compression);
+        }
+        return self::pack($packLen, strlen($string)) . $string;
+    }
+
+    // }}}
+    // {{{ public static function encodeArray()
+
+    /**
+     * encode key array
+     *
+     * @param array $array
+     * @param Callable $func
+     * @static
+     * @access public
+     * @return string
+     */
+    public static function encodeArray(array $array, $func, $options = null)
+    {
+        if (!is_callable($func, false)) {
+            throw new \Kafka\Exception\Protocol('Encode array failed, given function is not callable.');
+        }
+
+        $arrayCount = count($array);
+
+        $body = '';
+        foreach ($array as $value) {
+            if (!is_null($options)) {
+                $body .= call_user_func($func, $value, $options);
+            } else {
+                $body .= call_user_func($func, $value);
+            }
+        }
+
+        return self::pack(self::BIT_B32, $arrayCount) . $body;
+    }
+
+    // }}}
+    // {{{ public static function encodeMessageSet()
+
+    /**
+     * encode message set
+     * N.B., MessageSets are not preceded by an int32 like other array elements
+     * in the protocol.
+     *
+     * @param array $messages
+     * @static
+     * @access public
+     * @return string
+     */
+    public static function encodeMessageSet($messages, $compression = self::COMPRESSION_NONE)
+    {
+        if (!is_array($messages)) {
+            $messages = array($messages);
+        }
+
+        $data = '';
+        foreach ($messages as $message) {
+            $tmpMessage = self::_encodeMessage($message, $compression);
+
+            // int64 -- message offset     Message
+            $data .= self::pack(self::BIT_B64, 0) . self::encodeString($tmpMessage, self::PACK_INT32);
+        }
+        return $data;
+    }
+
+    // }}}
+    // {{{ public static function requestHeader()
+
+    /**
+     * get request header
+     *
+     * @param string $clientId
+     * @param integer $correlationId
+     * @param integer $apiKey
+     * @static
+     * @access public
+     * @return void
+     */
+    public static function requestHeader($clientId, $correlationId, $apiKey)
+    {
+        // int16 -- apiKey int16 -- apiVersion int32 correlationId
+        $binData  = self::pack(self::BIT_B16, $apiKey);
+        $binData .= self::pack(self::BIT_B16, self::API_VERSION);
+        $binData .= self::pack(self::BIT_B32, $correlationId);
+
+        // concat client id
+        $binData .= self::encodeString($clientId, self::PACK_INT16);
+
+        return $binData;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeMessage()
+
+    /**
+     * encode signal message
+     *
+     * @param string $message
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeMessage($message, $compression = self::COMPRESSION_NONE)
+    {
+        // int8 -- magic  int8 -- attribute
+        $data  = self::pack(self::BIT_B8, self::MESSAGE_MAGIC);
+        $data .= self::pack(self::BIT_B8, $compression);
+
+        // message key
+        $data .= self::encodeString('', self::PACK_INT32);
+
+        // message value
+        $data .= self::encodeString($message, self::PACK_INT32, $compression);
+
+        $crc = crc32($data);
+
+        // int32 -- crc code  string data
+        $message = self::pack(self::BIT_B32, $crc) . $data;
+
+        return $message;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeProcudePartion()
+
+    /**
+     * encode signal part
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeProcudePartion($values, $compression)
+    {
+        if (!isset($values['partition_id'])) {
+            throw new \Kafka\Exception\Protocol('given produce data invalid. `partition_id` is undefined.');
+        }
+
+        if (!isset($values['messages']) || empty($values['messages'])) {
+            throw new \Kafka\Exception\Protocol('given produce data invalid. `messages` is undefined.');
+        }
+
+        $data = self::pack(self::BIT_B32, $values['partition_id']);
+        $data .= self::encodeString(self::encodeMessageSet($values['messages'], $compression), self::PACK_INT32);
+
+        return $data;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeProcudeTopic()
+
+    /**
+     * encode signal topic
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeProcudeTopic($values, $compression)
+    {
+        if (!isset($values['topic_name'])) {
+            throw new \Kafka\Exception\Protocol('given produce data invalid. `topic_name` is undefined.');
+        }
+
+        if (!isset($values['partitions']) || empty($values['partitions'])) {
+            throw new \Kafka\Exception\Protocol('given produce data invalid. `partitions` is undefined.');
+        }
+
+        $topic = self::encodeString($values['topic_name'], self::PACK_INT16);
+        $partitions = self::encodeArray($values['partitions'], array(__CLASS__, '_encodeProcudePartion'), $compression);
+
+        return $topic . $partitions;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeFetchPartion()
+
+    /**
+     * encode signal part
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeFetchPartion($values)
+    {
+        if (!isset($values['partition_id'])) {
+            throw new \Kafka\Exception\Protocol('given fetch data invalid. `partition_id` is undefined.');
+        }
+
+        if (!isset($values['offset'])) {
+            $values['offset'] = 0;
+        }
+
+        if (!isset($values['max_bytes'])) {
+            $values['max_bytes'] = 100 * 1024 * 1024;
+        }
+
+        $data = self::pack(self::BIT_B32, $values['partition_id']);
+        $data .= self::pack(self::BIT_B64, $values['offset']);
+        $data .= self::pack(self::BIT_B32, $values['max_bytes']);
+
+        return $data;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeFetchTopic()
+
+    /**
+     * encode signal topic
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeFetchTopic($values)
+    {
+        if (!isset($values['topic_name'])) {
+            throw new \Kafka\Exception\Protocol('given fetch data invalid. `topic_name` is undefined.');
+        }
+
+        if (!isset($values['partitions']) || empty($values['partitions'])) {
+            throw new \Kafka\Exception\Protocol('given fetch data invalid. `partitions` is undefined.');
+        }
+
+        $topic = self::encodeString($values['topic_name'], self::PACK_INT16);
+        $partitions = self::encodeArray($values['partitions'], array(__CLASS__, '_encodeFetchPartion'));
+
+        return $topic . $partitions;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeOffsetPartion()
+
+    /**
+     * encode signal part
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeOffsetPartion($values)
+    {
+        if (!isset($values['partition_id'])) {
+            throw new \Kafka\Exception\Protocol('given offset data invalid. `partition_id` is undefined.');
+        }
+
+        if (!isset($values['time'])) {
+            $values['time'] = -1; // -1
+        }
+
+        if (!isset($values['max_offset'])) {
+            $values['max_offset'] = 100000;
+        }
+
+        $data = self::pack(self::BIT_B32, $values['partition_id']);
+        $data .= self::pack(self::BIT_B64, $values['time']);
+        $data .= self::pack(self::BIT_B32, $values['max_offset']);
+
+        return $data;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeOffsetTopic()
+
+    /**
+     * encode signal topic
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeOffsetTopic($values)
+    {
+        if (!isset($values['topic_name'])) {
+            throw new \Kafka\Exception\Protocol('given offset data invalid. `topic_name` is undefined.');
+        }
+
+        if (!isset($values['partitions']) || empty($values['partitions'])) {
+            throw new \Kafka\Exception\Protocol('given offset data invalid. `partitions` is undefined.');
+        }
+
+        $topic = self::encodeString($values['topic_name'], self::PACK_INT16);
+        $partitions = self::encodeArray($values['partitions'], array(__CLASS__, '_encodeOffsetPartion'));
+
+        return $topic . $partitions;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeCommitOffsetPartion()
+
+    /**
+     * encode signal part
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeCommitOffsetPartion($values)
+    {
+        if (!isset($values['partition_id'])) {
+            throw new \Kafka\Exception\Protocol('given commit offset data invalid. `partition_id` is undefined.');
+        }
+
+        if (!isset($values['offset'])) {
+            throw new \Kafka\Exception\Protocol('given commit offset data invalid. `offset` is undefined.');
+        }
+
+        if (!isset($values['time'])) {
+            $values['time'] = -1;
+        }
+
+        if (!isset($values['metadata'])) {
+            $values['metadata'] = 'm';
+        }
+
+        $data = self::pack(self::BIT_B32, $values['partition_id']);
+        $data .= self::pack(self::BIT_B64, $values['offset']);
+        $data .= self::pack(self::BIT_B64, $values['time']);
+        $data .= self::encodeString($values['metadata'], self::PACK_INT16);
+
+        return $data;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeCommitOffset()
+
+    /**
+     * encode signal topic
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeCommitOffset($values)
+    {
+        if (!isset($values['topic_name'])) {
+            throw new \Kafka\Exception\Protocol('given commit offset data invalid. `topic_name` is undefined.');
+        }
+
+        if (!isset($values['partitions']) || empty($values['partitions'])) {
+            throw new \Kafka\Exception\Protocol('given commit offset data invalid. `partitions` is undefined.');
+        }
+
+        $topic = self::encodeString($values['topic_name'], self::PACK_INT16);
+        $partitions = self::encodeArray($values['partitions'], array(__CLASS__, '_encodeCommitOffsetPartion'));
+
+        return $topic . $partitions;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeFetchOffsetPartion()
+
+    /**
+     * encode signal part
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeFetchOffsetPartion($values)
+    {
+        if (!isset($values['partition_id'])) {
+            throw new \Kafka\Exception\Protocol('given fetch offset data invalid. `partition_id` is undefined.');
+        }
+
+        $data = self::pack(self::BIT_B32, $values['partition_id']);
+
+        return $data;
+    }
+
+    // }}}
+    // {{{ protected static function _encodeFetchOffset()
+
+    /**
+     * encode signal topic
+     *
+     * @param partions
+     * @static
+     * @access protected
+     * @return string
+     */
+    protected static function _encodeFetchOffset($values)
+    {
+        if (!isset($values['topic_name'])) {
+            throw new \Kafka\Exception\Protocol('given fetch offset data invalid. `topic_name` is undefined.');
+        }
+
+        if (!isset($values['partitions']) || empty($values['partitions'])) {
+            throw new \Kafka\Exception\Protocol('given fetch offset data invalid. `partitions` is undefined.');
+        }
+
+        $topic = self::encodeString($values['topic_name'], self::PACK_INT16);
+        $partitions = self::encodeArray($values['partitions'], array(__CLASS__, '_encodeFetchOffsetPartion'));
+
+        return $topic . $partitions;
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/CommitOffset.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/CommitOffset.php
new file mode 100644
index 00000000..8424cf78
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/CommitOffset.php
@@ -0,0 +1,119 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol\Fetch\Helper;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class CommitOffset extends HelperAbstract
+{
+    // {{{ members
+
+    /**
+     * consumer group
+     *
+     * @var string
+     * @access protected
+     */
+    protected $group = '';
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @access public
+     * @return void
+     */
+    public function __construct($client)
+    {
+        $this->client = $client;
+    }
+
+    // }}}
+    // {{{ public function setGroup()
+
+    /**
+     * set consumer group
+     *
+     * @access public
+     * @return void
+     */
+    public function setGroup($group)
+    {
+        $this->group = $group;
+    }
+
+    // }}}
+    // {{{ public function onStreamEof()
+
+    /**
+     * on stream eof call
+     *
+     * @param string $streamKey
+     * @access public
+     * @return void
+     */
+    public function onStreamEof($streamKey)
+    {
+    }
+
+    // }}}
+    // {{{ public function onTopicEof()
+
+    /**
+     * on topic eof call
+     *
+     * @param string $topicName
+     * @access public
+     * @return void
+     */
+    public function onTopicEof($topicName)
+    {
+    }
+
+    // }}}
+    // {{{ public function onPartitionEof()
+
+    /**
+     * on partition eof call
+     *
+     * @param \Kafka\Protocol\Fetch\Partition $partition
+     * @access public
+     * @return void
+     */
+    public function onPartitionEof($partition)
+    {
+        $partitionId = $partition->key();
+        $topicName = $partition->getTopicName();
+        $offset    = $partition->getMessageOffset();
+        $offsetObject = new  \Kafka\Offset($this->client, $this->group, $topicName, $partitionId);
+        $offsetObject->setOffset($offset);
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/Consumer.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/Consumer.php
new file mode 100644
index 00000000..acf0223e
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/Consumer.php
@@ -0,0 +1,39 @@
+<?php
+namespace Kafka\Protocol\Fetch\Helper;
+/**
+ * Description of Consumer
+ *
+ * @author daniel
+ */
+class Consumer extends HelperAbstract
+{
+    protected $consumer;
+
+    protected $offsetStrategy;
+
+
+    public function __construct(\Kafka\Consumer $consumer)
+    {
+        $this->consumer = $consumer;
+    }
+
+
+    public function onPartitionEof($partition)
+    {
+        $partitionId = $partition->key();
+        $topicName = $partition->getTopicName();
+        $offset    = $partition->getMessageOffset();
+        $this->consumer->setFromOffset(true);
+        $this->consumer->setPartition($topicName, $partitionId, ($offset +1));
+    }
+
+    public function onStreamEof($streamKey)
+    {
+
+    }
+
+    public function onTopicEof($topicName)
+    {
+
+    }
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/FreeStream.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/FreeStream.php
new file mode 100644
index 00000000..bba38dd6
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/FreeStream.php
@@ -0,0 +1,117 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol\Fetch\Helper;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class FreeStream extends HelperAbstract
+{
+    // {{{ members
+
+    /**
+     * streams
+     *
+     * @var array
+     * @access protected
+     */
+    protected $streams = array();
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @access public
+     * @return void
+     */
+    public function __construct($client)
+    {
+        $this->client = $client;
+    }
+
+    // }}}
+    // {{{ public function setStreams()
+
+    /**
+     * set streams
+     *
+     * @access public
+     * @return void
+     */
+    public function setStreams($streams)
+    {
+        $this->streams = $streams;
+    }
+
+    // }}}
+    // {{{ public function onStreamEof()
+
+    /**
+     * on stream eof call
+     *
+     * @param string $streamKey
+     * @access public
+     * @return void
+     */
+    public function onStreamEof($streamKey)
+    {
+        if (isset($this->streams[$streamKey])) {
+            $this->client->freeStream($streamKey);
+        }
+    }
+
+    // }}}
+    // {{{ public function onTopicEof()
+
+    /**
+     * on topic eof call
+     *
+     * @param string $topicName
+     * @access public
+     * @return void
+     */
+    public function onTopicEof($topicName)
+    {
+    }
+
+    // }}}
+    // {{{ public function onPartitionEof()
+
+    /**
+     * on partition eof call
+     *
+     * @param \Kafka\Protocol\Fetch\Partition $partition
+     * @access public
+     * @return void
+     */
+    public function onPartitionEof($partition)
+    {
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/Helper.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/Helper.php
new file mode 100644
index 00000000..4ec23926
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/Helper.php
@@ -0,0 +1,160 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol\Fetch\Helper;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Helper
+{
+    // {{{ members
+
+    /**
+     * helper object
+     */
+    private static $helpers = array();
+
+    // }}}
+    // {{{ functions
+    // {{{ public staitc function registerHelper()
+
+    /**
+     * register helper
+     *
+     * @param string $key
+     * @param \Kafka\Protocol\Fetch\Helper\HelperAbstract $helper
+     * @static
+     * @access public
+     * @return void
+     */
+    public static function registerHelper($key, $helper = null)
+    {
+        if (is_null($helper)) {
+            $className = '\\Kafka\\Protocol\\Fetch\\Helper\\' . $key;
+            if (!class_exists($className)) {
+                throw new \Kafka\Exception('helper is not exists.');
+            }
+            $helper = new $className();
+        }
+
+        if ($helper instanceof \Kafka\Protocol\Fetch\Helper\HelperAbstract) {
+            self::$helpers[$key] = $helper;
+        } else {
+            throw new \Kafka\Exception('this helper not instance of `\Kafka\Protocol\Fetch\Helper\HelperAbstract`');
+        }
+    }
+
+    // }}}
+    // {{{ public staitc function unRegisterHelper()
+
+    /**
+     * unregister helper
+     *
+     * @param string $key
+     * @static
+     * @access public
+     * @return void
+     */
+    public static function unRegisterHelper($key)
+    {
+        if (isset(self::$helpers[$key])) {
+            unset(self::$helpers[$key]);
+        }
+    }
+
+    // }}}
+    // {{{ public static function onStreamEof()
+
+    /**
+     * on stream eof call
+     *
+     * @param string $streamKey
+     * @static
+     * @access public
+     * @return void
+     */
+    public static function onStreamEof($streamKey)
+    {
+        if (empty(self::$helpers)) {
+            return false;
+        }
+
+        foreach (self::$helpers as $key => $helper) {
+            if (method_exists($helper, 'onStreamEof')) {
+                $helper->onStreamEof($streamKey);
+            }
+        }
+    }
+
+    // }}}
+    // {{{ public static function onTopicEof()
+
+    /**
+     * on topic eof call
+     *
+     * @param string $topicName
+     * @static
+     * @access public
+     * @return void
+     */
+    public static function onTopicEof($topicName)
+    {
+        if (empty(self::$helpers)) {
+            return false;
+        }
+
+        foreach (self::$helpers as $key => $helper) {
+            if (method_exists($helper, 'onTopicEof')) {
+                $helper->onStreamEof($topicName);
+            }
+        }
+    }
+
+    // }}}
+    // {{{ public static function onPartitionEof()
+
+    /**
+     * on partition eof call
+     *
+     * @param \Kafka\Protocol\Fetch\Partition $partition
+     * @static
+     * @access public
+     * @return void
+     */
+    public static function onPartitionEof($partition)
+    {
+        if (empty(self::$helpers)) {
+            return false;
+        }
+
+        foreach (self::$helpers as $key => $helper) {
+            if (method_exists($helper, 'onPartitionEof')) {
+                $helper->onPartitionEof($partition);
+            }
+        }
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/HelperAbstract.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/HelperAbstract.php
new file mode 100644
index 00000000..476f3da1
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Helper/HelperAbstract.php
@@ -0,0 +1,71 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol\Fetch\Helper;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+abstract class HelperAbstract
+{
+    // {{{ members
+    // }}}
+    // {{{ functions
+    // {{{ abstract public function onStreamEof()
+
+    /**
+     * on stream eof
+     *
+     * @param string $streamKey
+     * @access public
+     * @return void
+     */
+    abstract public function onStreamEof($streamKey);
+
+    // }}}
+    // {{{ abstract public function onTopicEof()
+
+    /**
+     * on topic eof
+     *
+     * @param string $topicName
+     * @access public
+     * @return void
+     */
+    abstract public function onTopicEof($topicName);
+
+    // }}}
+    // {{{ abstract public function onPartitionEof()
+
+    /**
+     * on partition eof
+     *
+     * @param \Kafka\Protocol\Fetch\Partition $partition
+     * @access public
+     * @return void
+     */
+    abstract public function onPartitionEof($partition);
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Message.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Message.php
new file mode 100644
index 00000000..42d7da1d
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Message.php
@@ -0,0 +1,175 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol\Fetch;
+
+use \Kafka\Protocol\Decoder;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Message
+{
+    // {{{ members
+
+    /**
+     * init read bytes
+     *
+     * @var float
+     * @access private
+     */
+    private $initOffset = 0;
+
+    /**
+     * validByteCount
+     *
+     * @var float
+     * @access private
+     */
+    private $validByteCount = 0;
+
+    /**
+     * crc32 code
+     *
+     * @var float
+     * @access private
+     */
+    private $crc = 0;
+
+    /**
+     * This is a version id used to allow backwards compatible evolution of the
+     * message binary format.
+     *
+     * @var float
+     * @access private
+     */
+    private $magic = 0;
+
+    /**
+     * The lowest 2 bits contain the compression codec used for the message. The
+     * other bits should be set to 0.
+     *
+     * @var float
+     * @access private
+     */
+    private $attribute = 0;
+
+    /**
+     * message key
+     *
+     * @var string
+     * @access private
+     */
+    private $key = '';
+
+    /**
+     * message value
+     *
+     * @var string
+     * @access private
+     */
+    private $value = '';
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @param string(raw) $msg
+     * @access public
+     * @return void
+     */
+    public function __construct($msg)
+    {
+        $offset = 0;
+        $crc = Decoder::unpack(Decoder::BIT_B32, substr($msg, $offset, 4));
+        $offset += 4;
+        $this->crc = array_shift($crc);
+        $magic = Decoder::unpack(Decoder::BIT_B8, substr($msg, $offset, 1));
+        $this->magic = array_shift($magic);
+        $offset += 1;
+        $attr  = Decoder::unpack(Decoder::BIT_B8, substr($msg, $offset, 1));
+        $this->attribute = array_shift($attr);
+        $offset += 1;
+        $keyLen = Decoder::unpack(Decoder::BIT_B32, substr($msg, $offset, 4));
+        $keyLen = array_shift($keyLen);
+        $offset += 4;
+        if ($keyLen > 0 && $keyLen != 0xFFFFFFFF) {
+            $this->key = substr($msg, $offset, $keyLen);
+            $offset += $keyLen;
+        }
+        $messageSize = Decoder::unpack(Decoder::BIT_B32, substr($msg, $offset, 4));
+        $messageSize = array_shift($messageSize);
+        $offset += 4;
+        if ($messageSize) {
+            $this->value = substr($msg, $offset, $messageSize);
+        }
+    }
+
+    // }}}
+    // {{{ public function getMessage()
+
+    /**
+     * get message data
+     *
+     * @access public
+     * @return string (raw)
+     */
+    public function getMessage()
+    {
+        return $this->value;
+    }
+
+    // }}}
+    // {{{ public function getMessageKey()
+
+    /**
+     * get message key
+     *
+     * @access public
+     * @return string (raw)
+     */
+    public function getMessageKey()
+    {
+        return $this->key;
+    }
+
+    // }}}
+    // {{{ public function __toString()
+
+    /**
+     * __toString
+     *
+     * @access public
+     * @return void
+     */
+    public function __toString()
+    {
+        return $this->value;
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/MessageSet.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/MessageSet.php
new file mode 100644
index 00000000..50413b67
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/MessageSet.php
@@ -0,0 +1,269 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol\Fetch;
+
+use \Kafka\Protocol\Decoder;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class MessageSet implements \Iterator
+{
+    // {{{ members
+
+    /**
+     * kafka socket object
+     *
+     * @var mixed
+     * @access private
+     */
+    private $stream = null;
+
+    /**
+     * messageSet size
+     *
+     * @var float
+     * @access private
+     */
+    private $messageSetSize = 0;
+
+    /**
+     * validByteCount
+     *
+     * @var float
+     * @access private
+     */
+    private $validByteCount = 0;
+
+    /**
+     * messageSet offset
+     *
+     * @var float
+     * @access private
+     */
+    private $offset = 0;
+
+    /**
+     * valid
+     *
+     * @var mixed
+     * @access private
+     */
+    private $valid = false;
+
+    /**
+     * partition object
+     *
+     * @var \Kafka\Protocol\Fetch\Partition
+     * @access private
+     */
+    private $partition = null;
+
+    /**
+     * request fetch context
+     *
+     * @var array
+     */
+    private $context = array();
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @param \Kafka\Socket $stream
+     * @param int $initOffset
+     * @access public
+     * @return void
+     */
+    public function __construct(\Kafka\Protocol\Fetch\Partition $partition, $context = array())
+    {
+        $this->stream = $partition->getStream();
+        $this->partition = $partition;
+        $this->context   = $context;
+        $this->messageSetSize = $this->getMessageSetSize();
+        \Kafka\Log::log("messageSetSize: {$this->messageSetSize}", LOG_INFO);
+    }
+
+    // }}}
+    // {{{ public function current()
+
+    /**
+     * current
+     *
+     * @access public
+     * @return void
+     */
+    public function current()
+    {
+        return $this->current;
+    }
+
+    // }}}
+    // {{{ public function key()
+
+    /**
+     * key
+     *
+     * @access public
+     * @return void
+     */
+    public function key()
+    {
+        return $this->validByteCount;
+    }
+
+    // }}}
+    // {{{ public function rewind()
+
+    /**
+     * implements Iterator function
+     *
+     * @access public
+     * @return integer
+     */
+    public function rewind()
+    {
+        $this->valid = $this->loadNextMessage();
+    }
+
+    // }}}
+    // {{{ public function valid()
+
+    /**
+     * implements Iterator function
+     *
+     * @access public
+     * @return integer
+     */
+    public function valid()
+    {
+        if (!$this->valid) {
+            $this->partition->setMessageOffset($this->offset);
+
+            // one partition iterator end
+            \Kafka\Protocol\Fetch\Helper\Helper::onPartitionEof($this->partition);
+        }
+
+        return $this->valid;
+    }
+
+    // }}}
+    // {{{ public function next()
+
+    /**
+     * implements Iterator function
+     *
+     * @access public
+     * @return integer
+     */
+    public function next()
+    {
+        $this->valid = $this->loadNextMessage();
+    }
+
+    // }}}
+    // {{{ protected function getMessageSetSize()
+
+    /**
+     * get message set size
+     *
+     * @access protected
+     * @return integer
+     */
+    protected function getMessageSetSize()
+    {
+        // read message size
+        $data = $this->stream->read(4, true);
+        $data = Decoder::unpack(Decoder::BIT_B32, $data);
+        $size = array_shift($data);
+        if ($size <= 0) {
+            throw new \Kafka\Exception\OutOfRange($size . ' is not a valid message size');
+        }
+
+        return $size;
+    }
+
+    // }}}
+    // {{{ public function loadNextMessage()
+
+    /**
+     * load next message
+     *
+     * @access public
+     * @return void
+     */
+    public function loadNextMessage()
+    {
+        if ($this->validByteCount >= $this->messageSetSize) {
+            return false;
+        }
+
+        try {
+            if ($this->validByteCount + 12 > $this->messageSetSize) {
+                // read socket buffer dirty data
+                $this->stream->read($this->messageSetSize - $this->validByteCount);
+                return false;
+            }
+            $offset = $this->stream->read(8, true);
+            $this->offset  = \Kafka\Protocol\Decoder::unpack(Decoder::BIT_B64, $offset);
+            $messageSize = $this->stream->read(4, true);
+            $messageSize = Decoder::unpack(Decoder::BIT_B32, $messageSize);
+            $messageSize = array_shift($messageSize);
+            $this->validByteCount += 12;
+            if (($this->validByteCount + $messageSize) > $this->messageSetSize) {
+                // read socket buffer dirty data
+                $this->stream->read($this->messageSetSize - $this->validByteCount);
+                return false;
+            }
+            $msg  = $this->stream->read($messageSize, true);
+            $this->current = new Message($msg);
+        } catch (\Kafka\Exception $e) {
+            \Kafka\Log::log("already fetch: {$this->validByteCount}, {$e->getMessage()}", LOG_INFO);
+            return false;
+        }
+
+        $this->validByteCount += $messageSize;
+
+        return true;
+    }
+
+    // }}}
+    // {{{ public function messageOffset()
+
+    /**
+     * current message offset in producer
+     *
+     * @return void
+     */
+    public function messageOffset()
+    {
+        return $this->offset;
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Partition.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Partition.php
new file mode 100644
index 00000000..9f8578d5
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Partition.php
@@ -0,0 +1,375 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol\Fetch;
+
+use \Kafka\Protocol\Decoder;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Partition implements \Iterator, \Countable
+{
+    // {{{ members
+
+    /**
+     * kafka socket object
+     *
+     * @var mixed
+     * @access private
+     */
+    private $stream = null;
+
+    /**
+     * validCount
+     *
+     * @var float
+     * @access private
+     */
+    private $validCount = 0;
+
+    /**
+     * partitions count
+     *
+     * @var float
+     * @access private
+     */
+    private $partitionCount = false;
+
+    /**
+     * current topic
+     *
+     * @var mixed
+     * @access private
+     */
+    private $current = null;
+
+    /**
+     * current iterator key
+     * partition id
+     *
+     * @var string
+     * @access private
+     */
+    private $key = null;
+
+    /**
+     * partition errCode
+     *
+     * @var float
+     * @access private
+     */
+    private $errCode = 0;
+
+    /**
+     * partition offset
+     *
+     * @var float
+     * @access private
+     */
+    private $offset = 0;
+
+    /**
+     * partition current fetch offset
+     *
+     * @var float
+     * @access private
+     */
+    private $currentOffset = 0;
+
+    /**
+     * valid
+     *
+     * @var mixed
+     * @access private
+     */
+    private $valid = false;
+
+    /**
+     * cuerrent topic name
+     *
+     * @var string
+     * @access private
+     */
+    private $topicName = '';
+
+    /**
+     * request fetch context
+     *
+     * @var array
+     */
+    private $context = array();
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @param \Kafka\Protocol\Fetch\Topic $topic
+     * @param int $initOffset
+     * @access public
+     * @return void
+     */
+    public function __construct(\Kafka\Protocol\Fetch\Topic $topic, $context = array())
+    {
+        $this->stream    = $topic->getStream();
+        $this->topicName = $topic->key();
+        $this->context   = $context;
+        $this->partitionCount = $this->getPartitionCount();
+    }
+
+    // }}}
+    // {{{ public function current()
+
+    /**
+     * current
+     *
+     * @access public
+     * @return void
+     */
+    public function current()
+    {
+        return $this->current;
+    }
+
+    // }}}
+    // {{{ public function key()
+
+    /**
+     * key
+     *
+     * @access public
+     * @return void
+     */
+    public function key()
+    {
+        return $this->key;
+    }
+
+    // }}}
+    // {{{ public function rewind()
+
+    /**
+     * implements Iterator function
+     *
+     * @access public
+     * @return integer
+     */
+    public function rewind()
+    {
+        $this->valid = $this->loadNextPartition();
+    }
+
+    // }}}
+    // {{{ public function valid()
+
+    /**
+     * implements Iterator function
+     *
+     * @access public
+     * @return integer
+     */
+    public function valid()
+    {
+        return $this->valid && $this->validCount <= $this->partitionCount;
+    }
+
+    // }}}
+    // {{{ public function next()
+
+    /**
+     * implements Iterator function
+     *
+     * @access public
+     * @return integer
+     */
+    public function next()
+    {
+        $this->valid = $this->loadNextPartition();
+    }
+
+    // }}}
+    // {{{ public function count()
+
+    /**
+     * implements Countable function
+     *
+     * @access public
+     * @return integer
+     */
+    public function count()
+    {
+        return $this->partitionCount;
+    }
+
+    // }}}
+    // {{{ public function getErrCode()
+
+    /**
+     * get partition errcode
+     *
+     * @access public
+     * @return void
+     */
+    public function getErrCode()
+    {
+        return $this->errCode;
+    }
+
+    // }}}
+    // {{{ public function getHighOffset()
+
+    /**
+     * get partition high offset
+     *
+     * @access public
+     * @return void
+     */
+    public function getHighOffset()
+    {
+        return $this->offset;
+    }
+
+    // }}}
+    // {{{ public function getTopicName()
+
+    /**
+     * get partition topic name
+     *
+     * @access public
+     * @return void
+     */
+    public function getTopicName()
+    {
+        return $this->topicName;
+    }
+
+    // }}}
+    // {{{ public function getStream()
+
+    /**
+     * get current stream
+     *
+     * @access public
+     * @return \Kafka\Socket
+     */
+    public function getStream()
+    {
+        return $this->stream;
+    }
+
+    // }}}
+    // {{{ protected function getPartitionCount()
+
+    /**
+     * get message size
+     * only use to object init
+     *
+     * @access protected
+     * @return integer
+     */
+    protected function getPartitionCount()
+    {
+        // read topic count
+        $data = $this->stream->read(4, true);
+        $data = Decoder::unpack(Decoder::BIT_B32, $data);
+        $count = array_shift($data);
+        if ($count <= 0) {
+            throw new \Kafka\Exception\OutOfRange($size . ' is not a valid partition count');
+        }
+
+        return $count;
+    }
+
+    // }}}
+    // {{{ public function loadNextPartition()
+
+    /**
+     * load next partition
+     *
+     * @access public
+     * @return void
+     */
+    public function loadNextPartition()
+    {
+        if ($this->validCount >= $this->partitionCount) {
+            return false;
+        }
+
+        try {
+            $partitionId = $this->stream->read(4, true);
+            $partitionId = Decoder::unpack(Decoder::BIT_B32, $partitionId);
+            $partitionId = array_shift($partitionId);
+            \Kafka\Log::log("kafka client:fetch partition:" . $partitionId, LOG_INFO);
+
+            $errCode = $this->stream->read(2, true);
+            $errCode = Decoder::unpack(Decoder::BIT_B16, $errCode);
+            $this->errCode = array_shift($errCode);
+            if ($this->errCode != 0) {
+                throw new \Kafka\Exception(\Kafka\Protocol\Decoder::getError($this->errCode));
+            }
+            $offset = $this->stream->read(8, true);
+            $this->offset  = \Kafka\Protocol\Decoder::unpack(Decoder::BIT_B64, $offset);
+
+            $this->key = $partitionId;
+            $this->current = new MessageSet($this, $this->context);
+        } catch (\Kafka\Exception $e) {
+            return false;
+        }
+
+        $this->validCount++;
+        return true;
+    }
+
+    // }}}
+    // {{{ public function setMessageOffset()
+
+    /**
+     * set messageSet fetch offset current
+     *
+     * @param  intger $offset
+     * @return void
+     */
+    public function setMessageOffset($offset)
+    {
+        $this->currentOffset = $offset;
+    }
+
+    // }}}
+    // {{{ public function getMessageOffset()
+
+    /**
+     * get messageSet fetch offset current
+     *
+     * @return int
+     */
+    public function getMessageOffset()
+    {
+        return $this->currentOffset;
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Topic.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Topic.php
new file mode 100644
index 00000000..500e6b1f
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Fetch/Topic.php
@@ -0,0 +1,345 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol\Fetch;
+
+use \Kafka\Protocol\Decoder;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Topic implements \Iterator, \Countable
+{
+    // {{{ members
+
+    /**
+     * kafka socket object
+     *
+     * @var array
+     * @access private
+     */
+    private $streams = array();
+
+    /**
+     * each topic count
+     *
+     * @var array
+     * @access private
+     */
+    private $topicCounts = array();
+
+    /**
+     * current iterator stream
+     *
+     * @var mixed
+     * @access private
+     */
+    private $currentStreamKey = 0;
+
+    /**
+     * current lock key
+     *
+     * @var string
+     * @access private
+     */
+    private $currentStreamLockKey = '';
+
+    /**
+     * currentStreamCount
+     *
+     * @var float
+     * @access private
+     */
+    private $currentStreamCount = 0;
+
+    /**
+     * validCount
+     *
+     * @var float
+     * @access private
+     */
+    private $validCount = 0;
+
+    /**
+     * topic count
+     *
+     * @var float
+     * @access private
+     */
+    private $topicCount = false;
+
+    /**
+     * current topic
+     *
+     * @var mixed
+     * @access private
+     */
+    private $current = null;
+
+    /**
+     * current iterator key
+     * topic name
+     *
+     * @var string
+     * @access private
+     */
+    private $key = null;
+
+    /**
+     * valid
+     *
+     * @var mixed
+     * @access private
+     */
+    private $valid = false;
+
+    /**
+     * request fetch context
+     *
+     * @var array
+     */
+    private $context = array();
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @param \Kafka\Socket $stream
+     * @param int $initOffset
+     * @access public
+     * @return void
+     */
+    public function __construct($streams, $context = array())
+    {
+        if (!is_array($streams)) {
+            $streams = array($streams);
+        }
+        $this->streams = $streams;
+        $topicInfos = array();
+        foreach ($context as $values) {
+            if (!isset($values['data'])) {
+                continue;
+            }
+
+            foreach ($values['data'] as $value) {
+                if (!isset($value['topic_name']) || !isset($value['partitions'])) {
+                    continue;
+                }
+
+                $topicName = $value['topic_name'];
+                foreach ($value['partitions'] as $part) {
+                    $topicInfos[$topicName][$part['partition_id']] = array(
+                        'offset' => $part['offset'],
+                    );
+                }
+            }
+        }
+        $this->context = $topicInfos;
+        $this->topicCount = $this->getTopicCount();
+    }
+
+    // }}}
+    // {{{ public function current()
+
+    /**
+     * current
+     *
+     * @access public
+     * @return void
+     */
+    public function current()
+    {
+        return $this->current;
+    }
+
+    // }}}
+    // {{{ public function key()
+
+    /**
+     * key
+     *
+     * @access public
+     * @return void
+     */
+    public function key()
+    {
+        return $this->key;
+    }
+
+    // }}}
+    // {{{ public function rewind()
+
+    /**
+     * implements Iterator function
+     *
+     * @access public
+     * @return integer
+     */
+    public function rewind()
+    {
+        $this->valid = $this->loadNextTopic();
+    }
+
+    // }}}
+    // {{{ public function valid()
+
+    /**
+     * implements Iterator function
+     *
+     * @access public
+     * @return integer
+     */
+    public function valid()
+    {
+        return $this->valid;
+    }
+
+    // }}}
+    // {{{ public function next()
+
+    /**
+     * implements Iterator function
+     *
+     * @access public
+     * @return integer
+     */
+    public function next()
+    {
+        $this->valid = $this->loadNextTopic();
+    }
+
+    // }}}
+    // {{{ public function count()
+
+    /**
+     * implements Countable function
+     *
+     * @access public
+     * @return integer
+     */
+    public function count()
+    {
+        return $this->topicCount;
+    }
+
+    // }}}
+    // {{{ protected function getTopicCount()
+
+    /**
+     * get message size
+     * only use to object init
+     *
+     * @access protected
+     * @return integer
+     */
+    protected function getTopicCount()
+    {
+        $count = 0;
+        foreach (array_values($this->streams) as $key => $stream) {
+            // read topic count
+            $stream->read(8, true);
+            $data = $stream->read(4, true);
+            $data = Decoder::unpack(Decoder::BIT_B32, $data);
+            $topicCount = array_shift($data);
+            $count += $topicCount;
+            $this->topicCounts[$key] = $topicCount;
+            if ($count <= 0) {
+                throw new \Kafka\Exception\OutOfRange($count . ' is not a valid topic count');
+            }
+        }
+
+        return $count;
+    }
+
+    // }}}
+    // {{{ public function loadNextTopic()
+
+    /**
+     * load next topic
+     *
+     * @access public
+     * @return void
+     */
+    public function loadNextTopic()
+    {
+        if ($this->validCount >= $this->topicCount) {
+            \Kafka\Protocol\Fetch\Helper\Helper::onStreamEof($this->currentStreamLockKey);
+            return false;
+        }
+
+        if ($this->currentStreamCount >= $this->topicCounts[$this->currentStreamKey]) {
+            \Kafka\Protocol\Fetch\Helper\Helper::onStreamEof($this->currentStreamLockKey);
+            $this->currentStreamKey++;
+        }
+
+        $lockKeys = array_keys($this->streams);
+        $streams  = array_values($this->streams);
+        if (!isset($streams[$this->currentStreamKey])) {
+            return false;
+        }
+
+        $stream = $streams[$this->currentStreamKey];
+        $this->currentStreamLockKey = $lockKeys[$this->currentStreamKey];
+
+        try {
+            $topicLen = $stream->read(2, true);
+            $topicLen = Decoder::unpack(Decoder::BIT_B16, $topicLen);
+            $topicLen = array_shift($topicLen);
+            if ($topicLen <= 0) {
+                return false;
+            }
+
+            // topic name
+            $this->key = $stream->read($topicLen, true);
+            $this->current = new Partition($this, $this->context);
+        } catch (\Kafka\Exception $e) {
+            return false;
+        }
+
+        $this->validCount++;
+        $this->currentStreamCount++;
+
+        return true;
+    }
+
+    // }}}
+    // {{{ public function getStream()
+
+    /**
+     * get current stream
+     *
+     * @access public
+     * @return \Kafka\Socket
+     */
+    public function getStream()
+    {
+        $streams = array_values($this->streams);
+        return $streams[$this->currentStreamKey];
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Protocol/Protocol.php b/vendor/nmred/kafka-php/src/Kafka/Protocol/Protocol.php
new file mode 100644
index 00000000..a31067b5
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Protocol/Protocol.php
@@ -0,0 +1,230 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka\Protocol;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+abstract class Protocol
+{
+    // {{{ consts
+
+    /**
+     *  Kafka server protocol version
+     */
+    const API_VERSION = 0;
+
+    /**
+     * use encode message, This is a version id used to allow backwards
+     * compatible evolution of the message binary format.
+     */
+    const MESSAGE_MAGIC = 0;
+
+    /**
+     * message no compression
+     */
+    const COMPRESSION_NONE = 0;
+
+    /**
+     * Message using gzip compression
+     */
+    const COMPRESSION_GZIP = 1;
+
+    /**
+     * Message using Snappy compression
+     */
+    const COMPRESSION_SNAPPY = 2;
+
+    /**
+     *  pack int32 type
+     */
+    const PACK_INT32 = 0;
+
+    /**
+     * pack int16 type
+     */
+    const PACK_INT16 = 1;
+
+    /**
+     * protocol request code
+     */
+    const PRODUCE_REQUEST = 0;
+    const FETCH_REQUEST   = 1;
+    const OFFSET_REQUEST  = 2;
+    const METADATA_REQUEST      = 3;
+    const OFFSET_COMMIT_REQUEST = 8;
+    const OFFSET_FETCH_REQUEST  = 9;
+    const CONSUMER_METADATA_REQUEST = 10;
+
+    // unpack/pack bit
+    const BIT_B64 = 'N2';
+    const BIT_B32 = 'N';
+    const BIT_B16 = 'n';
+    const BIT_B8  = 'C';
+
+    // }}}
+    // {{{ members
+
+    /**
+     * stream
+     *
+     * @var mixed
+     * @access protected
+     */
+    protected $stream = null;
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @param \Kafka\Socket $stream
+     * @access public
+     * @return void
+     */
+    public function __construct(\Kafka\Socket $stream)
+    {
+        $this->stream = $stream;
+    }
+
+    // }}}
+    // {{{ public static function Khex2bin()
+
+    /**
+     * hex to bin
+     *
+     * @param string $string
+     * @static
+     * @access protected
+     * @return string (raw)
+     */
+    public static function Khex2bin($string)
+    {
+        if (function_exists('\hex2bin')) {
+            return \hex2bin($string);
+        } else {
+            $bin = '';
+            $len = strlen($string);
+            for ($i = 0; $i < $len; $i += 2) {
+                $bin .= pack('H*', substr($string, $i, 2));
+            }
+
+            return $bin;
+        }
+    }
+
+    // }}}
+    // {{{ public static function unpack()
+
+    /**
+     * Unpack a bit integer as big endian long
+     *
+     * @static
+     * @access public
+     * @return integer
+     */
+    public static function unpack($type, $bytes)
+    {
+        self::checkLen($type, $bytes);
+        if ($type == self::BIT_B64) {
+            $set = unpack($type, $bytes);
+            $original = ($set[1] & 0xFFFFFFFF) << 32 | ($set[2] & 0xFFFFFFFF);
+            return $original;
+        } else {
+            return unpack($type, $bytes);
+        }
+    }
+
+    // }}}
+    // {{{ public static function pack()
+
+    /**
+     * pack a bit integer as big endian long
+     *
+     * @static
+     * @access public
+     * @return integer
+     */
+    public static function pack($type, $data)
+    {
+        if ($type == self::BIT_B64) {
+            if ($data == -1) { // -1L
+                $data = self::Khex2bin('ffffffffffffffff');
+            } elseif ($data == -2) { // -2L
+                $data = self::Khex2bin('fffffffffffffffe');
+            } else {
+                $left  = 0xffffffff00000000;
+                $right = 0x00000000ffffffff;
+
+                $l = ($data & $left) >> 32;
+                $r = $data & $right;
+                $data = pack($type, $l, $r);
+            }
+        } else {
+            $data = pack($type, $data);
+        }
+
+        return $data;
+    }
+
+    // }}}
+    // {{{ protected static function checkLen()
+
+    /**
+     * check unpack bit is valid
+     *
+     * @param string $type
+     * @param string(raw) $bytes
+     * @static
+     * @access protected
+     * @return void
+     */
+    protected static function checkLen($type, $bytes)
+    {
+        $len = 0;
+        switch($type) {
+            case self::BIT_B64:
+                $len = 8;
+                break;
+            case self::BIT_B32:
+                $len = 4;
+                break;
+            case self::BIT_B16:
+                $len = 2;
+                break;
+            case self::BIT_B8:
+                $len = 1;
+                break;
+        }
+
+        if (strlen($bytes) != $len) {
+            throw new \Kafka\Exception\Protocol('unpack failed. string(raw) length is ' . strlen($bytes) . ' , TO ' . $type);
+        }
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/Socket.php b/vendor/nmred/kafka-php/src/Kafka/Socket.php
new file mode 100644
index 00000000..be7321f3
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/Socket.php
@@ -0,0 +1,365 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class Socket
+{
+    // {{{ consts
+
+    const READ_MAX_LEN = 5242880; // read socket max length 5MB
+
+    // }}}
+    // {{{ members
+
+    /**
+     * Send timeout in seconds.
+     *
+     * @var float
+     * @access private
+     */
+    private $sendTimeoutSec = 0;
+
+    /**
+     * Send timeout in microseconds.
+     *
+     * @var float
+     * @access private
+     */
+    private $sendTimeoutUsec = 100000;
+
+    /**
+     * Recv timeout in seconds
+     *
+     * @var float
+     * @access private
+     */
+    private $recvTimeoutSec = 0;
+
+    /**
+     * Recv timeout in microseconds
+     *
+     * @var float
+     * @access private
+     */
+    private $recvTimeoutUsec = 750000;
+
+    /**
+     * Stream resource
+     *
+     * @var mixed
+     * @access private
+     */
+    private $stream = null;
+
+    /**
+     * Socket host
+     *
+     * @var mixed
+     * @access private
+     */
+    private $host = null;
+
+    /**
+     * Socket port
+     *
+     * @var mixed
+     * @access private
+     */
+    private $port = -1;
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @access public
+     * @return void
+     */
+    public function __construct($host, $port, $recvTimeoutSec = 0, $recvTimeoutUsec = 750000, $sendTimeoutSec = 0, $sendTimeoutUsec = 100000)
+    {
+        $this->host = $host;
+        $this->port = $port;
+        $this->setRecvTimeoutSec($recvTimeoutSec);
+        $this->setRecvTimeoutUsec($recvTimeoutUsec);
+        $this->setSendTimeoutSec($sendTimeoutSec);
+        $this->setSendTimeoutUsec($sendTimeoutUsec);
+    }
+
+    /**
+     * @param float $sendTimeoutSec
+     */
+    public function setSendTimeoutSec($sendTimeoutSec)
+    {
+        $this->sendTimeoutSec = $sendTimeoutSec;
+    }
+
+    /**
+     * @param float $sendTimeoutUsec
+     */
+    public function setSendTimeoutUsec($sendTimeoutUsec)
+    {
+        $this->sendTimeoutUsec = $sendTimeoutUsec;
+    }
+
+    /**
+     * @param float $recvTimeoutSec
+     */
+    public function setRecvTimeoutSec($recvTimeoutSec)
+    {
+        $this->recvTimeoutSec = $recvTimeoutSec;
+    }
+
+    /**
+     * @param float $recvTimeoutUsec
+     */
+    public function setRecvTimeoutUsec($recvTimeoutUsec)
+    {
+        $this->recvTimeoutUsec = $recvTimeoutUsec;
+    }
+
+
+
+    // }}}
+    // {{{ public static function createFromStream()
+
+    /**
+     * Optional method to set the internal stream handle
+     *
+     * @static
+     * @access public
+     * @return void
+     */
+    public static function createFromStream($stream)
+    {
+        $socket = new self('localhost', 0);
+        $socket->setStream($stream);
+        return $socket;
+    }
+
+    // }}}
+    // {{{ public function setStream()
+
+    /**
+     * Optional method to set the internal stream handle
+     *
+     * @param mixed $stream
+     * @access public
+     * @return void
+     */
+    public function setStream($stream)
+    {
+        $this->stream = $stream;
+    }
+
+    // }}}
+    // {{{ public function connect()
+
+    /**
+     * Connects the socket
+     *
+     * @access public
+     * @return void
+     */
+    public function connect()
+    {
+        if (is_resource($this->stream)) {
+            return false;
+        }
+
+        if (empty($this->host)) {
+            throw new \Kafka\Exception('Cannot open null host.');
+        }
+        if ($this->port <= 0) {
+            throw new \Kafka\Exception('Cannot open without port.');
+        }
+
+        $this->stream = @fsockopen(
+            $this->host,
+            $this->port,
+            $errno,
+            $errstr,
+            $this->sendTimeoutSec + ($this->sendTimeoutUsec / 1000000)
+        );
+
+        if ($this->stream == false) {
+            $error = 'Could not connect to '
+                    . $this->host . ':' . $this->port
+                    . ' ('.$errstr.' ['.$errno.'])';
+            throw new \Kafka\Exception\SocketConnect($error);
+        }
+
+        stream_set_blocking($this->stream, 0);
+    }
+
+    // }}}
+    // {{{ public function close()
+
+    /**
+     * close the socket
+     *
+     * @access public
+     * @return void
+     */
+    public function close()
+    {
+        if (is_resource($this->stream)) {
+            fclose($this->stream);
+        }
+    }
+
+    // }}}
+    // {{{ public function read()
+
+    /**
+     * Read from the socket at most $len bytes.
+     *
+     * This method will not wait for all the requested data, it will return as
+     * soon as any data is received.
+     *
+     * @param integer $len               Maximum number of bytes to read.
+     * @param boolean $verifyExactLength Throw an exception if the number of read bytes is less than $len
+     *
+     * @return string Binary data
+     * @throws Kafka_Exception_Socket
+     */
+    public function read($len, $verifyExactLength = false)
+    {
+        if ($len > self::READ_MAX_LEN) {
+            throw new \Kafka\Exception\SocketEOF('Could not read '.$len.' bytes from stream, length too longer.');
+        }
+
+        $null = null;
+        $read = array($this->stream);
+        $readable = @stream_select($read, $null, $null, $this->recvTimeoutSec, $this->recvTimeoutUsec);
+        if ($readable > 0) {
+            $remainingBytes = $len;
+            $data = $chunk = '';
+            while ($remainingBytes > 0) {
+                $chunk = fread($this->stream, $remainingBytes);
+                if ($chunk === false) {
+                    $this->close();
+                    throw new \Kafka\Exception\SocketEOF('Could not read '.$len.' bytes from stream (no data)');
+                }
+                if (strlen($chunk) === 0) {
+                    // Zero bytes because of EOF?
+                    if (feof($this->stream)) {
+                        $this->close();
+                        throw new \Kafka\Exception\SocketEOF('Unexpected EOF while reading '.$len.' bytes from stream (no data)');
+                    }
+                    // Otherwise wait for bytes
+                    $readable = @stream_select($read, $null, $null, $this->recvTimeoutSec, $this->recvTimeoutUsec);
+                    if ($readable !== 1) {
+                        throw new \Kafka\Exception\SocketTimeout('Timed out reading socket while reading ' . $len . ' bytes with ' . $remainingBytes . ' bytes to go');
+                    }
+                    continue; // attempt another read
+                }
+                $data .= $chunk;
+                $remainingBytes -= strlen($chunk);
+            }
+            if ($len === $remainingBytes || ($verifyExactLength && $len !== strlen($data))) {
+                // couldn't read anything at all OR reached EOF sooner than expected
+                $this->close();
+                throw new \Kafka\Exception\SocketEOF('Read ' . strlen($data) . ' bytes instead of the requested ' . $len . ' bytes');
+            }
+
+            return $data;
+        }
+        if (false !== $readable) {
+            $res = stream_get_meta_data($this->stream);
+            if (!empty($res['timed_out'])) {
+                $this->close();
+                throw new \Kafka\Exception\SocketTimeout('Timed out reading '.$len.' bytes from stream');
+            }
+        }
+        $this->close();
+        throw new \Kafka\Exception\SocketEOF('Could not read '.$len.' bytes from stream (not readable)');
+
+    }
+
+    // }}}
+    // {{{ public function write()
+
+    /**
+     * Write to the socket.
+     *
+     * @param string $buf The data to write
+     *
+     * @return integer
+     * @throws Kafka_Exception_Socket
+     */
+    public function write($buf)
+    {
+        $null = null;
+        $write = array($this->stream);
+
+        // fwrite to a socket may be partial, so loop until we
+        // are done with the entire buffer
+        $written = 0;
+        $buflen = strlen($buf);
+        while ( $written < $buflen ) {
+            // wait for stream to become available for writing
+            $writable = stream_select($null, $write, $null, $this->sendTimeoutSec, $this->sendTimeoutUsec);
+            if ($writable > 0) {
+                // write remaining buffer bytes to stream
+                $wrote = fwrite($this->stream, substr($buf, $written));
+                if ($wrote === -1 || $wrote === false) {
+                    throw new \Kafka\Exception\Socket('Could not write ' . strlen($buf) . ' bytes to stream, completed writing only ' . $written . ' bytes');
+                }
+                $written += $wrote;
+                continue;
+            }
+            if (false !== $writable) {
+                $res = stream_get_meta_data($this->stream);
+                if (!empty($res['timed_out'])) {
+                    throw new \Kafka\Exception\SocketTimeout('Timed out writing ' . strlen($buf) . ' bytes to stream after writing ' . $written . ' bytes');
+                }
+            }
+            throw new \Kafka\Exception\Socket('Could not write ' . strlen($buf) . ' bytes to stream');
+        }
+        return $written;
+    }
+
+    // }}}
+    // {{{ public function rewind()
+
+    /**
+     * Rewind the stream
+     *
+     * @return void
+     */
+    public function rewind()
+    {
+        if (is_resource($this->stream)) {
+            rewind($this->stream);
+        }
+    }
+
+    // }}}
+    // }}}
+}
diff --git a/vendor/nmred/kafka-php/src/Kafka/ZooKeeper.php b/vendor/nmred/kafka-php/src/Kafka/ZooKeeper.php
new file mode 100644
index 00000000..f48b5cb9
--- /dev/null
+++ b/vendor/nmred/kafka-php/src/Kafka/ZooKeeper.php
@@ -0,0 +1,364 @@
+<?php
+/* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
+// +---------------------------------------------------------------------------
+// | SWAN [ $_SWANBR_SLOGAN_$ ]
+// +---------------------------------------------------------------------------
+// | Copyright $_SWANBR_COPYRIGHT_$
+// +---------------------------------------------------------------------------
+// | Version  $_SWANBR_VERSION_$
+// +---------------------------------------------------------------------------
+// | Licensed ( $_SWANBR_LICENSED_URL_$ )
+// +---------------------------------------------------------------------------
+// | $_SWANBR_WEB_DOMAIN_$
+// +---------------------------------------------------------------------------
+
+namespace Kafka;
+
+/**
++------------------------------------------------------------------------------
+* Kafka protocol since Kafka v0.8
++------------------------------------------------------------------------------
+*
+* @package
+* @version $_SWANBR_VERSION_$
+* @copyright Copyleft
+* @author $_SWANBR_AUTHOR_$
++------------------------------------------------------------------------------
+*/
+
+class ZooKeeper implements \Kafka\ClusterMetaData
+{
+    // {{{ consts
+
+    /**
+     * get all broker
+     */
+    const BROKER_PATH = '/brokers/ids';
+
+    /**
+     * get broker detail
+     */
+    const BROKER_DETAIL_PATH = '/brokers/ids/%d';
+
+    /**
+     * get topic detail
+     */
+    const TOPIC_PATCH = '/brokers/topics/%s';
+
+    /**
+     * get partition state
+     */
+    const PARTITION_STATE = '/brokers/topics/%s/partitions/%d/state';
+
+    /**
+     * register consumer
+     */
+    const REG_CONSUMER = '/consumers/%s/ids/%s';
+
+    /**
+     * list consumer
+     */
+    const LIST_CONSUMER = '/consumers/%s/ids';
+
+    /**
+     * partition owner
+     */
+    const PARTITION_OWNER = '/consumers/%s/owners/%s/%d';
+
+    // }}}
+    // {{{ members
+
+    /**
+     * zookeeper
+     *
+     * @var mixed
+     * @access private
+     */
+    private $zookeeper = null;
+
+    // }}}
+    // {{{ functions
+    // {{{ public function __construct()
+
+    /**
+     * __construct
+     *
+     * @access public
+     * @return void
+     */
+    public function __construct($hostList, $timeout = null)
+    {
+        if (!is_null($timeout) && is_numeric($timeout)) {
+            $this->zookeeper = new \ZooKeeper($hostList, null, $timeout);
+        } else {
+            $this->zookeeper = new \ZooKeeper($hostList);
+        }
+    }
+
+    // }}}
+    // {{{ public function listBrokers()
+
+    /**
+     * get broker list using zookeeper
+     *
+     * @access public
+     * @return array
+     */
+    public function listBrokers()
+    {
+        $result = array();
+        $lists = $this->zookeeper->getChildren(self::BROKER_PATH);
+        if (!empty($lists)) {
+            foreach ($lists as $brokerId) {
+                $brokerDetail = $this->getBrokerDetail($brokerId);
+                if (!$brokerDetail) {
+                    continue;
+                }
+                $result[$brokerId] = $brokerDetail;
+            }
+        }
+
+        return $result;
+    }
+
+    // }}}
+    // {{{ public function getBrokerDetail()
+
+    /**
+     * get broker detail
+     *
+     * @param integer $brokerId
+     * @access public
+     * @return void
+     */
+    public function getBrokerDetail($brokerId)
+    {
+        $result = array();
+        $path = sprintf(self::BROKER_DETAIL_PATH, (int) $brokerId);
+        if ($this->zookeeper->exists($path)) {
+            $result = $this->zookeeper->get($path);
+            if (!$result) {
+                return false;
+            }
+
+            $result = json_decode($result, true);
+        }
+
+        return $result;
+    }
+
+    // }}}
+    // {{{ public function getTopicDetail()
+
+    /**
+     * get topic detail
+     *
+     * @param string $topicName
+     * @access public
+     * @return void
+     */
+    public function getTopicDetail($topicName)
+    {
+        $result = array();
+        $path = sprintf(self::TOPIC_PATCH, (string) $topicName);
+        if ($this->zookeeper->exists($path)) {
+            $result = $this->zookeeper->get($path);
+            if (!$result) {
+                return false;
+            }
+            $result = json_decode($result, true);
+        }
+
+        return $result;
+    }
+
+    // }}}
+    // {{{ public function getPartitionState()
+
+    /**
+     * get partition state
+     *
+     * @param string $topicName
+     * @param integer $partitionId
+     * @access public
+     * @return void
+     */
+    public function getPartitionState($topicName, $partitionId = 0)
+    {
+        $result = array();
+        $path = sprintf(self::PARTITION_STATE, (string) $topicName, (int) $partitionId);
+        if ($this->zookeeper->exists($path)) {
+            $result = $this->zookeeper->get($path);
+            if (!$result) {
+                return false;
+            }
+            $result = json_decode($result, true);
+        }
+
+        return $result;
+    }
+
+    // }}}
+    // {{{ public function registerConsumer()
+
+    /**
+     * register consumer
+     *
+     * @param string $topicName
+     * @param integer $partitionId
+     * @access public
+     * @return void
+     */
+    public function registerConsumer($groupId, $consumerId, $topics = array())
+    {
+        if (empty($topics)) {
+            return true;
+        }
+
+        $path = sprintf(self::REG_CONSUMER, (string) $groupId, (string) $consumerId);
+        $subData = array();
+        foreach ($topics as $topic) {
+            $subData[$topic] = 1;
+        }
+        $data = array(
+            'version' => '1',
+            'pattern' => 'white_list',
+            'subscription' => $subData,
+        );
+        if (!$this->zookeeper->exists($path)) {
+            $this->makeZkPath($path);
+            $this->makeZkNode($path, json_encode($data));
+        } else {
+            $this->zookeeper->set($path, json_encode($data));
+        }
+    }
+
+    // }}}
+    // {{{ public function listConsumer()
+
+    /**
+     * list consumer
+     *
+     * @param string $groupId
+     * @access public
+     * @return void
+     */
+    public function listConsumer($groupId)
+    {
+        $path = sprintf(self::LIST_CONSUMER, (string) $groupId);
+        if (!$this->zookeeper->exists($path)) {
+            return array();
+        } else {
+            return $this->zookeeper->getChildren($path);
+        }
+    }
+
+    // }}}
+    // {{{ public function getConsumersPerTopic()
+
+    /**
+     * get consumer per topic
+     *
+     * @param string $groupId
+     * @access public
+     * @return array
+     */
+    public function getConsumersPerTopic($groupId)
+    {
+        $consumers = $this->listConsumer($groupId);
+        if (empty($consumers)) {
+            return array();
+        }
+
+        $topics = array();
+        foreach ($consumers as $consumerId) {
+            $path = sprintf(self::REG_CONSUMER, (string) $groupId, (string) $consumerId);
+            if (!$this->zookeeper->exists($path)) {
+                continue;
+            }
+
+            $info = $this->zookeeper->get($path);
+            $info = json_decode($info, true);
+            $subTopic = isset($info['subscription']) ? $info['subscription'] : array();
+            foreach ($subTopic as $topic => $num) {
+                $topics[$topic] = $consumerId;
+            }
+        }
+
+        return $topics;
+    }
+
+    // }}}
+    // {{{ public function addPartitionOwner()
+
+    /**
+     * add partition owner
+     *
+     * @param string $groupId
+     * @param string $topicName
+     * @param integer $partitionId
+     * @param string $consumerId
+     * @access public
+     * @return void
+     */
+    public function addPartitionOwner($groupId, $topicName, $partitionId, $consumerId)
+    {
+        $path = sprintf(self::PARTITION_OWNER, (string) $groupId, $topicName, (string) $partitionId);
+        if (!$this->zookeeper->exists($path)) {
+            $this->makeZkPath($path);
+            $this->makeZkNode($path, $consumerId);
+        } else {
+            $this->zookeeper->set($path, $consumerId);
+        }
+    }
+
+    // }}}
+    // {{{ protected function makeZkPath()
+
+    /**
+     * Equivalent of "mkdir -p" on ZooKeeper
+     *
+     * @param string $path  The path to the node
+     * @param mixed  $value The value to assign to each new node along the path
+     *
+     * @return bool
+     */
+    protected function makeZkPath($path, $value = 0)
+    {
+        $parts = explode('/', $path);
+        $parts = array_filter($parts);
+        $subpath = '';
+        while (count($parts) > 1) {
+            $subpath .= '/' . array_shift($parts);
+            if (!$this->zookeeper->exists($subpath)) {
+                $this->makeZkNode($subpath, $value);
+            }
+        }
+    }
+
+    // }}}
+    // {{{ protected function makeZkNode()
+
+    /**
+     * Create a node on ZooKeeper at the given path
+     *
+     * @param string $path  The path to the node
+     * @param mixed  $value The value to assign to the new node
+     *
+     * @return bool
+     */
+    protected function makeZkNode($path, $value)
+    {
+        $params = array(
+            array(
+                'perms'  => \Zookeeper::PERM_ALL,
+                'scheme' => 'world',
+                'id'     => 'anyone',
+            )
+        );
+        return $this->zookeeper->create($path, $value, $params);
+    }
+
+    // }}}
+    // }}}
+}