added new socket manager, errorhandling and testing is not finished yet

libs/React/Stream/Buffer.php

@@ -0,0 +1,135 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitter;
+use React\EventLoop\LoopInterface;
+
+/** @event full-drain */
+class Buffer extends EventEmitter implements WritableStreamInterface
+{
+    public $stream;
+    public $listening = false;
+    public $softLimit = 2048;
+    private $writable = true;
+    private $loop;
+    private $data = '';
+    private $lastError = array(
+        'number'  => 0,
+        'message' => '',
+        'file'    => '',
+        'line'    => 0,
+    );
+
+    public function __construct($stream, LoopInterface $loop)
+    {
+        $this->stream = $stream;
+        $this->loop = $loop;
+    }
+
+    public function isWritable()
+    {
+        return $this->writable;
+    }
+
+    public function write($data)
+    {
+        if (!$this->writable) {
+            return;
+        }
+
+        $this->data .= $data;
+
+        if (!$this->listening) {
+            $this->listening = true;
+
+            $this->loop->addWriteStream($this->stream, array($this, 'handleWrite'));
+        }
+
+        $belowSoftLimit = strlen($this->data) < $this->softLimit;
+
+        return $belowSoftLimit;
+    }
+
+    public function end($data = null)
+    {
+        if (null !== $data) {
+            $this->write($data);
+        }
+
+        $this->writable = false;
+
+        if ($this->listening) {
+            $this->on('full-drain', array($this, 'close'));
+        } else {
+            $this->close();
+        }
+    }
+
+    public function close()
+    {
+        $this->writable = false;
+        $this->listening = false;
+        $this->data = '';
+
+        $this->emit('close', [$this]);
+    }
+
+    public function handleWrite()
+    {
+        if (!is_resource($this->stream)) {
+            $this->emit('error', array(new \RuntimeException('Tried to write to invalid stream.'), $this));
+
+            return;
+        }
+
+        set_error_handler(array($this, 'errorHandler'));
+
+        $sent = fwrite($this->stream, $this->data);
+
+        restore_error_handler();
+
+        if (false === $sent) {
+            $this->emit('error', array(
+                new \ErrorException(
+                    $this->lastError['message'],
+                    0,
+                    $this->lastError['number'],
+                    $this->lastError['file'],
+                    $this->lastError['line']
+                ),
+                $this
+            ));
+
+            return;
+        }
+
+        if (0 === $sent && feof($this->stream)) {
+            $this->emit('error', array(new \RuntimeException('Tried to write to closed stream.'), $this));
+
+            return;
+        }
+
+        $len = strlen($this->data);
+        if ($len >= $this->softLimit && $len - $sent < $this->softLimit) {
+            $this->emit('drain', [$this]);
+        }
+
+        $this->data = (string) substr($this->data, $sent);
+
+        if (0 === strlen($this->data)) {
+            $this->loop->removeWriteStream($this->stream);
+            $this->listening = false;
+
+            $this->emit('full-drain', [$this]);
+        }
+    }
+
+    private function errorHandler($errno, $errstr, $errfile, $errline)
+    {
+        $this->lastError['number']  = $errno;
+        $this->lastError['message'] = $errstr;
+        $this->lastError['file']    = $errfile;
+        $this->lastError['line']    = $errline;
+    }
+}

libs/React/Stream/BufferedSink.php

@@ -0,0 +1,59 @@
+<?php
+
+namespace React\Stream;
+
+use React\Promise\Deferred;
+use React\Promise\PromisorInterface;
+
+class BufferedSink extends WritableStream implements PromisorInterface
+{
+    private $buffer = '';
+    private $deferred;
+
+    public function __construct()
+    {
+        $this->deferred = new Deferred();
+
+        $this->on('pipe', array($this, 'handlePipeEvent'));
+        $this->on('error', array($this, 'handleErrorEvent'));
+    }
+
+    public function handlePipeEvent($source)
+    {
+        Util::forwardEvents($source, $this, array('error'));
+    }
+
+    public function handleErrorEvent($e)
+    {
+        $this->deferred->reject($e);
+    }
+
+    public function write($data)
+    {
+        $this->buffer .= $data;
+        $this->deferred->progress($data);
+    }
+
+    public function close()
+    {
+        if ($this->closed) {
+            return;
+        }
+
+        parent::close();
+        $this->deferred->resolve($this->buffer);
+    }
+
+    public function promise()
+    {
+        return $this->deferred->promise();
+    }
+
+    public static function createPromise(ReadableStreamInterface $stream)
+    {
+        $sink = new static();
+        $stream->pipe($sink);
+
+        return $sink->promise();
+    }
+}

libs/React/Stream/CompositeStream.php

@@ -0,0 +1,84 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitter;
+
+class CompositeStream extends EventEmitter implements DuplexStreamInterface
+{
+    protected $readable;
+    protected $writable;
+    protected $pipeSource;
+
+    public function __construct(ReadableStreamInterface $readable, WritableStreamInterface $writable)
+    {
+        $this->readable = $readable;
+        $this->writable = $writable;
+
+        Util::forwardEvents($this->readable, $this, array('data', 'end', 'error', 'close'));
+        Util::forwardEvents($this->writable, $this, array('drain', 'error', 'close', 'pipe'));
+
+        $this->readable->on('close', array($this, 'close'));
+        $this->writable->on('close', array($this, 'close'));
+
+        $this->on('pipe', array($this, 'handlePipeEvent'));
+    }
+
+    public function handlePipeEvent($source)
+    {
+        $this->pipeSource = $source;
+    }
+
+    public function isReadable()
+    {
+        return $this->readable->isReadable();
+    }
+
+    public function pause()
+    {
+        if ($this->pipeSource) {
+            $this->pipeSource->pause();
+        }
+
+        $this->readable->pause();
+    }
+
+    public function resume()
+    {
+        if ($this->pipeSource) {
+            $this->pipeSource->resume();
+        }
+
+        $this->readable->resume();
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+
+    public function isWritable()
+    {
+        return $this->writable->isWritable();
+    }
+
+    public function write($data)
+    {
+        return $this->writable->write($data);
+    }
+
+    public function end($data = null)
+    {
+        $this->writable->end($data);
+    }
+
+    public function close()
+    {
+        $this->pipeSource = null;
+
+        $this->readable->close();
+        $this->writable->close();
+    }
+}

libs/React/Stream/DuplexStreamInterface.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace React\Stream;
+
+interface DuplexStreamInterface extends ReadableStreamInterface, WritableStreamInterface
+{
+}

libs/React/Stream/ReadableStream.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitter;
+
+class ReadableStream extends EventEmitter implements ReadableStreamInterface
+{
+    protected $closed = false;
+
+    public function isReadable()
+    {
+        return !$this->closed;
+    }
+
+    public function pause()
+    {
+    }
+
+    public function resume()
+    {
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+
+    public function close()
+    {
+        if ($this->closed) {
+            return;
+        }
+
+        $this->closed = true;
+        $this->emit('end', array($this));
+        $this->emit('close', array($this));
+        $this->removeAllListeners();
+    }
+}

libs/React/Stream/ReadableStreamInterface.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitterInterface;
+
+/**
+ * @event data
+ * @event end
+ * @event error
+ * @event close
+ */
+interface ReadableStreamInterface extends EventEmitterInterface
+{
+    public function isReadable();
+    public function pause();
+    public function resume();
+    public function pipe(WritableStreamInterface $dest, array $options = array());
+    public function close();
+}

libs/React/Stream/Stream.php

@@ -0,0 +1,141 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitter;
+use React\EventLoop\LoopInterface;
+use InvalidArgumentException;
+
+class Stream extends EventEmitter implements DuplexStreamInterface
+{
+    public $bufferSize = 4096;
+    public $stream;
+    protected $readable = true;
+    protected $writable = true;
+    protected $closing = false;
+    protected $loop;
+    protected $buffer;
+
+    public function __construct($stream, LoopInterface $loop)
+    {
+        $this->stream = $stream;
+        if (!is_resource($this->stream) || get_resource_type($this->stream) !== "stream") {
+             throw new InvalidArgumentException('First parameter must be a valid stream resource');
+        }
+
+        stream_set_blocking($this->stream, 0);
+
+        $this->loop = $loop;
+        $this->buffer = new Buffer($this->stream, $this->loop);
+
+        $this->buffer->on('error', function ($error) {
+            $this->emit('error', array($error, $this));
+            $this->close();
+        });
+
+        $this->buffer->on('drain', function () {
+            $this->emit('drain', array($this));
+        });
+
+        $this->resume();
+    }
+
+    public function isReadable()
+    {
+        return $this->readable;
+    }
+
+    public function isWritable()
+    {
+        return $this->writable;
+    }
+
+    public function pause()
+    {
+        $this->loop->removeReadStream($this->stream);
+    }
+
+    public function resume()
+    {
+        if ($this->readable) {
+            $this->loop->addReadStream($this->stream, array($this, 'handleData'));
+        }
+    }
+
+    public function write($data)
+    {
+        if (!$this->writable) {
+            return;
+        }
+
+        return $this->buffer->write($data);
+    }
+
+    public function close()
+    {
+        if (!$this->writable && !$this->closing) {
+            return;
+        }
+
+        $this->closing = false;
+
+        $this->readable = false;
+        $this->writable = false;
+
+        $this->emit('end', array($this));
+        $this->emit('close', array($this));
+        $this->loop->removeStream($this->stream);
+        $this->buffer->removeAllListeners();
+        $this->removeAllListeners();
+
+        $this->handleClose();
+    }
+
+    public function end($data = null)
+    {
+        if (!$this->writable) {
+            return;
+        }
+
+        $this->closing = true;
+
+        $this->readable = false;
+        $this->writable = false;
+
+        $this->buffer->on('close', function () {
+            $this->close();
+        });
+
+        $this->buffer->end($data);
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+
+    public function handleData($stream)
+    {
+        $data = fread($stream, $this->bufferSize);
+
+        $this->emit('data', array($data, $this));
+
+        if (!is_resource($stream) || feof($stream)) {
+            $this->end();
+        }
+    }
+
+    public function handleClose()
+    {
+        if (is_resource($this->stream)) {
+            fclose($this->stream);
+        }
+    }
+
+    public function getBuffer()
+    {
+        return $this->buffer;
+    }
+}

libs/React/Stream/ThroughStream.php

@@ -0,0 +1,33 @@
+<?php
+
+namespace React\Stream;
+
+class ThroughStream extends CompositeStream
+{
+    public function __construct()
+    {
+        $readable = new ReadableStream();
+        $writable = new WritableStream();
+
+        parent::__construct($readable, $writable);
+    }
+
+    public function filter($data)
+    {
+        return $data;
+    }
+
+    public function write($data)
+    {
+        $this->readable->emit('data', array($this->filter($data), $this));
+    }
+
+    public function end($data = null)
+    {
+        if (null !== $data) {
+            $this->readable->emit('data', array($this->filter($data), $this));
+        }
+
+        $this->writable->end($data);
+    }
+}

libs/React/Stream/Util.php

@@ -0,0 +1,45 @@
+<?php
+
+namespace React\Stream;
+
+// TODO: move to a trait
+
+class Util
+{
+    public static function pipe(ReadableStreamInterface $source, WritableStreamInterface $dest, array $options = array())
+    {
+        // TODO: use stream_copy_to_stream
+        // it is 4x faster than this
+        // but can lose data under load with no way to recover it
+
+        $dest->emit('pipe', array($source));
+
+        $source->on('data', function ($data) use ($source, $dest) {
+            $feedMore = $dest->write($data);
+
+            if (false === $feedMore) {
+                $source->pause();
+            }
+        });
+
+        $dest->on('drain', function () use ($source) {
+            $source->resume();
+        });
+
+        $end = isset($options['end']) ? $options['end'] : true;
+        if ($end && $source !== $dest) {
+            $source->on('end', function () use ($dest) {
+                $dest->end();
+            });
+        }
+    }
+
+    public static function forwardEvents($source, $target, array $events)
+    {
+        foreach ($events as $event) {
+            $source->on($event, function () use ($event, $target) {
+                $target->emit($event, func_get_args());
+            });
+        }
+    }
+}

libs/React/Stream/WritableStream.php

@@ -0,0 +1,40 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitter;
+
+class WritableStream extends EventEmitter implements WritableStreamInterface
+{
+    protected $closed = false;
+
+    public function write($data)
+    {
+    }
+
+    public function end($data = null)
+    {
+        if (null !== $data) {
+            $this->write($data);
+        }
+
+        $this->close();
+    }
+
+    public function isWritable()
+    {
+        return !$this->closed;
+    }
+
+    public function close()
+    {
+        if ($this->closed) {
+            return;
+        }
+
+        $this->closed = true;
+        $this->emit('end', array($this));
+        $this->emit('close', array($this));
+        $this->removeAllListeners();
+    }
+}

libs/React/Stream/WritableStreamInterface.php

@@ -0,0 +1,19 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitterInterface;
+
+/**
+ * @event drain
+ * @event error
+ * @event close
+ * @event pipe
+ */
+interface WritableStreamInterface extends EventEmitterInterface
+{
+    public function isWritable();
+    public function write($data);
+    public function end($data = null);
+    public function close();
+}