2 /* vim: set expandtab tabstop=4 shiftwidth=4 softtabstop=4 foldmethod=marker: */
3 // +---------------------------------------------------------------------------
4 // | SWAN [ $_SWANBR_SLOGAN_$ ]
5 // +---------------------------------------------------------------------------
6 // | Copyright $_SWANBR_COPYRIGHT_$
7 // +---------------------------------------------------------------------------
8 // | Version $_SWANBR_VERSION_$
9 // +---------------------------------------------------------------------------
10 // | Licensed ( $_SWANBR_LICENSED_URL_$ )
11 // +---------------------------------------------------------------------------
12 // | $_SWANBR_WEB_DOMAIN_$
13 // +---------------------------------------------------------------------------
18 +------------------------------------------------------------------------------
19 * Kafka protocol since Kafka v0.8
20 +------------------------------------------------------------------------------
23 * @version $_SWANBR_VERSION_$
25 * @author $_SWANBR_AUTHOR_$
26 +------------------------------------------------------------------------------
38 * @var \Kafka\ClusterMetaData
41 private $metadata = null;
49 private $hostList = array();
52 * save broker connection
57 private static $stream = array();
60 * default stream options
65 private $streamOptions = array(
66 'RecvTimeoutSec' => 0,
67 'RecvTimeoutUsec' => 750000,
68 'SendTimeoutSec' => 0,
69 'SendTimeoutUsec' => 100000,
74 // {{{ public function __construct()
80 * @param ClusterMetaData $metadata
82 public function __construct(ClusterMetaData $metadata)
84 $this->metadata = $metadata;
85 if (method_exists($metadata, 'setClient')) {
86 $this->metadata->setClient($this);
91 * update stream options
93 * @param array $options
95 public function setStreamOptions($options = array())
98 $this->streamOptions = array_merge($this->streamOptions, $options);
99 $this->updateStreamOptions();
104 * @param $name - name of stream option
105 * @param $value - value for option
107 public function setStreamOption($name, $value)
109 $this->streamOptions[$name] = $value;
110 $this->updateStreamOptions();
115 * @param $name - name of option
118 public function getStreamOption($name)
120 if (array_key_exists($name, $this->streamOptions)) {
121 return $this->streamOptions[$name];
129 private function updateStreamOptions()
131 // Loop thru each stream
132 foreach (self::$stream as $host => $streams) {
133 foreach ($streams as $key => $info) {
135 if (isset($info['stream'])) {
136 /** @var \Kafka\Socket $stream */
137 $stream = $info['stream'];
138 $stream->setRecvTimeoutSec($this->streamOptions['RecvTimeoutSec']);
139 $stream->setRecvTimeoutUsec($this->streamOptions['SendTimeoutUsec']);
140 $stream->setSendTimeoutSec($this->streamOptions['SendTimeoutSec']);
141 $stream->setSendTimeoutUsec($this->streamOptions['SendTimeoutUsec']);
148 // {{{ public function getBrokers()
156 public function getBrokers()
158 if (empty($this->hostList)) {
159 $brokerList = $this->metadata->listBrokers();
160 foreach ($brokerList as $brokerId => $info) {
161 if (!isset($info['host']) || !isset($info['port'])) {
164 $this->hostList[$brokerId] = $info['host'] . ':' . $info['port'];
168 return $this->hostList;
172 // {{{ public function getHostByPartition()
175 * get broker host by topic partition
177 * @param string $topicName
178 * @param int $partitionId
182 public function getHostByPartition($topicName, $partitionId = 0)
184 $partitionInfo = $this->metadata->getPartitionState($topicName, $partitionId);
185 if (!$partitionInfo) {
186 throw new \Kafka\Exception('topic:' . $topicName . ', partition id: ' . $partitionId . ' is not exists.');
189 $hostList = $this->getBrokers();
190 if (isset($partitionInfo['leader']) && isset($hostList[$partitionInfo['leader']])) {
191 return $hostList[$partitionInfo['leader']];
193 throw new \Kafka\Exception('can\'t find broker host.');
198 // {{{ public function getZooKeeper()
201 * get kafka zookeeper object
204 * @return \Kafka\ZooKeeper
206 public function getZooKeeper()
208 if ($this->metadata instanceof \Kafka\ZooKeeper) {
209 return $this->metadata;
211 throw new \Kafka\Exception( 'ZooKeeper was not provided' );
216 // {{{ public function getStream()
219 * get broker broker connect
221 * @param string $host
222 * @param null $lockKey
226 public function getStream($host, $lockKey = null)
229 $lockKey = uniqid($host);
232 list($hostname, $port) = explode(':', $host);
233 // find unlock stream
234 if (isset(self::$stream[$host])) {
235 foreach (self::$stream[$host] as $key => $info) {
236 if ($info['locked']) {
239 self::$stream[$host][$key]['locked'] = true;
240 $info['stream']->connect();
241 return array('key' => $key, 'stream' => $info['stream']);
247 $stream = new \Kafka\Socket($hostname, $port, $this->getStreamOption('RecvTimeoutSec'), $this->getStreamOption('RecvTimeoutUsec'), $this->getStreamOption('SendTimeoutSec'), $this->getStreamOption('SendTimeoutUsec'));
249 self::$stream[$host][$lockKey] = array(
253 return array('key' => $lockKey, 'stream' => $stream);
257 // {{{ public function freeStream()
266 public function freeStream($key)
268 foreach (self::$stream as $host => $values) {
269 if (isset($values[$key])) {
270 self::$stream[$host][$key]['locked'] = false;
276 // {{{ public function getTopicDetail()
279 * get topic detail info
281 * @param string $topicName
284 public function getTopicDetail($topicName)
286 return $this->metadata->getTopicDetail($topicName);