--- /dev/null
+<?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 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
+ * @param $client
+ * @param $groupId
+ * @param $topicName
+ * @param int $partitionId
+ */
+ 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 int
+ */
+ 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:
+ Log::log("topic name: $topicName, partitionId: $partitionId, get offset value is default last.", LOG_INFO);
+ return $maxOffset;
+ 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;
+ }
+ } 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 integer $timeLine
+ * @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']);
+ }
+
+ // }}}
+ // }}}
+}
scripts.mit.edu / autoinstallsdev / mediawiki.git / blobdiff