summaryrefslogtreecommitdiffstats
path: root/vendor/react/stream/src/CompositeStream.php
diff options
context:
space:
mode:
Diffstat (limited to 'vendor/react/stream/src/CompositeStream.php')
-rw-r--r--vendor/react/stream/src/CompositeStream.php83
1 files changed, 83 insertions, 0 deletions
diff --git a/vendor/react/stream/src/CompositeStream.php b/vendor/react/stream/src/CompositeStream.php
new file mode 100644
index 0000000..dde091d
--- /dev/null
+++ b/vendor/react/stream/src/CompositeStream.php
@@ -0,0 +1,83 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitter;
+
+final class CompositeStream extends EventEmitter implements DuplexStreamInterface
+{
+ private $readable;
+ private $writable;
+ private $closed = false;
+
+ public function __construct(ReadableStreamInterface $readable, WritableStreamInterface $writable)
+ {
+ $this->readable = $readable;
+ $this->writable = $writable;
+
+ if (!$readable->isReadable() || !$writable->isWritable()) {
+ $this->close();
+ return;
+ }
+
+ Util::forwardEvents($this->readable, $this, array('data', 'end', 'error'));
+ Util::forwardEvents($this->writable, $this, array('drain', 'error', 'pipe'));
+
+ $this->readable->on('close', array($this, 'close'));
+ $this->writable->on('close', array($this, 'close'));
+ }
+
+ public function isReadable()
+ {
+ return $this->readable->isReadable();
+ }
+
+ public function pause()
+ {
+ $this->readable->pause();
+ }
+
+ public function resume()
+ {
+ if (!$this->writable->isWritable()) {
+ return;
+ }
+
+ $this->readable->resume();
+ }
+
+ public function pipe(WritableStreamInterface $dest, array $options = array())
+ {
+ return Util::pipe($this, $dest, $options);
+ }
+
+ public function isWritable()
+ {
+ return $this->writable->isWritable();
+ }
+
+ public function write($data)
+ {
+ return $this->writable->write($data);
+ }
+
+ public function end($data = null)
+ {
+ $this->readable->pause();
+ $this->writable->end($data);
+ }
+
+ public function close()
+ {
+ if ($this->closed) {
+ return;
+ }
+
+ $this->closed = true;
+ $this->readable->close();
+ $this->writable->close();
+
+ $this->emit('close');
+ $this->removeAllListeners();
+ }
+}