diff --git a/Kafka/Consumer/State.php b/Kafka/Consumer/State.php index 6b77dc8c..a61371f3 100644 --- a/Kafka/Consumer/State.php +++ b/Kafka/Consumer/State.php @@ -14,444 +14,446 @@ namespace Kafka\Consumer; -/** -+------------------------------------------------------------------------------ -* Kafka protocol since Kafka v0.8 -+------------------------------------------------------------------------------ -* -* @package -* @version $_SWANBR_VERSION_$ -* @copyright Copyleft -* @author $_SWANBR_AUTHOR_$ -+------------------------------------------------------------------------------ -*/ +use Swoole\Timer; +/** + * +------------------------------------------------------------------------------ + * Kafka protocol since Kafka v0.8 + * +------------------------------------------------------------------------------ + * + * @package + * @version $_SWANBR_VERSION_$ + * @copyright Copyleft + * @author $_SWANBR_AUTHOR_$ + * +------------------------------------------------------------------------------ + */ class State { - use \Kafka\SingletonTrait; - // {{{ consts + use \Kafka\SingletonTrait; - const REQUEST_METADATA = 1; - const REQUEST_GETGROUP = 2; - const REQUEST_JOINGROUP = 3; - const REQUEST_SYNCGROUP = 4; - const REQUEST_HEARTGROUP = 5; - const REQUEST_OFFSET = 6; - const REQUEST_FETCH = 7; - const REQUEST_FETCH_OFFSET = 8; - const REQUEST_COMMIT_OFFSET = 9; + // {{{ consts - const STATUS_INIT = 0; - const STATUS_STOP = 1; - const STATUS_START = 2; - const STATUS_LOOP = 4; - const STATUS_PROCESS = 8; - const STATUS_FINISH = 16; + const REQUEST_METADATA = 1; + const REQUEST_GETGROUP = 2; + const REQUEST_JOINGROUP = 3; + const REQUEST_SYNCGROUP = 4; + const REQUEST_HEARTGROUP = 5; + const REQUEST_OFFSET = 6; + const REQUEST_FETCH = 7; + const REQUEST_FETCH_OFFSET = 8; + const REQUEST_COMMIT_OFFSET = 9; - // }}} - // {{{ members - - private $callStatus = array(); - - private $requests = array( - self::REQUEST_METADATA => array(), - self::REQUEST_GETGROUP => array(), - self::REQUEST_JOINGROUP => array(), - self::REQUEST_SYNCGROUP => array(), - self::REQUEST_HEARTGROUP => array(), - self::REQUEST_OFFSET => array( - 'interval' => 2000, - ), - self::REQUEST_FETCH => array( - 'interval' => 100, - ), - self::REQUEST_FETCH_OFFSET => array( - 'interval' => 2000, - ), - self::REQUEST_COMMIT_OFFSET => array( - 'norepeat' => true, - ), - ); + const STATUS_INIT = 0; + const STATUS_STOP = 1; + const STATUS_START = 2; + const STATUS_LOOP = 4; + const STATUS_PROCESS = 8; + const STATUS_FINISH = 16; - // }}} - // {{{ functions - // {{{ public function init() + // }}} + // {{{ members - public function init() - { - $this->callStatus = array( - self::REQUEST_METADATA => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_GETGROUP => array( - 'status'=> self::STATUS_START, - ), - self::REQUEST_JOINGROUP => array( - 'status'=> self::STATUS_START, - ), - self::REQUEST_SYNCGROUP => array( - 'status'=> self::STATUS_START, - ), - self::REQUEST_HEARTGROUP => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_OFFSET => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_FETCH => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_FETCH_OFFSET => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_COMMIT_OFFSET => array( - 'status'=> self::STATUS_LOOP, - ), - ); + private $callStatus = array(); - // instances clear + private $requests = array( + self::REQUEST_METADATA => array(), + self::REQUEST_GETGROUP => array(), + self::REQUEST_JOINGROUP => array(), + self::REQUEST_SYNCGROUP => array(), + self::REQUEST_HEARTGROUP => array(), + self::REQUEST_OFFSET => array( + 'interval' => 2000, + ), + self::REQUEST_FETCH => array( + 'interval' => 100, + ), + self::REQUEST_FETCH_OFFSET => array( + 'interval' => 2000, + ), + self::REQUEST_COMMIT_OFFSET => array( + 'norepeat' => true, + ), + ); - // init requests - $config = \Kafka\ConsumerConfig::getInstance(); - foreach ($this->requests as $request => $option) { - switch ($request) { - case self::REQUEST_METADATA: - $this->requests[$request]['interval'] = $config->getMetadataRefreshIntervalMs(); - break; - default: - $this->requests[$request]['interval'] = 1000; - } - } - } + // }}} + // {{{ functions + // {{{ public function init() - // }}} - // {{{ public function start() + public function init() + { + $this->callStatus = array( + self::REQUEST_METADATA => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_GETGROUP => array( + 'status' => self::STATUS_START, + ), + self::REQUEST_JOINGROUP => array( + 'status' => self::STATUS_START, + ), + self::REQUEST_SYNCGROUP => array( + 'status' => self::STATUS_START, + ), + self::REQUEST_HEARTGROUP => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_OFFSET => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_FETCH => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_FETCH_OFFSET => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_COMMIT_OFFSET => array( + 'status' => self::STATUS_LOOP, + ), + ); - public function start() - { - foreach ($this->requests as $request => $option) { - if (isset($option['norepeat']) && $option['norepeat']) { - continue; - } - $interval = isset($option['interval']) ? $option['interval'] : 200; - \Amp\repeat(function ($watcherId) use ($request, $option) { - if ($this->checkRun($request) && $option['func'] != null) { - $context = call_user_func($option['func']); - $this->processing($request, $context); - } - $this->requests[$request]['watcher'] = $watcherId; - }, $msInterval = $interval); - } + // instances clear - // start sync metadata - if (isset($this->requests[self::REQUEST_METADATA]['func'])) { - $context = call_user_func($this->requests[self::REQUEST_METADATA]['func']); - $this->processing($request, $context); - } - \Amp\repeat(function ($watcherId) { - $this->report(); - }, $msInterval = 1000); - } + // init requests + $config = \Kafka\ConsumerConfig::getInstance(); + foreach ($this->requests as $request => $option) { + switch ($request) { + case self::REQUEST_METADATA: + $this->requests[$request]['interval'] = $config->getMetadataRefreshIntervalMs(); + break; + default: + $this->requests[$request]['interval'] = 1000; + } + } + } - // }}} - // {{{ public function succRun() + // }}} + // {{{ public function start() - public function succRun($key, $context = null) - { - if (!isset($this->callStatus[$key])) { - return false; - } + public function start() + { + foreach ($this->requests as $request => $option) { + if (isset($option['norepeat']) && $option['norepeat']) { + continue; + } + $interval = isset($option['interval']) ? $option['interval'] : 200; + Timer::tick($msInterval = $interval, function ($watcherId) use ($request, $option) { + if ($this->checkRun($request) && $option['func'] != null) { + $context = call_user_func($option['func']); + $this->processing($request, $context); + } + $this->requests[$request]['watcher'] = $watcherId; + }); + } - switch ($key) { - case self::REQUEST_METADATA: - $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); - if ($context) { // if kafka broker is change - $this->recover(); - } - break; - case self::REQUEST_GETGROUP: - case self::REQUEST_JOINGROUP: - case self::REQUEST_SYNCGROUP: - $this->callStatus[$key]['status'] = (self::STATUS_STOP | self::STATUS_FINISH); - break; - case self::REQUEST_HEARTGROUP: - case self::REQUEST_FETCH_OFFSET: - case self::REQUEST_COMMIT_OFFSET: - $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); - break; - case self::REQUEST_OFFSET: - if (!isset($this->callStatus[$key]['context'])) { - $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); - break; - } - unset($this->callStatus[$key]['context'][$context]); - $contextStatus = $this->callStatus[$key]['context']; - if (empty($contextStatus)) { - $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); - } - break; - case self::REQUEST_FETCH: - if (!isset($this->callStatus[$key]['context'])) { - $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); - break; - } - unset($this->callStatus[$key]['context'][$context]); - $contextStatus = $this->callStatus[$key]['context']; - if (empty($contextStatus)) { - $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); - call_user_func($this->requests[self::REQUEST_COMMIT_OFFSET]['func']); - } - break; - } - } + // start sync metadata + if (isset($this->requests[self::REQUEST_METADATA]['func'])) { + $context = call_user_func($this->requests[self::REQUEST_METADATA]['func']); + $this->processing($request, $context); + } + Timer::tick($msInterval = 1000, function ($watcherId) { + $this->report(); + }); + } - // }}} - // {{{ public function failRun() + // }}} + // {{{ public function succRun() - public function failRun($key, $context = null) - { - if (!isset($this->callStatus[$key])) { - return false; - } + public function succRun($key, $context = null) + { + if (!isset($this->callStatus[$key])) { + return false; + } - switch ($key) { - case self::REQUEST_METADATA: - $this->callStatus[$key]['status'] = self::STATUS_LOOP; - break; - case self::REQUEST_GETGROUP: - case self::REQUEST_JOINGROUP: - case self::REQUEST_SYNCGROUP: - $this->recover(); - break; - } - } + switch ($key) { + case self::REQUEST_METADATA: + $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); + if ($context) { // if kafka broker is change + $this->recover(); + } + break; + case self::REQUEST_GETGROUP: + case self::REQUEST_JOINGROUP: + case self::REQUEST_SYNCGROUP: + $this->callStatus[$key]['status'] = (self::STATUS_STOP | self::STATUS_FINISH); + break; + case self::REQUEST_HEARTGROUP: + case self::REQUEST_FETCH_OFFSET: + case self::REQUEST_COMMIT_OFFSET: + $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); + break; + case self::REQUEST_OFFSET: + if (!isset($this->callStatus[$key]['context'])) { + $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); + break; + } + unset($this->callStatus[$key]['context'][$context]); + $contextStatus = $this->callStatus[$key]['context']; + if (empty($contextStatus)) { + $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); + } + break; + case self::REQUEST_FETCH: + if (!isset($this->callStatus[$key]['context'])) { + $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); + break; + } + unset($this->callStatus[$key]['context'][$context]); + $contextStatus = $this->callStatus[$key]['context']; + if (empty($contextStatus)) { + $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); + call_user_func($this->requests[self::REQUEST_COMMIT_OFFSET]['func']); + } + break; + } + } - // }}} - // {{{ public function setCallback() + // }}} + // {{{ public function failRun() - public function setCallback($callbacks) - { - foreach ($callbacks as $request => $callback) { - $this->requests[$request]['func'] = $callback; - } - } + public function failRun($key, $context = null) + { + if (!isset($this->callStatus[$key])) { + return false; + } - // }}} - // {{{ public function rejoin() + switch ($key) { + case self::REQUEST_METADATA: + $this->callStatus[$key]['status'] = self::STATUS_LOOP; + break; + case self::REQUEST_GETGROUP: + case self::REQUEST_JOINGROUP: + case self::REQUEST_SYNCGROUP: + $this->recover(); + break; + } + } - public function rejoin() - { - $joinGroupStatus = $this->callStatus[self::REQUEST_JOINGROUP]['status']; - if (($joinGroupStatus & self::STATUS_PROCESS) == self::STATUS_PROCESS) { - return; - } + // }}} + // {{{ public function setCallback() - $this->callStatus = array( - self::REQUEST_METADATA => $this->callStatus[self::REQUEST_METADATA], - self::REQUEST_GETGROUP => $this->callStatus[self::REQUEST_GETGROUP], - self::REQUEST_JOINGROUP => array( - 'status'=> self::STATUS_START, - ), - self::REQUEST_SYNCGROUP => array( - 'status'=> self::STATUS_START, - ), - self::REQUEST_HEARTGROUP => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_OFFSET => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_FETCH => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_FETCH_OFFSET => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_COMMIT_OFFSET => array( - 'status'=> self::STATUS_LOOP, - ), - ); - } + public function setCallback($callbacks) + { + foreach ($callbacks as $request => $callback) { + $this->requests[$request]['func'] = $callback; + } + } - // }}} - // {{{ public function recover() + // }}} + // {{{ public function rejoin() - public function recover() - { - $this->callStatus = array( - self::REQUEST_METADATA => $this->callStatus[self::REQUEST_METADATA], - self::REQUEST_GETGROUP => array( - 'status'=> self::STATUS_START, - ), - self::REQUEST_JOINGROUP => array( - 'status'=> self::STATUS_START, - ), - self::REQUEST_SYNCGROUP => array( - 'status'=> self::STATUS_START, - ), - self::REQUEST_HEARTGROUP => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_OFFSET => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_FETCH => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_FETCH_OFFSET => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_COMMIT_OFFSET => array( - 'status'=> self::STATUS_LOOP, - ), - ); - } + public function rejoin() + { + $joinGroupStatus = $this->callStatus[self::REQUEST_JOINGROUP]['status']; + if (($joinGroupStatus & self::STATUS_PROCESS) == self::STATUS_PROCESS) { + return; + } - // }}} - // {{{ protected function checkRun() + $this->callStatus = array( + self::REQUEST_METADATA => $this->callStatus[self::REQUEST_METADATA], + self::REQUEST_GETGROUP => $this->callStatus[self::REQUEST_GETGROUP], + self::REQUEST_JOINGROUP => array( + 'status' => self::STATUS_START, + ), + self::REQUEST_SYNCGROUP => array( + 'status' => self::STATUS_START, + ), + self::REQUEST_HEARTGROUP => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_OFFSET => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_FETCH => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_FETCH_OFFSET => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_COMMIT_OFFSET => array( + 'status' => self::STATUS_LOOP, + ), + ); + } - protected function checkRun($key) - { - if (!isset($this->callStatus[$key])) { - return false; - } + // }}} + // {{{ public function recover() - $status = $this->callStatus[$key]['status']; - switch ($key) { - case self::REQUEST_METADATA: - if ($status & self::STATUS_PROCESS == self::STATUS_PROCESS) { - return false; - } - if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { - return true; - } - return false; - case self::REQUEST_GETGROUP: - if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { - return false; - } - $metaStatus = $this->callStatus[self::REQUEST_METADATA]['status']; - if (($metaStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { - return false; - } - if (($status & self::STATUS_START) == self::STATUS_START) { - return true; - } - return false; - case self::REQUEST_JOINGROUP: - if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { - return false; - } - $groupStatus = $this->callStatus[self::REQUEST_GETGROUP]['status']; - if (($groupStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { - return false; - } - if (($status & self::STATUS_START) == self::STATUS_START) { - return true; - } - return false; - case self::REQUEST_SYNCGROUP: - if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { - return false; - } - $joinStatus = $this->callStatus[self::REQUEST_JOINGROUP]['status']; - if (($joinStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { - return false; - } - if (($status & self::STATUS_START) == self::STATUS_START) { - return true; - } - return false; - case self::REQUEST_HEARTGROUP: - case self::REQUEST_OFFSET: - if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { - return false; - } - $syncStatus = $this->callStatus[self::REQUEST_SYNCGROUP]['status']; - if (($syncStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { - return false; - } - if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { - return true; - } - return false; - case self::REQUEST_FETCH_OFFSET: - if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { - return false; - } - $syncStatus = $this->callStatus[self::REQUEST_SYNCGROUP]['status']; - if (($syncStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { - return false; - } - $offsetStatus = $this->callStatus[self::REQUEST_OFFSET]['status']; - if (($offsetStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { - return false; - } - if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { - return true; - } - return false; - case self::REQUEST_FETCH: - if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { - return false; - } - $fetchOffsetStatus = $this->callStatus[self::REQUEST_FETCH_OFFSET]['status']; - if (($fetchOffsetStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { - return false; - } - $commitOffsetStatus = $this->callStatus[self::REQUEST_COMMIT_OFFSET]['status']; - if (($commitOffsetStatus & self::STATUS_PROCESS) == self::STATUS_PROCESS) { - return false; - } - if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { - return true; - } - return false; - } - } + public function recover() + { + $this->callStatus = array( + self::REQUEST_METADATA => $this->callStatus[self::REQUEST_METADATA], + self::REQUEST_GETGROUP => array( + 'status' => self::STATUS_START, + ), + self::REQUEST_JOINGROUP => array( + 'status' => self::STATUS_START, + ), + self::REQUEST_SYNCGROUP => array( + 'status' => self::STATUS_START, + ), + self::REQUEST_HEARTGROUP => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_OFFSET => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_FETCH => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_FETCH_OFFSET => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_COMMIT_OFFSET => array( + 'status' => self::STATUS_LOOP, + ), + ); + } - // }}} - // {{{ protected function processing() + // }}} + // {{{ protected function checkRun() - protected function processing($key, $context) - { - if (!isset($this->callStatus[$key])) { - return false; - } + protected function checkRun($key) + { + if (!isset($this->callStatus[$key])) { + return false; + } - // set process start time - $this->callStatus[$key]['time'] = microtime(true); - switch ($key) { - case self::REQUEST_METADATA: - case self::REQUEST_GETGROUP: - case self::REQUEST_JOINGROUP: - case self::REQUEST_SYNCGROUP: - case self::REQUEST_HEARTGROUP: - case self::REQUEST_FETCH_OFFSET: - case self::REQUEST_COMMIT_OFFSET: - $this->callStatus[$key]['status'] |= self::STATUS_PROCESS; - break; - case self::REQUEST_OFFSET: - case self::REQUEST_FETCH: - $this->callStatus[$key]['status'] |= self::STATUS_PROCESS; - $contextStatus = array(); - foreach ($context as $fd) { - $contextStatus[$fd] = self::STATUS_PROCESS; - } - $this->callStatus[$key]['context'] = $contextStatus; - break; + $status = $this->callStatus[$key]['status']; + switch ($key) { + case self::REQUEST_METADATA: + if ($status & self::STATUS_PROCESS == self::STATUS_PROCESS) { + return false; + } + if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { + return true; + } + return false; + case self::REQUEST_GETGROUP: + if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { + return false; + } + $metaStatus = $this->callStatus[self::REQUEST_METADATA]['status']; + if (($metaStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { + return false; + } + if (($status & self::STATUS_START) == self::STATUS_START) { + return true; + } + return false; + case self::REQUEST_JOINGROUP: + if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { + return false; + } + $groupStatus = $this->callStatus[self::REQUEST_GETGROUP]['status']; + if (($groupStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { + return false; + } + if (($status & self::STATUS_START) == self::STATUS_START) { + return true; + } + return false; + case self::REQUEST_SYNCGROUP: + if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { + return false; + } + $joinStatus = $this->callStatus[self::REQUEST_JOINGROUP]['status']; + if (($joinStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { + return false; + } + if (($status & self::STATUS_START) == self::STATUS_START) { + return true; + } + return false; + case self::REQUEST_HEARTGROUP: + case self::REQUEST_OFFSET: + if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { + return false; + } + $syncStatus = $this->callStatus[self::REQUEST_SYNCGROUP]['status']; + if (($syncStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { + return false; + } + if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { + return true; + } + return false; + case self::REQUEST_FETCH_OFFSET: + if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { + return false; + } + $syncStatus = $this->callStatus[self::REQUEST_SYNCGROUP]['status']; + if (($syncStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { + return false; + } + $offsetStatus = $this->callStatus[self::REQUEST_OFFSET]['status']; + if (($offsetStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { + return false; + } + if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { + return true; + } + return false; + case self::REQUEST_FETCH: + if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { + return false; + } + $fetchOffsetStatus = $this->callStatus[self::REQUEST_FETCH_OFFSET]['status']; + if (($fetchOffsetStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { + return false; + } + $commitOffsetStatus = $this->callStatus[self::REQUEST_COMMIT_OFFSET]['status']; + if (($commitOffsetStatus & self::STATUS_PROCESS) == self::STATUS_PROCESS) { + return false; + } + if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { + return true; + } + return false; + } + } - } - } + // }}} + // {{{ protected function processing() - // }}} - // {{{ protected function report() + protected function processing($key, $context) + { + if (!isset($this->callStatus[$key])) { + return false; + } - protected function report() - { - //var_dump($this->callStatus[self::REQUEST_COMMIT_OFFSET]); - } + // set process start time + $this->callStatus[$key]['time'] = microtime(true); + switch ($key) { + case self::REQUEST_METADATA: + case self::REQUEST_GETGROUP: + case self::REQUEST_JOINGROUP: + case self::REQUEST_SYNCGROUP: + case self::REQUEST_HEARTGROUP: + case self::REQUEST_FETCH_OFFSET: + case self::REQUEST_COMMIT_OFFSET: + $this->callStatus[$key]['status'] |= self::STATUS_PROCESS; + break; + case self::REQUEST_OFFSET: + case self::REQUEST_FETCH: + $this->callStatus[$key]['status'] |= self::STATUS_PROCESS; + $contextStatus = array(); + foreach ($context as $fd) { + $contextStatus[$fd] = self::STATUS_PROCESS; + } + $this->callStatus[$key]['context'] = $contextStatus; + break; - // }}} - // }}} + } + } + + // }}} + // {{{ protected function report() + + protected function report() + { + //var_dump($this->callStatus[self::REQUEST_COMMIT_OFFSET]); + } + + // }}} + // }}} } diff --git a/Kafka/Producer/Process.php b/Kafka/Producer/Process.php index 2e1f16a7..42500a21 100644 --- a/Kafka/Producer/Process.php +++ b/Kafka/Producer/Process.php @@ -14,371 +14,372 @@ namespace Kafka\Producer; -/** -+------------------------------------------------------------------------------ -* Kafka protocol since Kafka v0.8 -+------------------------------------------------------------------------------ -* -* @package -* @version $_SWANBR_VERSION_$ -* @copyright Copyleft -* @author $_SWANBR_AUTHOR_$ -+------------------------------------------------------------------------------ -*/ +use Swoole\Timer; +/** + * +------------------------------------------------------------------------------ + * Kafka protocol since Kafka v0.8 + * +------------------------------------------------------------------------------ + * + * @package + * @version $_SWANBR_VERSION_$ + * @copyright Copyleft + * @author $_SWANBR_AUTHOR_$ + * +------------------------------------------------------------------------------ + */ class Process { - use \Psr\Log\LoggerAwareTrait; - use \Kafka\LoggerTrait; + use \Psr\Log\LoggerAwareTrait; + use \Kafka\LoggerTrait; - // {{{ consts - // }}} - // {{{ members + // {{{ consts + // }}} + // {{{ members - protected $producer = null; + protected $producer = null; - protected $isRunning = true; + protected $isRunning = true; - protected $success = null; + protected $success = null; - protected $error = null; + protected $error = null; - // }}} - // {{{ functions - // {{{ public function __construct() + // }}} + // {{{ functions + // {{{ public function __construct() - public function __construct(\Closure $producer = null) - { - $this->producer = $producer; - } + public function __construct(\Closure $producer = null) + { + $this->producer = $producer; + } - // }}} - // {{{ public function init() + // }}} + // {{{ public function init() - /** - * start consumer - * - * @access public - * @return void - */ - public function init() - { - // init protocol - $config = \Kafka\ProducerConfig::getInstance(); - \Kafka\Protocol::init($config->getBrokerVersion(), $this->logger); + /** + * start consumer + * + * @access public + * @return void + */ + public function init() + { + // init protocol + $config = \Kafka\ProducerConfig::getInstance(); + \Kafka\Protocol::init($config->getBrokerVersion(), $this->logger); - // init process request - $broker = \Kafka\Broker::getInstance(); - $broker->setProcess(function ($data, $fd) { - $this->processRequest($data, $fd); - }); + // init process request + $broker = \Kafka\Broker::getInstance(); + $broker->setProcess(function ($data, $fd) { + $this->processRequest($data, $fd); + }); - // init state - $this->state = \Kafka\Producer\State::getInstance(); - if ($this->logger) { - $this->state->setLogger($this->logger); - } - $this->state->setCallback(array( - \Kafka\Producer\State::REQUEST_METADATA => function () { - return $this->syncMeta(); - }, - \Kafka\Producer\State::REQUEST_PRODUCE => function () { - return $this->produce(); - }, - )); - $this->state->init(); + // init state + $this->state = \Kafka\Producer\State::getInstance(); + if ($this->logger) { + $this->state->setLogger($this->logger); + } + $this->state->setCallback(array( + \Kafka\Producer\State::REQUEST_METADATA => function () { + return $this->syncMeta(); + }, + \Kafka\Producer\State::REQUEST_PRODUCE => function () { + return $this->produce(); + }, + )); + $this->state->init(); - $topics = $broker->getTopics(); - if (!empty($topics)) { - $this->state->succRun(\Kafka\Producer\State::REQUEST_METADATA); - } - } + $topics = $broker->getTopics(); + if (!empty($topics)) { + $this->state->succRun(\Kafka\Producer\State::REQUEST_METADATA); + } + } - // }}} - // {{{ public function start() + // }}} + // {{{ public function start() - /** - * start consumer - * - * @access public - * @return void - */ - public function start() - { - $this->init(); - $this->state->start(); - $config = \Kafka\ProducerConfig::getInstance(); - $isAsyn = $config->getIsAsyn(); - if (!$isAsyn) { - \Amp\repeat(function ($watcherId) { - if ($this->error) { - call_user_func($this->error, 1000); - } - \Amp\cancel($watcherId); - \Amp\stop(); - }, $msInterval = $config->getRequestTimeout()); - }; - } + /** + * start consumer + * + * @access public + * @return void + */ + public function start() + { + $this->init(); + $this->state->start(); + $config = \Kafka\ProducerConfig::getInstance(); + $isAsyn = $config->getIsAsyn(); + if (!$isAsyn) { + Timer::tick($msInterval = $config->getRequestTimeout(), function ($watcherId) { + if ($this->error) { + call_user_func($this->error, 1000); + } + \Amp\cancel($watcherId); + \Amp\stop(); + }); + }; + } - // }}} - // {{{ public function stop() + // }}} + // {{{ public function stop() - /** - * stop consumer - * - * @access public - * @return void - */ - public function stop() - { - $this->isRunning = false; - } + /** + * stop consumer + * + * @access public + * @return void + */ + public function stop() + { + $this->isRunning = false; + } - // }}} - // {{{ public function setSuccess() + // }}} + // {{{ public function setSuccess() - /** - * set success callback - * - * @access public - * @return void - */ - public function setSuccess($success) - { - $this->success = $success; - } + /** + * set success callback + * + * @access public + * @return void + */ + public function setSuccess($success) + { + $this->success = $success; + } - // }}} - // {{{ public function setError() + // }}} + // {{{ public function setError() - /** - * set error callback - * - * @access public - * @return void - */ - public function setError($error) - { - $this->error = $error; - } + /** + * set error callback + * + * @access public + * @return void + */ + public function setError($error) + { + $this->error = $error; + } - // }}} - // {{{ public function syncMeta() + // }}} + // {{{ public function syncMeta() - public function syncMeta() - { - $this->debug('Start sync metadata request'); - $brokerList = explode(',', \Kafka\ProducerConfig::getInstance()->getMetadataBrokerList()); - $brokerHost = array(); - foreach ($brokerList as $key => $val) { - if (trim($val)) { - $brokerHost[] = $val; - } - } - if (count($brokerHost) == 0) { - throw new \Kafka\Exception('Not set config `metadataBrokerList`'); - } - shuffle($brokerHost); - $broker = \Kafka\Broker::getInstance(); - foreach ($brokerHost as $host) { - $socket = $broker->getMetaConnect($host); - if ($socket) { - $params = array(); - $this->debug('Start sync metadata request params:' . json_encode($params)); - $requestData = \Kafka\Protocol::encode(\Kafka\Protocol::METADATA_REQUEST, $params); - $socket->write($requestData); - return; - } - } - throw new \Kafka\Exception('Not has broker can connection `metadataBrokerList`'); - } + public function syncMeta() + { + $this->debug('Start sync metadata request'); + $brokerList = explode(',', \Kafka\ProducerConfig::getInstance()->getMetadataBrokerList()); + $brokerHost = array(); + foreach ($brokerList as $key => $val) { + if (trim($val)) { + $brokerHost[] = $val; + } + } + if (count($brokerHost) == 0) { + throw new \Kafka\Exception('Not set config `metadataBrokerList`'); + } + shuffle($brokerHost); + $broker = \Kafka\Broker::getInstance(); + foreach ($brokerHost as $host) { + $socket = $broker->getMetaConnect($host); + if ($socket) { + $params = array(); + $this->debug('Start sync metadata request params:' . json_encode($params)); + $requestData = \Kafka\Protocol::encode(\Kafka\Protocol::METADATA_REQUEST, $params); + $socket->write($requestData); + return; + } + } + throw new \Kafka\Exception('Not has broker can connection `metadataBrokerList`'); + } - // }}} - // {{{ protected function processRequest() + // }}} + // {{{ protected function processRequest() - /** - * process Request - * - * @access public - * @return void - */ - protected function processRequest($data, $fd) - { - $correlationId = \Kafka\Protocol\Protocol::unpack(\Kafka\Protocol\Protocol::BIT_B32, substr($data, 0, 4)); - switch ($correlationId) { - case \Kafka\Protocol::METADATA_REQUEST: - $result = \Kafka\Protocol::decode(\Kafka\Protocol::METADATA_REQUEST, substr($data, 4)); - if (!isset($result['brokers']) || !isset($result['topics'])) { - $this->error('Get metadata is fail, brokers or topics is null.'); - $this->state->failRun(\Kafka\Producer\State::REQUEST_METADATA); - } else { - $broker = \Kafka\Broker::getInstance(); - $isChange = $broker->setData($result['topics'], $result['brokers']); - $this->state->succRun(\Kafka\Producer\State::REQUEST_METADATA, $isChange); - } - break; - case \Kafka\Protocol::PRODUCE_REQUEST: - $result = \Kafka\Protocol::decode(\Kafka\Protocol::PRODUCE_REQUEST, substr($data, 4)); - $this->succProduce($result, $fd); - break; - default: - $this->error('Error request, correlationId:' . $correlationId); - } - } + /** + * process Request + * + * @access public + * @return void + */ + protected function processRequest($data, $fd) + { + $correlationId = \Kafka\Protocol\Protocol::unpack(\Kafka\Protocol\Protocol::BIT_B32, substr($data, 0, 4)); + switch ($correlationId) { + case \Kafka\Protocol::METADATA_REQUEST: + $result = \Kafka\Protocol::decode(\Kafka\Protocol::METADATA_REQUEST, substr($data, 4)); + if (!isset($result['brokers']) || !isset($result['topics'])) { + $this->error('Get metadata is fail, brokers or topics is null.'); + $this->state->failRun(\Kafka\Producer\State::REQUEST_METADATA); + } else { + $broker = \Kafka\Broker::getInstance(); + $isChange = $broker->setData($result['topics'], $result['brokers']); + $this->state->succRun(\Kafka\Producer\State::REQUEST_METADATA, $isChange); + } + break; + case \Kafka\Protocol::PRODUCE_REQUEST: + $result = \Kafka\Protocol::decode(\Kafka\Protocol::PRODUCE_REQUEST, substr($data, 4)); + $this->succProduce($result, $fd); + break; + default: + $this->error('Error request, correlationId:' . $correlationId); + } + } - // }}} - // {{{ protected function produce() + // }}} + // {{{ protected function produce() - protected function produce() - { - $context = array(); - $broker = \Kafka\Broker::getInstance(); - $requiredAck = \Kafka\ProducerConfig::getInstance()->getRequiredAck(); - $timeout = \Kafka\ProducerConfig::getInstance()->getTimeout(); + protected function produce() + { + $context = array(); + $broker = \Kafka\Broker::getInstance(); + $requiredAck = \Kafka\ProducerConfig::getInstance()->getRequiredAck(); + $timeout = \Kafka\ProducerConfig::getInstance()->getTimeout(); - // get send message - // data struct - // topic: - // partId: - // key: - // value: - $data = call_user_func($this->producer); - if (empty($data)) { - return $context; - } + // get send message + // data struct + // topic: + // partId: + // key: + // value: + $data = call_user_func($this->producer); + if (empty($data)) { + return $context; + } - $sendData = $this->convertMessage($data); - foreach ($sendData as $brokerId => $topicList) { - $connect = $broker->getDataConnect($brokerId); - if (!$connect) { - return; - } + $sendData = $this->convertMessage($data); + foreach ($sendData as $brokerId => $topicList) { + $connect = $broker->getDataConnect($brokerId); + if (!$connect) { + return; + } - $requiredAck = \Kafka\ProducerConfig::getInstance()->getRequiredAck(); - $params = array( - 'required_ack' => $requiredAck, - 'timeout' => \Kafka\ProducerConfig::getInstance()->getTimeout(), - 'data' => $topicList, - ); - $this->debug("Send message start, params:" . json_encode($params)); - $requestData = \Kafka\Protocol::encode(\Kafka\Protocol::PRODUCE_REQUEST, $params); - if ($requiredAck == 0) { // If it is 0 the server will not send any response - $this->state->succRun(\Kafka\Producer\State::REQUEST_PRODUCE); - } else { - $connect->write($requestData); - $context[] = (int)$connect->getSocket(); - } - } + $requiredAck = \Kafka\ProducerConfig::getInstance()->getRequiredAck(); + $params = array( + 'required_ack' => $requiredAck, + 'timeout' => \Kafka\ProducerConfig::getInstance()->getTimeout(), + 'data' => $topicList, + ); + $this->debug("Send message start, params:" . json_encode($params)); + $requestData = \Kafka\Protocol::encode(\Kafka\Protocol::PRODUCE_REQUEST, $params); + if ($requiredAck == 0) { // If it is 0 the server will not send any response + $this->state->succRun(\Kafka\Producer\State::REQUEST_PRODUCE); + } else { + $connect->write($requestData); + $context[] = (int)$connect->getSocket(); + } + } - return $context; - } + return $context; + } - // }}} - // {{{ protected function succProduce() + // }}} + // {{{ protected function succProduce() - protected function succProduce($result, $fd) - { - $msg = sprintf('Send message sucess, result: %s', json_encode($result)); - $this->debug($msg); - if ($this->success) { - call_user_func($this->success, $result); - } - $this->state->succRun(\Kafka\Producer\State::REQUEST_PRODUCE, $fd); - } + protected function succProduce($result, $fd) + { + $msg = sprintf('Send message sucess, result: %s', json_encode($result)); + $this->debug($msg); + if ($this->success) { + call_user_func($this->success, $result); + } + $this->state->succRun(\Kafka\Producer\State::REQUEST_PRODUCE, $fd); + } - // }}} - // {{{ protected function stateConvert() + // }}} + // {{{ protected function stateConvert() - protected function stateConvert($errorCode, $context = null) - { - $retry = false; - $this->error(\Kafka\Protocol::getError($errorCode)); - if ($this->error) { - call_user_func($this->error, $errorCode); - } - $recoverCodes = array( - \Kafka\Protocol::UNKNOWN_TOPIC_OR_PARTITION, - \Kafka\Protocol::INVALID_REQUIRED_ACKS, - \Kafka\Protocol::RECORD_LIST_TOO_LARGE, - \Kafka\Protocol::NOT_ENOUGH_REPLICAS_AFTER_APPEND, - \Kafka\Protocol::NOT_ENOUGH_REPLICAS, - \Kafka\Protocol::NOT_LEADER_FOR_PARTITION, - \Kafka\Protocol::BROKER_NOT_AVAILABLE, - \Kafka\Protocol::GROUP_LOAD_IN_PROGRESS, - \Kafka\Protocol::GROUP_COORDINATOR_NOT_AVAILABLE, - \Kafka\Protocol::NOT_COORDINATOR_FOR_GROUP, - \Kafka\Protocol::INVALID_TOPIC, - \Kafka\Protocol::INCONSISTENT_GROUP_PROTOCOL, - \Kafka\Protocol::INVALID_GROUP_ID, - ); - if (in_array($errorCode, $recoverCodes)) { - $this->state->recover(); - return false; - } - return true; - } + protected function stateConvert($errorCode, $context = null) + { + $retry = false; + $this->error(\Kafka\Protocol::getError($errorCode)); + if ($this->error) { + call_user_func($this->error, $errorCode); + } + $recoverCodes = array( + \Kafka\Protocol::UNKNOWN_TOPIC_OR_PARTITION, + \Kafka\Protocol::INVALID_REQUIRED_ACKS, + \Kafka\Protocol::RECORD_LIST_TOO_LARGE, + \Kafka\Protocol::NOT_ENOUGH_REPLICAS_AFTER_APPEND, + \Kafka\Protocol::NOT_ENOUGH_REPLICAS, + \Kafka\Protocol::NOT_LEADER_FOR_PARTITION, + \Kafka\Protocol::BROKER_NOT_AVAILABLE, + \Kafka\Protocol::GROUP_LOAD_IN_PROGRESS, + \Kafka\Protocol::GROUP_COORDINATOR_NOT_AVAILABLE, + \Kafka\Protocol::NOT_COORDINATOR_FOR_GROUP, + \Kafka\Protocol::INVALID_TOPIC, + \Kafka\Protocol::INCONSISTENT_GROUP_PROTOCOL, + \Kafka\Protocol::INVALID_GROUP_ID, + ); + if (in_array($errorCode, $recoverCodes)) { + $this->state->recover(); + return false; + } + return true; + } - // }}} - // {{{ protected function convertMessage() + // }}} + // {{{ protected function convertMessage() - protected function convertMessage($data) - { - $sendData = array(); - $broker = \Kafka\Broker::getInstance(); - $topicInfos = $broker->getTopics(); - foreach ($data as $value) { - if (!isset($value['topic']) || !trim($value['topic'])) { - continue; - } + protected function convertMessage($data) + { + $sendData = array(); + $broker = \Kafka\Broker::getInstance(); + $topicInfos = $broker->getTopics(); + foreach ($data as $value) { + if (!isset($value['topic']) || !trim($value['topic'])) { + continue; + } - if (!isset($topicInfos[$value['topic']])) { - continue; - } + if (!isset($topicInfos[$value['topic']])) { + continue; + } - if (!isset($value['value']) || !trim($value['value'])) { - continue; - } + if (!isset($value['value']) || !trim($value['value'])) { + continue; + } - if (!isset($value['key'])) { - $value['key'] = ''; - } + if (!isset($value['key'])) { + $value['key'] = ''; + } - $topicMeta = $topicInfos[$value['topic']]; - $partNums = array_keys($topicMeta); - shuffle($partNums); - $partId = 0; - if (!isset($value['partId']) || !isset($topicMeta[$value['partId']])) { - $partId = $partNums[0]; - } else { - $partId = $value['partId']; - } + $topicMeta = $topicInfos[$value['topic']]; + $partNums = array_keys($topicMeta); + shuffle($partNums); + $partId = 0; + if (!isset($value['partId']) || !isset($topicMeta[$value['partId']])) { + $partId = $partNums[0]; + } else { + $partId = $value['partId']; + } - $brokerId = $topicMeta[$partId]; - $topicData = array(); - if (isset($sendData[$brokerId][$value['topic']])) { - $topicData = $sendData[$brokerId][$value['topic']]; - } + $brokerId = $topicMeta[$partId]; + $topicData = array(); + if (isset($sendData[$brokerId][$value['topic']])) { + $topicData = $sendData[$brokerId][$value['topic']]; + } - $partition = array(); - if (isset($topicData['partitions'][$partId])) { - $partition = $topicData['partitions'][$partId]; - } + $partition = array(); + if (isset($topicData['partitions'][$partId])) { + $partition = $topicData['partitions'][$partId]; + } - $partition['partition_id'] = $partId; - if (trim($value['key']) != '') { - $partition['messages'][] = array('value' => $value['value'], 'key' => $value['key']); - } else { - $partition['messages'][] = $value['value']; - } - - $topicData['partitions'][$partId] = $partition; - $topicData['topic_name'] = $value['topic']; - $sendData[$brokerId][$value['topic']] = $topicData; - } + $partition['partition_id'] = $partId; + if (trim($value['key']) != '') { + $partition['messages'][] = array('value' => $value['value'], 'key' => $value['key']); + } else { + $partition['messages'][] = $value['value']; + } - return $sendData; - } + $topicData['partitions'][$partId] = $partition; + $topicData['topic_name'] = $value['topic']; + $sendData[$brokerId][$value['topic']] = $topicData; + } - // }}} - // }}} + return $sendData; + } + + // }}} + // }}} } diff --git a/Kafka/Producer/State.php b/Kafka/Producer/State.php index 5b16ee13..002b07b6 100644 --- a/Kafka/Producer/State.php +++ b/Kafka/Producer/State.php @@ -14,262 +14,264 @@ namespace Kafka\Producer; -/** -+------------------------------------------------------------------------------ -* Kafka protocol since Kafka v0.8 -+------------------------------------------------------------------------------ -* -* @package -* @version $_SWANBR_VERSION_$ -* @copyright Copyleft -* @author $_SWANBR_AUTHOR_$ -+------------------------------------------------------------------------------ -*/ +use Swoole\Timer; +/** + * +------------------------------------------------------------------------------ + * Kafka protocol since Kafka v0.8 + * +------------------------------------------------------------------------------ + * + * @package + * @version $_SWANBR_VERSION_$ + * @copyright Copyleft + * @author $_SWANBR_AUTHOR_$ + * +------------------------------------------------------------------------------ + */ class State { - use \Kafka\SingletonTrait; - // {{{ consts + use \Kafka\SingletonTrait; - const REQUEST_METADATA = 1; - const REQUEST_PRODUCE = 2; + // {{{ consts - const STATUS_INIT = 0; - const STATUS_STOP = 1; - const STATUS_START = 2; - const STATUS_LOOP = 4; - const STATUS_PROCESS = 8; - const STATUS_FINISH = 16; + const REQUEST_METADATA = 1; + const REQUEST_PRODUCE = 2; - // }}} - // {{{ members - - private $callStatus = array(); - - private $requests = array( - self::REQUEST_METADATA => array(), - self::REQUEST_PRODUCE => array(), - ); + const STATUS_INIT = 0; + const STATUS_STOP = 1; + const STATUS_START = 2; + const STATUS_LOOP = 4; + const STATUS_PROCESS = 8; + const STATUS_FINISH = 16; - // }}} - // {{{ functions - // {{{ public function init() + // }}} + // {{{ members - public function init() - { - $this->callStatus = array( - self::REQUEST_METADATA => array( - 'status'=> self::STATUS_LOOP, - ), - self::REQUEST_PRODUCE => array( - 'status'=> self::STATUS_LOOP, - ), - ); + private $callStatus = array(); - // instances clear + private $requests = array( + self::REQUEST_METADATA => array(), + self::REQUEST_PRODUCE => array(), + ); - // init requests - $config = \Kafka\ConsumerConfig::getInstance(); - foreach ($this->requests as $request => $option) { - switch ($request) { - case self::REQUEST_METADATA: - $this->requests[$request]['interval'] = $config->getMetadataRefreshIntervalMs(); - break; - default: - $isAsyn = $config->getIsAsyn(); - if ($isAsyn) { - $this->requests[$request]['interval'] = $config->getProduceInterval(); - } else { - $this->requests[$request]['interval'] = 1; - } - } - } - } + // }}} + // {{{ functions + // {{{ public function init() - // }}} - // {{{ public function start() + public function init() + { + $this->callStatus = array( + self::REQUEST_METADATA => array( + 'status' => self::STATUS_LOOP, + ), + self::REQUEST_PRODUCE => array( + 'status' => self::STATUS_LOOP, + ), + ); - public function start() - { - foreach ($this->requests as $request => $option) { - $interval = isset($option['interval']) ? $option['interval'] : 200; - \Amp\repeat(function ($watcherId) use ($request, $option) { - if ($this->checkRun($request) && $option['func'] != null) { - $context = call_user_func($option['func']); - $this->processing($request, $context); - } - $this->requests[$request]['watcher'] = $watcherId; - }, $msInterval = $interval); - } + // instances clear - // start sync metadata - if (isset($this->requests[self::REQUEST_METADATA]['func']) - && $this->callStatus[self::REQUEST_METADATA]['status'] == self::STATUS_LOOP) { - $context = call_user_func($this->requests[self::REQUEST_METADATA]['func']); - $this->processing($request, $context); - } - \Amp\repeat(function ($watcherId) { - $this->report(); - }, $msInterval = 1000); - } + // init requests + $config = \Kafka\ConsumerConfig::getInstance(); + foreach ($this->requests as $request => $option) { + switch ($request) { + case self::REQUEST_METADATA: + $this->requests[$request]['interval'] = $config->getMetadataRefreshIntervalMs(); + break; + default: + $isAsyn = $config->getIsAsyn(); + if ($isAsyn) { + $this->requests[$request]['interval'] = $config->getProduceInterval(); + } else { + $this->requests[$request]['interval'] = 1; + } + } + } + } - // }}} - // {{{ public function succRun() + // }}} + // {{{ public function start() - public function succRun($key, $context = null) - { - $config = \Kafka\ConsumerConfig::getInstance(); - $isAsyn = $config->getIsAsyn(); - if (!isset($this->callStatus[$key])) { - return false; - } + public function start() + { + foreach ($this->requests as $request => $option) { + $interval = isset($option['interval']) ? $option['interval'] : 200; + Timer::tick($msInterval = $interval, function ($watcherId) use ($request, $option) { + if ($this->checkRun($request) && $option['func'] != null) { + $context = call_user_func($option['func']); + $this->processing($request, $context); + } + $this->requests[$request]['watcher'] = $watcherId; + }); + } - switch ($key) { - case self::REQUEST_METADATA: - $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); - if ($context) { // if kafka broker is change - $this->recover(); - } - break; - case self::REQUEST_PRODUCE: - if ($context == null) { - if (!$isAsyn) { - $this->callStatus[$key]['status'] = self::STATUS_FINISH; - \Amp\stop(); - } else { - $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); - } - break; - } - unset($this->callStatus[$key]['context'][$context]); - $contextStatus = $this->callStatus[$key]['context']; - if (empty($contextStatus)) { - if (!$isAsyn) { - $this->callStatus[$key]['status'] = self::STATUS_FINISH; - \Amp\stop(); - } else { - $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); - } - } - break; - } - } + // start sync metadata + if (isset($this->requests[self::REQUEST_METADATA]['func']) + && $this->callStatus[self::REQUEST_METADATA]['status'] == self::STATUS_LOOP) { + $context = call_user_func($this->requests[self::REQUEST_METADATA]['func']); + $this->processing($request, $context); + } + Timer::tick($msInterval = 1000, function ($watcherId) { + $this->report(); + }); + } - // }}} - // {{{ public function failRun() + // }}} + // {{{ public function succRun() - public function failRun($key, $context = null) - { - if (!isset($this->callStatus[$key])) { - return false; - } + public function succRun($key, $context = null) + { + $config = \Kafka\ConsumerConfig::getInstance(); + $isAsyn = $config->getIsAsyn(); + if (!isset($this->callStatus[$key])) { + return false; + } - switch ($key) { - case self::REQUEST_METADATA: - $this->callStatus[$key]['status'] = self::STATUS_LOOP; - break; - case self::REQUEST_PRODUCE: - $this->recover(); - break; - } - } + switch ($key) { + case self::REQUEST_METADATA: + $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); + if ($context) { // if kafka broker is change + $this->recover(); + } + break; + case self::REQUEST_PRODUCE: + if ($context == null) { + if (!$isAsyn) { + $this->callStatus[$key]['status'] = self::STATUS_FINISH; + \Amp\stop(); + } else { + $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); + } + break; + } + unset($this->callStatus[$key]['context'][$context]); + $contextStatus = $this->callStatus[$key]['context']; + if (empty($contextStatus)) { + if (!$isAsyn) { + $this->callStatus[$key]['status'] = self::STATUS_FINISH; + \Amp\stop(); + } else { + $this->callStatus[$key]['status'] = (self::STATUS_LOOP | self::STATUS_FINISH); + } + } + break; + } + } - // }}} - // {{{ public function setCallback() + // }}} + // {{{ public function failRun() - public function setCallback($callbacks) - { - foreach ($callbacks as $request => $callback) { - $this->requests[$request]['func'] = $callback; - } - } + public function failRun($key, $context = null) + { + if (!isset($this->callStatus[$key])) { + return false; + } - // }}} - // {{{ public function recover() + switch ($key) { + case self::REQUEST_METADATA: + $this->callStatus[$key]['status'] = self::STATUS_LOOP; + break; + case self::REQUEST_PRODUCE: + $this->recover(); + break; + } + } - public function recover() - { - $this->callStatus = array( - self::REQUEST_METADATA => $this->callStatus[self::REQUEST_METADATA], - self::REQUEST_PRODUCE => array( - 'status'=> self::STATUS_LOOP, - ), - ); - } + // }}} + // {{{ public function setCallback() - // }}} - // {{{ protected function checkRun() + public function setCallback($callbacks) + { + foreach ($callbacks as $request => $callback) { + $this->requests[$request]['func'] = $callback; + } + } - protected function checkRun($key) - { - if (!isset($this->callStatus[$key])) { - return false; - } + // }}} + // {{{ public function recover() - $status = $this->callStatus[$key]['status']; - switch ($key) { - case self::REQUEST_METADATA: - if ($status & self::STATUS_PROCESS == self::STATUS_PROCESS) { - return false; - } - if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { - return true; - } - return false; - case self::REQUEST_PRODUCE: - if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { - return false; - } - $syncStatus = $this->callStatus[self::REQUEST_METADATA]['status']; - if (($syncStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { - return false; - } - if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { - return true; - } - return false; - } - } + public function recover() + { + $this->callStatus = array( + self::REQUEST_METADATA => $this->callStatus[self::REQUEST_METADATA], + self::REQUEST_PRODUCE => array( + 'status' => self::STATUS_LOOP, + ), + ); + } - // }}} - // {{{ protected function processing() + // }}} + // {{{ protected function checkRun() - protected function processing($key, $context) - { - if (!isset($this->callStatus[$key])) { - return false; - } + protected function checkRun($key) + { + if (!isset($this->callStatus[$key])) { + return false; + } - // set process start time - $this->callStatus[$key]['time'] = microtime(true); - switch ($key) { - case self::REQUEST_METADATA: - $this->callStatus[$key]['status'] |= self::STATUS_PROCESS; - break; - case self::REQUEST_PRODUCE: - if (empty($context)) { - break; - } - $this->callStatus[$key]['status'] |= self::STATUS_PROCESS; - $contextStatus = array(); - if (is_array($context)) { - foreach ($context as $fd) { - $contextStatus[$fd] = self::STATUS_PROCESS; - } - $this->callStatus[$key]['context'] = $contextStatus; - } - break; - } - } + $status = $this->callStatus[$key]['status']; + switch ($key) { + case self::REQUEST_METADATA: + if ($status & self::STATUS_PROCESS == self::STATUS_PROCESS) { + return false; + } + if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { + return true; + } + return false; + case self::REQUEST_PRODUCE: + if (($status & self::STATUS_PROCESS) == self::STATUS_PROCESS) { + return false; + } + $syncStatus = $this->callStatus[self::REQUEST_METADATA]['status']; + if (($syncStatus & self::STATUS_FINISH) != self::STATUS_FINISH) { + return false; + } + if (($status & self::STATUS_LOOP) == self::STATUS_LOOP) { + return true; + } + return false; + } + } - // }}} - // {{{ protected function report() + // }}} + // {{{ protected function processing() - protected function report() - { - //var_dump($this->callStatus[self::REQUEST_COMMIT_OFFSET]); - } + protected function processing($key, $context) + { + if (!isset($this->callStatus[$key])) { + return false; + } - // }}} - // }}} + // set process start time + $this->callStatus[$key]['time'] = microtime(true); + switch ($key) { + case self::REQUEST_METADATA: + $this->callStatus[$key]['status'] |= self::STATUS_PROCESS; + break; + case self::REQUEST_PRODUCE: + if (empty($context)) { + break; + } + $this->callStatus[$key]['status'] |= self::STATUS_PROCESS; + $contextStatus = array(); + if (is_array($context)) { + foreach ($context as $fd) { + $contextStatus[$fd] = self::STATUS_PROCESS; + } + $this->callStatus[$key]['context'] = $contextStatus; + } + break; + } + } + + // }}} + // {{{ protected function report() + + protected function report() + { + //var_dump($this->callStatus[self::REQUEST_COMMIT_OFFSET]); + } + + // }}} + // }}} }