改名
This commit is contained in:
+182
-181
@@ -15,232 +15,233 @@
|
|||||||
namespace Kafka;
|
namespace Kafka;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
+------------------------------------------------------------------------------
|
* +------------------------------------------------------------------------------
|
||||||
* Kafka Broker info manager
|
* Kafka Broker info manager
|
||||||
+------------------------------------------------------------------------------
|
* +------------------------------------------------------------------------------
|
||||||
*
|
*
|
||||||
* @package
|
* @package
|
||||||
* @version $_SWANBR_VERSION_$
|
* @version $_SWANBR_VERSION_$
|
||||||
* @copyright Copyleft
|
* @copyright Copyleft
|
||||||
* @author $_SWANBR_AUTHOR_$
|
* @author $_SWANBR_AUTHOR_$
|
||||||
+------------------------------------------------------------------------------
|
* +------------------------------------------------------------------------------
|
||||||
*/
|
*/
|
||||||
|
|
||||||
class Broker
|
class Broker
|
||||||
{
|
{
|
||||||
use SingletonTrait;
|
use SingletonTrait;
|
||||||
// {{{ consts
|
|
||||||
// }}}
|
|
||||||
// {{{ members
|
|
||||||
|
|
||||||
private $groupBrokerId = null;
|
// {{{ consts
|
||||||
|
// }}}
|
||||||
|
// {{{ members
|
||||||
|
|
||||||
private $topics = array();
|
private $groupBrokerId = null;
|
||||||
|
|
||||||
private $brokers = array();
|
private $topics = array();
|
||||||
|
|
||||||
private $metaSockets = array();
|
private $brokers = array();
|
||||||
|
|
||||||
private $dataSockets = array();
|
private $metaSockets = array();
|
||||||
|
|
||||||
private $process;
|
private $dataSockets = array();
|
||||||
|
|
||||||
private $socket;
|
private $process;
|
||||||
|
|
||||||
// }}}
|
private $socket;
|
||||||
// {{{ functions
|
|
||||||
// {{{ public function setProcess()
|
|
||||||
|
|
||||||
public function setProcess(\Closure $process)
|
// }}}
|
||||||
{
|
// {{{ functions
|
||||||
$this->process = $process;
|
// {{{ public function setProcess()
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
public function setProcess(\Closure $process)
|
||||||
// {{{ public function setGroupBrokerId()
|
{
|
||||||
|
$this->process = $process;
|
||||||
|
}
|
||||||
|
|
||||||
public function setGroupBrokerId($brokerId)
|
// }}}
|
||||||
{
|
// {{{ public function setGroupBrokerId()
|
||||||
$this->groupBrokerId = $brokerId;
|
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
public function setGroupBrokerId($brokerId)
|
||||||
// {{{ public function getGroupBrokerId()
|
{
|
||||||
|
$this->groupBrokerId = $brokerId;
|
||||||
|
}
|
||||||
|
|
||||||
public function getGroupBrokerId()
|
// }}}
|
||||||
{
|
// {{{ public function getGroupBrokerId()
|
||||||
return $this->groupBrokerId;
|
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
public function getGroupBrokerId()
|
||||||
// {{{ public function setData()
|
{
|
||||||
|
return $this->groupBrokerId;
|
||||||
|
}
|
||||||
|
|
||||||
public function setData($topics, $brokersResult)
|
// }}}
|
||||||
{
|
// {{{ public function setData()
|
||||||
$brokers = array();
|
|
||||||
foreach ($brokersResult as $value) {
|
|
||||||
$key = $value['nodeId'];
|
|
||||||
$hostname = $value['host'] . ':' . $value['port'];
|
|
||||||
$brokers[$key] = $hostname;
|
|
||||||
}
|
|
||||||
|
|
||||||
$change = false;
|
public function setData($topics, $brokersResult)
|
||||||
if (serialize($this->brokers) != serialize($brokers)) {
|
{
|
||||||
$this->brokers = $brokers;
|
$brokers = array();
|
||||||
$change = true;
|
foreach ($brokersResult as $value) {
|
||||||
}
|
$key = $value['nodeId'];
|
||||||
|
$hostname = $value['host'] . ':' . $value['port'];
|
||||||
|
$brokers[$key] = $hostname;
|
||||||
|
}
|
||||||
|
|
||||||
$newTopics = array();
|
$change = false;
|
||||||
foreach ($topics as $topic) {
|
if (serialize($this->brokers) != serialize($brokers)) {
|
||||||
if ($topic['errorCode'] != \Kafka\Protocol::NO_ERROR) {
|
$this->brokers = $brokers;
|
||||||
$this->error('Parse metadata for topic is error, error:' . \Kafka\Protocol::getError($topic['errorCode']));
|
$change = true;
|
||||||
continue;
|
}
|
||||||
}
|
|
||||||
$item = array();
|
|
||||||
foreach ($topic['partitions'] as $part) {
|
|
||||||
$item[$part['partitionId']] = $part['leader'];
|
|
||||||
}
|
|
||||||
$newTopics[$topic['topicName']] = $item;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (serialize($this->topics) != serialize($newTopics)) {
|
$newTopics = array();
|
||||||
$this->topics = $newTopics;
|
foreach ($topics as $topic) {
|
||||||
$change = true;
|
if ($topic['errorCode'] != \Kafka\Protocol::NO_ERROR) {
|
||||||
}
|
$this->error('Parse metadata for topic is error, error:' . \Kafka\Protocol::getError($topic['errorCode']));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
$item = array();
|
||||||
|
foreach ($topic['partitions'] as $part) {
|
||||||
|
$item[$part['partitionId']] = $part['leader'];
|
||||||
|
}
|
||||||
|
$newTopics[$topic['topicName']] = $item;
|
||||||
|
}
|
||||||
|
|
||||||
return $change;
|
if (serialize($this->topics) != serialize($newTopics)) {
|
||||||
}
|
$this->topics = $newTopics;
|
||||||
|
$change = true;
|
||||||
|
}
|
||||||
|
|
||||||
// }}}
|
return $change;
|
||||||
// {{{ public function getTopics()
|
}
|
||||||
|
|
||||||
public function getTopics()
|
// }}}
|
||||||
{
|
// {{{ public function getTopics()
|
||||||
return $this->topics;
|
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
public function getTopics()
|
||||||
// {{{ public function getBrokers()
|
{
|
||||||
|
return $this->topics;
|
||||||
|
}
|
||||||
|
|
||||||
public function getBrokers()
|
// }}}
|
||||||
{
|
// {{{ public function getBrokers()
|
||||||
return $this->brokers;
|
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
public function getBrokers()
|
||||||
// {{{ public function getMetaConnect()
|
{
|
||||||
|
return $this->brokers;
|
||||||
|
}
|
||||||
|
|
||||||
public function getMetaConnect($key, $modeSync = false)
|
// }}}
|
||||||
{
|
// {{{ public function getMetaConnect()
|
||||||
return $this->getConnect($key, 'metaSockets', $modeSync);
|
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
public function getMetaConnect($key, $modeSync = false)
|
||||||
// {{{ public function getRandConnect()
|
{
|
||||||
|
return $this->getConnect($key, 'metaSockets', $modeSync);
|
||||||
|
}
|
||||||
|
|
||||||
public function getRandConnect($modeSync = false)
|
// }}}
|
||||||
{
|
// {{{ public function getRandConnect()
|
||||||
$nodeIds = array_keys($this->brokers);
|
|
||||||
shuffle($nodeIds);
|
|
||||||
if (!isset($nodeIds[0])) {
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
return $this->getMetaConnect($nodeIds[0], $modeSync);
|
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
public function getRandConnect($modeSync = false)
|
||||||
// {{{ public function getDataConnect()
|
{
|
||||||
|
$nodeIds = array_keys($this->brokers);
|
||||||
public function getDataConnect($key, $modeSync = false)
|
shuffle($nodeIds);
|
||||||
{
|
if (!isset($nodeIds[0])) {
|
||||||
return $this->getConnect($key, 'dataSockets', $modeSync);
|
return false;
|
||||||
}
|
}
|
||||||
|
return $this->getMetaConnect($nodeIds[0], $modeSync);
|
||||||
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ public function getConnect()
|
// {{{ public function getDataConnect()
|
||||||
|
|
||||||
public function getConnect($key, $type, $modeSync = false)
|
public function getDataConnect($key, $modeSync = false)
|
||||||
{
|
{
|
||||||
if (isset($this->{$type}[$key])) {
|
return $this->getConnect($key, 'dataSockets', $modeSync);
|
||||||
return $this->{$type}[$key];
|
}
|
||||||
}
|
|
||||||
|
|
||||||
if (isset($this->brokers[$key])) {
|
// }}}
|
||||||
$hostname = $this->brokers[$key];
|
// {{{ public function getConnect()
|
||||||
if (isset($this->{$type}[$hostname])) {
|
|
||||||
return $this->{$type}[$hostname];
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
$host = null;
|
public function getConnect($key, $type, $modeSync = false)
|
||||||
$port = null;
|
{
|
||||||
if (isset($this->brokers[$key])) {
|
if (isset($this->{$type}[$key])) {
|
||||||
$hostname = $this->brokers[$key];
|
return $this->{$type}[$key];
|
||||||
list($host, $port) = explode(':', $hostname);
|
}
|
||||||
}
|
|
||||||
|
|
||||||
if (strpos($key, ':') !== false) {
|
if (isset($this->brokers[$key])) {
|
||||||
list($host, $port) = explode(':', $key);
|
$hostname = $this->brokers[$key];
|
||||||
}
|
if (isset($this->{$type}[$hostname])) {
|
||||||
|
return $this->{$type}[$hostname];
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if ($host && $port) {
|
$host = null;
|
||||||
try {
|
$port = null;
|
||||||
$socket = $this->getSocket($host, $port, $modeSync);
|
if (isset($this->brokers[$key])) {
|
||||||
|
$hostname = $this->brokers[$key];
|
||||||
|
list($host, $port) = explode(':', $hostname);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (strpos($key, ':') !== false) {
|
||||||
|
list($host, $port) = explode(':', $key);
|
||||||
|
}
|
||||||
|
|
||||||
|
if ($host && $port) {
|
||||||
|
try {
|
||||||
|
$socket = $this->getSocket($host, $port, $modeSync);
|
||||||
// if (!$modeSync) {
|
// if (!$modeSync) {
|
||||||
// $socket->SetonReadable($this->process);
|
// $socket->SetonReadable($this->process);
|
||||||
// }
|
// }
|
||||||
$socket->connect();
|
$socket->connect();
|
||||||
$this->{$type}[$key] = $socket;
|
$this->{$type}[$key] = $socket;
|
||||||
return $socket;
|
return $socket;
|
||||||
} catch (\Exception $e) {
|
} catch (\Exception $e) {
|
||||||
$this->error($e->getMessage());
|
$this->error($e->getMessage());
|
||||||
return false;
|
$this->error($e->getFile() . ':' . $e->getLine());
|
||||||
}
|
return false;
|
||||||
} else {
|
}
|
||||||
return false;
|
} else {
|
||||||
}
|
return false;
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ public function clear()
|
// {{{ public function clear()
|
||||||
|
|
||||||
public function clear()
|
public function clear()
|
||||||
{
|
{
|
||||||
foreach ($this->metaSockets as $key => $socket) {
|
foreach ($this->metaSockets as $key => $socket) {
|
||||||
$socket->close();
|
$socket->close();
|
||||||
}
|
}
|
||||||
foreach ($this->dataSockets as $key => $socket) {
|
foreach ($this->dataSockets as $key => $socket) {
|
||||||
$socket->close();
|
$socket->close();
|
||||||
}
|
}
|
||||||
$this->brokers = [];
|
$this->brokers = [];
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ public function getSocket()
|
// {{{ public function getSocket()
|
||||||
|
|
||||||
public function getSocket($host, $port, $modeSync)
|
public function getSocket($host, $port, $modeSync)
|
||||||
{
|
{
|
||||||
if ($this->socket != null) {
|
if ($this->socket != null) {
|
||||||
return $this->socket;
|
return $this->socket;
|
||||||
}
|
}
|
||||||
|
|
||||||
if ($modeSync) {
|
if ($modeSync) {
|
||||||
$socket = new \Kafka\SocketSync($host, $port);
|
$socket = new \Kafka\SocketSync($host, $port);
|
||||||
} else {
|
} else {
|
||||||
$socket = new \Kafka\Socket($host, $port);
|
$socket = new \Kafka\Socket($host, $port);
|
||||||
}
|
}
|
||||||
return $socket;
|
return $socket;
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ public function setSocket()
|
// {{{ public function setSocket()
|
||||||
|
|
||||||
// use unit test
|
// use unit test
|
||||||
public function setSocket($socket)
|
public function setSocket($socket)
|
||||||
{
|
{
|
||||||
$this->socket = $socket;
|
$this->socket = $socket;
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// }}}
|
// }}}
|
||||||
}
|
}
|
||||||
|
|||||||
+21
-21
@@ -243,27 +243,27 @@ class Socket
|
|||||||
// stream_set_blocking($this->stream->getpeername(), 0);
|
// stream_set_blocking($this->stream->getpeername(), 0);
|
||||||
// stream_set_read_buffer($this->stream, 0);
|
// stream_set_read_buffer($this->stream, 0);
|
||||||
|
|
||||||
$this->readWatcher = \Amp\onReadable($this->stream, function () {
|
// $this->readWatcher = \Amp\onReadable($this->stream, function () {
|
||||||
do {
|
// do {
|
||||||
try {
|
// try {
|
||||||
if (!$this->isSocketDead($this->stream)) {
|
// if (!$this->isSocketDead($this->stream)) {
|
||||||
$newData = $this->stream->recv(self::READ_MAX_LEN);
|
// $newData = $this->stream->recv(self::READ_MAX_LEN);
|
||||||
} else {
|
// } else {
|
||||||
$this->reconnect();
|
// $this->reconnect();
|
||||||
return;
|
// return;
|
||||||
}
|
// }
|
||||||
if ($newData) {
|
// if ($newData) {
|
||||||
$this->read($newData);
|
// $this->read($newData);
|
||||||
}
|
// }
|
||||||
} catch (\Exception $exception) {
|
// } catch (\Exception $exception) {
|
||||||
var_dump($exception->getMessage());
|
// var_dump($exception->getMessage());
|
||||||
}
|
// }
|
||||||
} while ($newData);
|
// } while ($newData);
|
||||||
});
|
// });
|
||||||
|
//
|
||||||
$this->writeWatcher = \Amp\onWritable($this->stream, function () {
|
// $this->writeWatcher = \Amp\onWritable($this->stream, function () {
|
||||||
$this->write();
|
// $this->write();
|
||||||
}, array('enable' => false)); // <-- let's initialize the watcher as "disabled"
|
// }, array('enable' => false)); // <-- let's initialize the watcher as "disabled"
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
|
|||||||
Reference in New Issue
Block a user