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