<?php
/**
* Wrapper around streams, pipes, and other things that have basic read/write
* I/O characteristics.
*
* Channels include buffering, so you can do fire-and-forget writes and reads
* without worrying about network/pipe buffers. Use @{method:read} and
* @{method:write} to read and write.
*
* Channels are nonblocking and provide a select()-oriented interface so you
* can reasonably write server-like and daemon-like things with them. Use
* @{method:waitForAny} to select channels.
*
* Channel operations other than @{method:update} generally operate on buffers.
* Writes and reads affect buffers, while @{method:update} flushes output
* buffers and fills input buffers.
*
* Channels are either "open" or "closed". You can detect that a channel has
* closed by calling @{method:isOpen} or examining the return value of
* @{method:update}.
*
* NOTE: Channels are new (as of June 2012) and subject to interface changes.
*
* @task io Reading and Writing
* @task wait Waiting for Activity
* @task update Responding to Activity
* @task impl Channel Implementation
*/
abstract class PhutilChannel extends Phobject {
private $ibuf = '';
private $obuf;
private $name;
private $readBufferSize;
public function __construct() {
$this->obuf = new PhutilRope();
}
/* -( Reading and Writing )------------------------------------------------ */
/**
* Read from the channel. A channel defines the format of data that is read
* from it, so this method may return strings, objects, or anything else.
*
* The default implementation returns bytes.
*
* @return wild Data from the channel, normally bytes.
*
* @task io
*/
public function read() {
$result = $this->ibuf;
$this->ibuf = '';
return $result;
}
/**
* Write to the channel. A channel defines what data format it accepts,
* so this method may take strings, objects, or anything else.
*
* The default implementation accepts bytes.
*
* @param wild Data to write to the channel, normally bytes.
* @return this
*
* @task io
*/
public function write($bytes) {
if (!is_scalar($bytes)) {
throw new Exception(
pht(
'%s may only write strings!',
__METHOD__.'()'));
}
$this->obuf->append($bytes);
return $this;
}
/* -( Waiting for Activity )----------------------------------------------- */
/**
* Wait for any activity on a list of channels. Convenience wrapper around
* @{method:waitForActivity}.
*
* @param list<PhutilChannel> A list of channels to wait for.
* @param dict Options, see above.
* @return void
*
* @task wait
*/
public static function waitForAny(array $channels, array $options = array()) {
return self::waitForActivity($channels, $channels, $options);
}
/**
* Wait (using select()) for channels to become ready for reads or writes.
* This method blocks until some channel is ready to be updated.
*
* It does not provide a way to determine which channels are ready to be
* updated. The expectation is that you'll just update every channel. This
* might change eventually.
*
* Available options are:
*
* - 'read' (list<stream>) Additional streams to select for read.
* - 'write' (list<stream>) Additional streams to select for write.
* - 'except' (list<stream>) Additional streams to select for except.
* - 'timeout' (float) Select timeout, defaults to 1.
*
* NOTE: Extra streams must be //streams//, not //sockets//, because this
* method uses `stream_select()`, not `socket_select()`.
*
* @param list<PhutilChannel> List of channels to wait for reads on.
* @param list<PhutilChannel> List of channels to wait for writes on.
* @return void
*
* @task wait
*/
public static function waitForActivity(
array $reads,
array $writes,
array $options = array()) {
assert_instances_of($reads, __CLASS__);
assert_instances_of($writes, __CLASS__);
$read = idx($options, 'read', array());
$write = idx($options, 'write', array());
$except = idx($options, 'except', array());
$wait = idx($options, 'timeout', 1);
// TODO: It would be nice to just be able to categorically reject these as
// unselectable.
foreach (array($reads, $writes) as $channels) {
foreach ($channels as $channel) {
$r_sockets = $channel->getReadSockets();
$w_sockets = $channel->getWriteSockets();
// If any channel has no read sockets and no write sockets, assume it
// isn't selectable and return immediately (effectively degrading to a
// busy wait).
if (!$r_sockets && !$w_sockets) {
return false;
}
}
}
foreach ($reads as $channel) {
// If any of the read channels have data in read buffers, return
// immediately. If we don't, we risk running select() on a bunch of
// sockets which won't become readable because the data the application
// expects is already in a read buffer.
if (!$channel->isReadBufferEmpty()) {
return;
}
$r_sockets = $channel->getReadSockets();
foreach ($r_sockets as $socket) {
$read[] = $socket;
$except[] = $socket;
}
}
foreach ($writes as $channel) {
if ($channel->isWriteBufferEmpty()) {
// If the channel's write buffer is empty, don't select the write
// sockets, since they're writable immediately.
$w_sockets = array();
} else {
$w_sockets = $channel->getWriteSockets();
}
foreach ($w_sockets as $socket) {
$write[] = $socket;
$except[] = $socket;
}
}
if (!$read && !$write && !$except) {
return false;
}
$wait_sec = (int)$wait;
$wait_usec = 1000000 * ($wait - $wait_sec);
@stream_select($read, $write, $except, $wait_sec, $wait_usec);
}
/* -( Responding to Activity )--------------------------------------------- */
/**
* Updates the channel, filling input buffers and flushing output buffers.
* Returns false if the channel has closed.
*
* @return bool True if the channel is still open.
*
* @task update
*/
public function update() {
$maximum_read = PHP_INT_MAX;
if ($this->readBufferSize !== null) {
$maximum_read = ($this->readBufferSize - strlen($this->ibuf));
}
while ($maximum_read > 0) {
$in = $this->readBytes($maximum_read);
if (!strlen($in)) {
// Reading is blocked for now.
break;
}
$this->ibuf .= $in;
$maximum_read -= strlen($in);
}
while ($this->obuf->getByteLength()) {
$len = $this->writeBytes($this->obuf->getAnyPrefix());
if (!$len) {
// Writing is blocked for now.
break;
}
$this->obuf->removeBytesFromHead($len);
}
return $this->isOpen();
}
/* -( Channel Implementation )--------------------------------------------- */
/**
* Set a channel name. This is primarily intended to allow you to debug
* channel code more easily, by naming channels something meaningful.
*
* @param string Channel name.
* @return this
*
* @task impl
*/
public function setName($name) {
$this->name = $name;
return $this;
}
/**
* Get the channel name, as set by @{method:setName}.
*
* @return string Name of the channel.
*
* @task impl
*/
public function getName() {
return coalesce($this->name, get_class($this));
}
/**
* Test if the channel is open: active, can be read from and written to, etc.
*
* @return bool True if the channel is open.
*
* @task impl
*/
abstract public function isOpen();
/**
* Close the channel for writing.
*
* @return void
* @task impl
*/
abstract public function closeWriteChannel();
/**
* Test if the channel is open for reading.
*
* @return bool True if the channel is open for reading.
*
* @task impl
*/
public function isOpenForReading() {
return $this->isOpen();
}
/**
* Test if the channel is open for writing.
*
* @return bool True if the channel is open for writing.
*
* @task impl
*/
public function isOpenForWriting() {
return $this->isOpen();
}
/**
* Read from the channel's underlying I/O.
*
* @param int Maximum number of bytes to read.
* @return string Bytes, if available.
*
* @task impl
*/
abstract protected function readBytes($length);
/**
* Write to the channel's underlying I/O.
*
* @param string Bytes to write.
* @return int Number of bytes written.
*
* @task impl
*/
abstract protected function writeBytes($bytes);
/**
* Get sockets to select for reading.
*
* @return list<stream> Read sockets.
*
* @task impl
*/
protected function getReadSockets() {
return array();
}
/**
* Get sockets to select for writing.
*
* @return list<stream> Write sockets.
*
* @task impl
*/
protected function getWriteSockets() {
return array();
}
/**
* Set the maximum size of the channel's read buffer. Reads will artificially
* block once the buffer reaches this size until the in-process buffer is
* consumed.
*
* @param int|null Maximum read buffer size, or `null` for a limitless buffer.
* @return this
* @task impl
*/
public function setReadBufferSize($size) {
$this->readBufferSize = $size;
return $this;
}
/**
* Test state of the read buffer.
*
* @return bool True if the read buffer is empty.
*
* @task impl
*/
public function isReadBufferEmpty() {
return (strlen($this->ibuf) == 0);
}
/**
* Test state of the write buffer.
*
* @return bool True if the write buffer is empty.
*
* @task impl
*/
public function isWriteBufferEmpty() {
return !$this->getWriteBufferSize();
}
/**
* Get the number of bytes we're currently waiting to write.
*
* @return int Number of waiting bytes.
*
* @task impl
*/
public function getWriteBufferSize() {
return $this->obuf->getByteLength();
}
/**
* Wait for any buffered writes to complete. This is a blocking call. When
* the call returns, the write buffer will be empty.
*
* @task impl
*/
public function flush() {
while (!$this->isWriteBufferEmpty()) {
self::waitForAny(array($this));
if (!$this->update()) {
throw new Exception(pht('Channel closed while flushing output!'));
}
}
return $this;
}
}