diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 12:38:42 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 12:38:42 +0000 |
commit | c3ca98e1b35123f226c7f4c596b5dee78caa4223 (patch) | |
tree | 9b6eb109283da55e7d9064baa9fac795a40264cb /vendor/react/stream/src/CompositeStream.php | |
parent | Initial commit. (diff) | |
download | icinga-php-thirdparty-upstream.tar.xz icinga-php-thirdparty-upstream.zip |
Adding upstream version 0.11.0.upstream/0.11.0upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'vendor/react/stream/src/CompositeStream.php')
-rw-r--r-- | vendor/react/stream/src/CompositeStream.php | 83 |
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(); + } +} |