summaryrefslogtreecommitdiffstats
path: root/vendor/gipfl/stream/src
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-14 13:23:16 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-14 13:23:16 +0000
commit3e97c51418e6d27e9a81906f347fcb7c78e27d4f (patch)
treeee596ce1bc9840661386f96f9b8d1f919a106317 /vendor/gipfl/stream/src
parentInitial commit. (diff)
downloadicingaweb2-module-incubator-3e97c51418e6d27e9a81906f347fcb7c78e27d4f.tar.xz
icingaweb2-module-incubator-3e97c51418e6d27e9a81906f347fcb7c78e27d4f.zip
Adding upstream version 0.20.0.upstream/0.20.0
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'vendor/gipfl/stream/src')
-rw-r--r--vendor/gipfl/stream/src/BufferedLineReader.php100
1 files changed, 100 insertions, 0 deletions
diff --git a/vendor/gipfl/stream/src/BufferedLineReader.php b/vendor/gipfl/stream/src/BufferedLineReader.php
new file mode 100644
index 0000000..bd43155
--- /dev/null
+++ b/vendor/gipfl/stream/src/BufferedLineReader.php
@@ -0,0 +1,100 @@
+<?php
+
+namespace gipfl\Stream;
+
+use Evenement\EventEmitterTrait;
+use React\EventLoop\LoopInterface;
+use React\Stream\WritableStreamInterface;
+use function strlen;
+use function strpos;
+use function substr;
+
+class BufferedLineReader implements WritableStreamInterface
+{
+ use EventEmitterTrait;
+
+ /** @var LoopInterface */
+ protected $loop;
+
+ protected $buffer = '';
+
+ protected $writable = true;
+
+ /** @var string */
+ protected $separator;
+
+ /** @var int */
+ protected $separatorLength;
+
+ protected $process;
+
+ // protected $maxBufferSize; // Not yet. Do we need this?
+
+ /**
+ * @param string $separator
+ * @param LoopInterface $loop
+ */
+ public function __construct($separator, LoopInterface $loop)
+ {
+ $this->loop = $loop;
+ $this->separator = $separator;
+ $this->separatorLength = strlen($separator);
+ $this->process = function () {
+ $this->processBuffer();
+ };
+ }
+
+ protected function processBuffer()
+ {
+ $lastPos = 0;
+ while (false !== ($pos = strpos($this->buffer, $this->separator, $lastPos))) {
+ $this->emit('line', [substr($this->buffer, $lastPos, $pos - $lastPos)]);
+ $lastPos = $pos + $this->separatorLength;
+ }
+ if ($lastPos !== 0) {
+ $this->buffer = substr($this->buffer, $lastPos);
+ }
+ }
+
+ public function isWritable()
+ {
+ return $this->writable;
+ }
+
+ public function write($data)
+ {
+ if (! $this->writable) {
+ return false;
+ }
+ $this->buffer .= $data;
+ if (strpos($data, $this->separator) !== false) {
+ $this->loop->futureTick($this->process);
+ }
+
+ return true;
+ }
+
+ public function end($data = null)
+ {
+ if ($data !== null) {
+ $this->buffer .= $data;
+ }
+ $this->close();
+ }
+
+ public function close()
+ {
+ $this->writable = false;
+ $this->processBuffer();
+ $remainingBuffer = $this->buffer;
+ $this->buffer = '';
+ if ($length = strlen($remainingBuffer)) {
+ $this->emit('error', [new \Exception(sprintf(
+ 'There are %d unprocessed bytes in our buffer: %s',
+ $length,
+ substr($remainingBuffer, 0, 64)
+ ))]);
+ }
+ $this->emit('close');
+ }
+}