改名
This commit is contained in:
@@ -73,9 +73,6 @@ class Consumer
|
|||||||
}
|
}
|
||||||
$process->start();
|
$process->start();
|
||||||
$this->isRunning = true;
|
$this->isRunning = true;
|
||||||
if ($isBlock) {
|
|
||||||
\Amp\run();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
|
|||||||
@@ -136,13 +136,13 @@ class State
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
$interval = isset($option['interval']) ? $option['interval'] : 200;
|
$interval = isset($option['interval']) ? $option['interval'] : 200;
|
||||||
\Amp\repeat(function ($watcherId) use ($request, $option) {
|
Timer::tick($msInterval = $interval, function ($watcherId) use ($request, $option) {
|
||||||
if ($this->checkRun($request) && $option['func'] != null) {
|
if ($this->checkRun($request) && $option['func'] != null) {
|
||||||
$context = call_user_func($option['func']);
|
$context = call_user_func($option['func']);
|
||||||
$this->processing($request, $context);
|
$this->processing($request, $context);
|
||||||
}
|
}
|
||||||
$this->requests[$request]['watcher'] = $watcherId;
|
$this->requests[$request]['watcher'] = $watcherId;
|
||||||
}, $msInterval = $interval);
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
// start sync metadata
|
// start sync metadata
|
||||||
@@ -150,9 +150,9 @@ class State
|
|||||||
$context = call_user_func($this->requests[self::REQUEST_METADATA]['func']);
|
$context = call_user_func($this->requests[self::REQUEST_METADATA]['func']);
|
||||||
$this->processing($request, $context);
|
$this->processing($request, $context);
|
||||||
}
|
}
|
||||||
\Amp\repeat(function ($watcherId) {
|
Timer::tick($msInterval = 1000, function ($watcherId) {
|
||||||
$this->report();
|
$this->report();
|
||||||
}, $msInterval = 1000);
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
|
|||||||
+359
-361
@@ -15,417 +15,415 @@
|
|||||||
namespace Kafka;
|
namespace Kafka;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
+------------------------------------------------------------------------------
|
* +------------------------------------------------------------------------------
|
||||||
* Kafka broker socket
|
* Kafka broker socket
|
||||||
+------------------------------------------------------------------------------
|
* +------------------------------------------------------------------------------
|
||||||
*
|
*
|
||||||
* @package
|
* @package
|
||||||
* @version $_SWANBR_VERSION_$
|
* @version $_SWANBR_VERSION_$
|
||||||
* @copyright Copyleft
|
* @copyright Copyleft
|
||||||
* @author $_SWANBR_AUTHOR_$
|
* @author $_SWANBR_AUTHOR_$
|
||||||
+------------------------------------------------------------------------------
|
* +------------------------------------------------------------------------------
|
||||||
*/
|
*/
|
||||||
|
|
||||||
class Socket
|
class Socket
|
||||||
{
|
{
|
||||||
// {{{ consts
|
// {{{ consts
|
||||||
|
|
||||||
const READ_MAX_LEN = 5242880; // read socket max length 5MB
|
const READ_MAX_LEN = 5242880; // read socket max length 5MB
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* max write socket buffer
|
* max write socket buffer
|
||||||
* fixed:send of 8192 bytes failed with errno=11 Resource temporarily
|
* fixed:send of 8192 bytes failed with errno=11 Resource temporarily
|
||||||
* fixed:'fwrite(): send of ???? bytes failed with errno=35 Resource temporarily unavailable'
|
* fixed:'fwrite(): send of ???? bytes failed with errno=35 Resource temporarily unavailable'
|
||||||
* unavailable error info
|
* unavailable error info
|
||||||
*/
|
*/
|
||||||
const MAX_WRITE_BUFFER = 2048;
|
const MAX_WRITE_BUFFER = 2048;
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ members
|
// {{{ members
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Send timeout in seconds.
|
* Send timeout in seconds.
|
||||||
*
|
*
|
||||||
* @var float
|
* @var float
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $sendTimeoutSec = 0;
|
private $sendTimeoutSec = 0;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Send timeout in microseconds.
|
* Send timeout in microseconds.
|
||||||
*
|
*
|
||||||
* @var float
|
* @var float
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $sendTimeoutUsec = 100000;
|
private $sendTimeoutUsec = 100000;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Recv timeout in seconds
|
* Recv timeout in seconds
|
||||||
*
|
*
|
||||||
* @var float
|
* @var float
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $recvTimeoutSec = 0;
|
private $recvTimeoutSec = 0;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Recv timeout in microseconds
|
* Recv timeout in microseconds
|
||||||
*
|
*
|
||||||
* @var float
|
* @var float
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $recvTimeoutUsec = 750000;
|
private $recvTimeoutUsec = 750000;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Stream resource
|
* Stream resource
|
||||||
*
|
*
|
||||||
* @var mixed
|
* @var \Swoole\Coroutine\Socket
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $stream = null;
|
private $stream = null;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Socket host
|
* Socket host
|
||||||
*
|
*
|
||||||
* @var mixed
|
* @var mixed
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $host = null;
|
private $host = null;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Socket port
|
* Socket port
|
||||||
*
|
*
|
||||||
* @var mixed
|
* @var mixed
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $port = -1;
|
private $port = -1;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Max Write Attempts
|
* Max Write Attempts
|
||||||
* @var int
|
* @var int
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $maxWriteAttempts = 3;
|
private $maxWriteAttempts = 3;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Reader watcher
|
* Reader watcher
|
||||||
* @var int
|
* @var int
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $readWatcher = 0;
|
private $readWatcher = 0;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Write watcher
|
* Write watcher
|
||||||
* @var int
|
* @var int
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $writeWatcher = 0;
|
private $writeWatcher = 0;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Write watcher
|
* Write watcher
|
||||||
* @var int
|
* @var int
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $writeBuffer = '';
|
private $writeBuffer = '';
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Reader buffer
|
* Reader buffer
|
||||||
* @var int
|
* @var int
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $readBuffer = '';
|
private $readBuffer = '';
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Reader need buffer length
|
* Reader need buffer length
|
||||||
* @var int
|
* @var int
|
||||||
* @access private
|
* @access private
|
||||||
*/
|
*/
|
||||||
private $readNeedLength = 0;
|
private $readNeedLength = 0;
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ functions
|
// {{{ functions
|
||||||
// {{{ public function __construct()
|
// {{{ public function __construct()
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* __construct
|
* __construct
|
||||||
*
|
*
|
||||||
* @access public
|
* @access public
|
||||||
* @param $host
|
* @param $host
|
||||||
* @param $port
|
* @param $port
|
||||||
* @param int $recvTimeoutSec
|
* @param int $recvTimeoutSec
|
||||||
* @param int $recvTimeoutUsec
|
* @param int $recvTimeoutUsec
|
||||||
* @param int $sendTimeoutSec
|
* @param int $sendTimeoutSec
|
||||||
* @param int $sendTimeoutUsec
|
* @param int $sendTimeoutUsec
|
||||||
*/
|
*/
|
||||||
public function __construct($host, $port, $recvTimeoutSec = 0, $recvTimeoutUsec = 750000, $sendTimeoutSec = 0, $sendTimeoutUsec = 100000)
|
public function __construct($host, $port, $recvTimeoutSec = 0, $recvTimeoutUsec = 750000, $sendTimeoutSec = 0, $sendTimeoutUsec = 100000)
|
||||||
{
|
{
|
||||||
$this->host = $host;
|
$this->host = $host;
|
||||||
$this->port = $port;
|
$this->port = $port;
|
||||||
$this->setRecvTimeoutSec($recvTimeoutSec);
|
$this->setRecvTimeoutSec($recvTimeoutSec);
|
||||||
$this->setRecvTimeoutUsec($recvTimeoutUsec);
|
$this->setRecvTimeoutUsec($recvTimeoutUsec);
|
||||||
$this->setSendTimeoutSec($sendTimeoutSec);
|
$this->setSendTimeoutSec($sendTimeoutSec);
|
||||||
$this->setSendTimeoutUsec($sendTimeoutUsec);
|
$this->setSendTimeoutUsec($sendTimeoutUsec);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param float $sendTimeoutSec
|
* @param float $sendTimeoutSec
|
||||||
*/
|
*/
|
||||||
public function setSendTimeoutSec($sendTimeoutSec)
|
public function setSendTimeoutSec($sendTimeoutSec)
|
||||||
{
|
{
|
||||||
$this->sendTimeoutSec = $sendTimeoutSec;
|
$this->sendTimeoutSec = $sendTimeoutSec;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param float $sendTimeoutUsec
|
* @param float $sendTimeoutUsec
|
||||||
*/
|
*/
|
||||||
public function setSendTimeoutUsec($sendTimeoutUsec)
|
public function setSendTimeoutUsec($sendTimeoutUsec)
|
||||||
{
|
{
|
||||||
$this->sendTimeoutUsec = $sendTimeoutUsec;
|
$this->sendTimeoutUsec = $sendTimeoutUsec;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param float $recvTimeoutSec
|
* @param float $recvTimeoutSec
|
||||||
*/
|
*/
|
||||||
public function setRecvTimeoutSec($recvTimeoutSec)
|
public function setRecvTimeoutSec($recvTimeoutSec)
|
||||||
{
|
{
|
||||||
$this->recvTimeoutSec = $recvTimeoutSec;
|
$this->recvTimeoutSec = $recvTimeoutSec;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param float $recvTimeoutUsec
|
* @param float $recvTimeoutUsec
|
||||||
*/
|
*/
|
||||||
public function setRecvTimeoutUsec($recvTimeoutUsec)
|
public function setRecvTimeoutUsec($recvTimeoutUsec)
|
||||||
{
|
{
|
||||||
$this->recvTimeoutUsec = $recvTimeoutUsec;
|
$this->recvTimeoutUsec = $recvTimeoutUsec;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param int $number
|
* @param int $number
|
||||||
*/
|
*/
|
||||||
public function setMaxWriteAttempts($number)
|
public function setMaxWriteAttempts($number)
|
||||||
{
|
{
|
||||||
$this->maxWriteAttempts = $number;
|
$this->maxWriteAttempts = $number;
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ public function connect()
|
// {{{ public function connect()
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Connects the socket
|
* Connects the socket
|
||||||
*
|
*
|
||||||
* @access public
|
* @access public
|
||||||
* @return void
|
* @return void
|
||||||
*/
|
*/
|
||||||
public function connect()
|
public function connect()
|
||||||
{
|
{
|
||||||
if (!$this->isSocketDead()) {
|
if (!$this->isSocketDead()) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (empty($this->host)) {
|
if (empty($this->host)) {
|
||||||
throw new \Kafka\Exception('Cannot open null host.');
|
throw new \Kafka\Exception('Cannot open null host.');
|
||||||
}
|
}
|
||||||
if ($this->port <= 0) {
|
if ($this->port <= 0) {
|
||||||
throw new \Kafka\Exception('Cannot open without port.');
|
throw new \Kafka\Exception('Cannot open without port.');
|
||||||
}
|
}
|
||||||
|
|
||||||
$this->stream = @fsockopen(
|
$this->stream = new \Swoole\Coroutine\Socket(AF_INET, SOCK_STREAM);;
|
||||||
$this->host,
|
|
||||||
$this->port,
|
|
||||||
$errno,
|
|
||||||
$errstr,
|
|
||||||
$this->sendTimeoutSec + ($this->sendTimeoutUsec / 1000000)
|
|
||||||
);
|
|
||||||
|
|
||||||
if ($this->stream == false) {
|
if (!$this->stream->connect(
|
||||||
$error = 'Could not connect to '
|
$this->host,
|
||||||
. $this->host . ':' . $this->port
|
$this->port,
|
||||||
. ' ('.$errstr.' ['.$errno.'])';
|
$this->sendTimeoutSec + ($this->sendTimeoutUsec / 1000000)
|
||||||
throw new \Kafka\Exception($error);
|
)) {
|
||||||
}
|
$error = 'Could not connect to '
|
||||||
|
. $this->host . ':' . $this->port
|
||||||
|
. ' (' . $this->stream->errMsg . ' [' . $this->stream->errCode . '])';
|
||||||
|
throw new \Kafka\Exception($error);
|
||||||
|
}
|
||||||
|
|
||||||
stream_set_blocking($this->stream, 0);
|
stream_set_blocking($this->stream, 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 = @fread($this->stream, self::READ_MAX_LEN);
|
$newData = @fread($this->stream, 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"
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ public function reconnect()
|
// {{{ public function reconnect()
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* reconnect the socket
|
* reconnect the socket
|
||||||
*
|
*
|
||||||
* @access public
|
* @access public
|
||||||
* @return void
|
* @return void
|
||||||
*/
|
*/
|
||||||
public function reconnect()
|
public function reconnect()
|
||||||
{
|
{
|
||||||
$this->close();
|
$this->close();
|
||||||
$this->connect();
|
$this->connect();
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ public function getSocket()
|
// {{{ public function getSocket()
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* get the socket
|
* get the socket
|
||||||
*
|
*
|
||||||
* @access public
|
* @access public
|
||||||
* @return void
|
* @return void
|
||||||
*/
|
*/
|
||||||
public function getSocket()
|
public function getSocket()
|
||||||
{
|
{
|
||||||
return $this->stream;
|
return $this->stream;
|
||||||
}
|
}
|
||||||
|
|
||||||
// }}}
|
// }}}
|
||||||
// {{{ public function SetOnReadable()
|
// {{{ public function SetOnReadable()
|
||||||
|
|
||||||
public $onReadable;
|
public $onReadable;
|
||||||
/**
|
|
||||||
* set on readable callback function
|
|
||||||
*
|
|
||||||
* @access public
|
|
||||||
* @return void
|
|
||||||
*/
|
|
||||||
public function SetOnReadable(\Closure $read)
|
|
||||||
{
|
|
||||||
$this->onReadable = $read;
|
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
/**
|
||||||
// {{{ public function close()
|
* set on readable callback function
|
||||||
|
*
|
||||||
|
* @access public
|
||||||
|
* @return void
|
||||||
|
*/
|
||||||
|
public function SetOnReadable(\Closure $read)
|
||||||
|
{
|
||||||
|
$this->onReadable = $read;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
// }}}
|
||||||
* close the socket
|
// {{{ public function close()
|
||||||
*
|
|
||||||
* @access public
|
|
||||||
* @return void
|
|
||||||
*/
|
|
||||||
public function close()
|
|
||||||
{
|
|
||||||
\Amp\cancel($this->readWatcher);
|
|
||||||
\Amp\cancel($this->writeWatcher);
|
|
||||||
if (is_resource($this->stream)) {
|
|
||||||
fclose($this->stream);
|
|
||||||
}
|
|
||||||
$this->readBuffer = '';
|
|
||||||
$this->writeBuffer = '';
|
|
||||||
$this->readNeedLength = 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* checks if the socket is a valid resource
|
* close the socket
|
||||||
*
|
*
|
||||||
* @access public
|
* @access public
|
||||||
* @return boolean
|
* @return void
|
||||||
*/
|
*/
|
||||||
public function isResource()
|
public function close()
|
||||||
{
|
{
|
||||||
return is_resource($this->stream);
|
\Amp\cancel($this->readWatcher);
|
||||||
}
|
\Amp\cancel($this->writeWatcher);
|
||||||
|
if (is_resource($this->stream)) {
|
||||||
|
fclose($this->stream);
|
||||||
|
}
|
||||||
|
$this->readBuffer = '';
|
||||||
|
$this->writeBuffer = '';
|
||||||
|
$this->readNeedLength = 0;
|
||||||
|
}
|
||||||
|
|
||||||
// }}}
|
/**
|
||||||
// {{{ public function read()
|
* checks if the socket is a valid resource
|
||||||
|
*
|
||||||
|
* @access public
|
||||||
|
* @return boolean
|
||||||
|
*/
|
||||||
|
public function isResource()
|
||||||
|
{
|
||||||
|
return is_resource($this->stream);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
// }}}
|
||||||
* Read from the socket at most $len bytes.
|
// {{{ public function read()
|
||||||
*
|
|
||||||
* This method will not wait for all the requested data, it will return as
|
|
||||||
* soon as any data is received.
|
|
||||||
*
|
|
||||||
* @param integer $len Maximum number of bytes to read.
|
|
||||||
* @param boolean $verifyExactLength Throw an exception if the number of read bytes is less than $len
|
|
||||||
*
|
|
||||||
* @return string Binary data
|
|
||||||
* @throws \Kafka\Exception\SocketEOF
|
|
||||||
*/
|
|
||||||
public function read($data)
|
|
||||||
{
|
|
||||||
$this->readBuffer .= $data;
|
|
||||||
do {
|
|
||||||
if ($this->readNeedLength == 0) { // response start
|
|
||||||
if (strlen($this->readBuffer) < 4) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
$dataLen = \Kafka\Protocol\Protocol::unpack(\Kafka\Protocol\Protocol::BIT_B32, substr($this->readBuffer, 0, 4));
|
|
||||||
$this->readNeedLength = $dataLen;
|
|
||||||
$this->readBuffer = substr($this->readBuffer, 4);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (strlen($this->readBuffer) < $this->readNeedLength) {
|
/**
|
||||||
return;
|
* Read from the socket at most $len bytes.
|
||||||
}
|
*
|
||||||
$data = substr($this->readBuffer, 0, $this->readNeedLength);
|
* This method will not wait for all the requested data, it will return as
|
||||||
|
* soon as any data is received.
|
||||||
|
*
|
||||||
|
* @param integer $len Maximum number of bytes to read.
|
||||||
|
* @param boolean $verifyExactLength Throw an exception if the number of read bytes is less than $len
|
||||||
|
*
|
||||||
|
* @return string Binary data
|
||||||
|
* @throws \Kafka\Exception\SocketEOF
|
||||||
|
*/
|
||||||
|
public function read($data)
|
||||||
|
{
|
||||||
|
$this->readBuffer .= $data;
|
||||||
|
do {
|
||||||
|
if ($this->readNeedLength == 0) { // response start
|
||||||
|
if (strlen($this->readBuffer) < 4) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
$dataLen = \Kafka\Protocol\Protocol::unpack(\Kafka\Protocol\Protocol::BIT_B32, substr($this->readBuffer, 0, 4));
|
||||||
|
$this->readNeedLength = $dataLen;
|
||||||
|
$this->readBuffer = substr($this->readBuffer, 4);
|
||||||
|
}
|
||||||
|
|
||||||
$this->readBuffer = substr($this->readBuffer, $this->readNeedLength);
|
if (strlen($this->readBuffer) < $this->readNeedLength) {
|
||||||
$this->readNeedLength = 0;
|
return;
|
||||||
call_user_func($this->onReadable, $data, (int)$this->stream);
|
}
|
||||||
} while (strlen($this->readBuffer));
|
$data = substr($this->readBuffer, 0, $this->readNeedLength);
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
$this->readBuffer = substr($this->readBuffer, $this->readNeedLength);
|
||||||
// {{{ public function write()
|
$this->readNeedLength = 0;
|
||||||
|
call_user_func($this->onReadable, $data, (int)$this->stream);
|
||||||
|
} while (strlen($this->readBuffer));
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
// }}}
|
||||||
* Write to the socket.
|
// {{{ public function write()
|
||||||
*
|
|
||||||
* @param string $buf The data to write
|
|
||||||
*
|
|
||||||
* @return integer
|
|
||||||
* @throws \Kafka\Exception\SocketEOF
|
|
||||||
*/
|
|
||||||
public function write($data = null)
|
|
||||||
{
|
|
||||||
if ($data != null) {
|
|
||||||
$this->writeBuffer .= $data;
|
|
||||||
}
|
|
||||||
$bytesToWrite = strlen($this->writeBuffer);
|
|
||||||
$bytesWritten = @fwrite($this->stream, $this->writeBuffer);
|
|
||||||
|
|
||||||
if ($bytesToWrite === $bytesWritten) {
|
/**
|
||||||
\Amp\disable($this->writeWatcher);
|
* Write to the socket.
|
||||||
} elseif ($bytesWritten >= 0) {
|
*
|
||||||
\Amp\enable($this->writeWatcher);
|
* @param string $buf The data to write
|
||||||
} elseif ($this->isSocketDead($this->stream)) {
|
*
|
||||||
$this->reconnect();
|
* @return integer
|
||||||
}
|
* @throws \Kafka\Exception\SocketEOF
|
||||||
$this->writeBuffer = substr($this->writeBuffer, $bytesWritten);
|
*/
|
||||||
}
|
public function write($data = null)
|
||||||
|
{
|
||||||
|
if ($data != null) {
|
||||||
|
$this->writeBuffer .= $data;
|
||||||
|
}
|
||||||
|
$bytesToWrite = strlen($this->writeBuffer);
|
||||||
|
$bytesWritten = @fwrite($this->stream, $this->writeBuffer);
|
||||||
|
|
||||||
// }}}
|
if ($bytesToWrite === $bytesWritten) {
|
||||||
// {{{ protected function isSocketDead()
|
\Amp\disable($this->writeWatcher);
|
||||||
|
} elseif ($bytesWritten >= 0) {
|
||||||
|
\Amp\enable($this->writeWatcher);
|
||||||
|
} elseif ($this->isSocketDead($this->stream)) {
|
||||||
|
$this->reconnect();
|
||||||
|
}
|
||||||
|
$this->writeBuffer = substr($this->writeBuffer, $bytesWritten);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
// }}}
|
||||||
* check the stream is close
|
// {{{ protected function isSocketDead()
|
||||||
*
|
|
||||||
* @return bool
|
|
||||||
*/
|
|
||||||
protected function isSocketDead()
|
|
||||||
{
|
|
||||||
return !is_resource($this->stream) || @feof($this->stream);
|
|
||||||
}
|
|
||||||
|
|
||||||
// }}}
|
/**
|
||||||
// }}}
|
* check the stream is close
|
||||||
|
*
|
||||||
|
* @return bool
|
||||||
|
*/
|
||||||
|
protected function isSocketDead()
|
||||||
|
{
|
||||||
|
return !is_resource($this->stream) || @feof($this->stream);
|
||||||
|
}
|
||||||
|
|
||||||
|
// }}}
|
||||||
|
// }}}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user