as2252258/kiri-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

改名

Browse Source
This commit is contained in:
Mr·x
2020-10-09 19:21:02 +08:00
parent eebc23da5b
commit 0e722df943
1 changed files with 55 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Consumer.php
+55 -54
View File
@@ -18,66 +18,67 @@ use function Amp\run;
/** /**
* +------------------------------------------------------------------------------ * +------------------------------------------------------------------------------
* Kafka protocol since Kafka v0.8 * Kafka protocol since Kafka v0.8
* +------------------------------------------------------------------------------ * +------------------------------------------------------------------------------
* *
* @package * @package
* @version $_SWANBR_VERSION_$ * @version $_SWANBR_VERSION_$
* @copyright Copyleft * @copyright Copyleft
* @author $_SWANBR_AUTHOR_$ * @author $_SWANBR_AUTHOR_$
* +------------------------------------------------------------------------------ * +------------------------------------------------------------------------------
*/ */
class Consumer class Consumer
{ {
use \Psr\Log\LoggerAwareTrait; use \Psr\Log\LoggerAwareTrait;
use \Kafka\LoggerTrait; use \Kafka\LoggerTrait;
// {{{ consts // {{{ consts
// }}} // }}}
// {{{ members // {{{ members
private $isRunning = false;
// }}} private $isRunning = false;
// {{{ functions
// {{{ public function __construct()
/** // }}}
* __construct // {{{ functions
* // {{{ public function __construct()
* @access public
* @param $hostList
* @param null $timeout
*/
public function __construct()
{
}
// }}} /**
// {{{ public function start() * __construct
*
* @access public
* @param $hostList
* @param null $timeout
*/
public function __construct()
{
}
/** // }}}
* start consumer // {{{ public function start()
*
* @access public
* @return void
*/
public function start(\Closure $consumer = null, $isBlock = true)
{
if ($this->isRunning) {
$this->error('Has start consumer');
return;
}
$process = new \Kafka\Consumer\Process($consumer);
if ($this->logger) {
$process->setLogger($this->logger);
}
$process->start();
$this->isRunning = true;
run();
}
// }}} /**
// }}} * start consumer
*
* @access public
* @return void
*/
public function start(\Closure $consumer = null, $isBlock = true)
{
if ($this->isRunning) {
$this->error('Has start consumer');
return;
}
$process = new \Kafka\Consumer\Process($consumer);
if ($this->logger) {
$process->setLogger($this->logger);
}
$process->start();
$this->isRunning = true;
if ($isBlock) {
run();
}
}
// }}}
// }}}
} }