Initial commit

src/NonBlockingInputInterface.php

@@ -0,0 +1,16 @@
+<?php
+
+namespace JoopSchilder\React\Stream\NonBlockingInput;
+
+interface NonBlockingInputInterface
+{
+
+	function open(): void;
+
+
+	function select(): ?PayloadInterface;
+	
+
+	function close(): void;
+
+}

src/PayloadInterface.php

@@ -0,0 +1,10 @@
+<?php
+
+namespace JoopSchilder\React\Stream\NonBlockingInput;
+
+/**
+ * Marker interface.
+ */
+interface PayloadInterface
+{
+}

src/ReadableNonBlockingInputStream.php

@@ -0,0 +1,140 @@
+<?php
+
+namespace JoopSchilder\React\Stream\NonBlockingInput;
+
+use Evenement\EventEmitter;
+use JoopSchilder\React\Stream\NonBlockingInput\ValueObject\PollingInterval;
+use React\EventLoop\LoopInterface;
+use React\EventLoop\TimerInterface;
+use React\Stream\ReadableResourceStream;
+use React\Stream\ReadableStreamInterface;
+use React\Stream\Util;
+use React\Stream\WritableStreamInterface;
+use RuntimeException;
+use Throwable;
+
+/**
+ * @see ReadableStreamInterface
+ * @see ReadableResourceStream
+ */
+final class ReadableNonBlockingInputStream extends EventEmitter implements ReadableStreamInterface
+{
+	private NonBlockingInputInterface $input;
+
+	private LoopInterface $loop;
+
+	private PollingInterval $interval;
+
+	private ?TimerInterface $periodicTimer = null;
+
+	private bool $closed = false;
+
+	private bool $listening = false;
+
+
+	/**
+	 * @see ReadableStreamInterface
+	 * @see ReadableResourceStream for an example
+	 */
+	public function __construct(NonBlockingInputInterface $input, LoopInterface $loop, ?PollingInterval $interval = null)
+	{
+		$this->input = $input;
+		$this->loop = $loop;
+		$this->interval = $interval ?? new PollingInterval();
+
+		$this->resume();
+	}
+
+
+	public function isReadable()
+	{
+		return !$this->closed;
+	}
+
+
+	/**
+	 * @see ReadableResourceStream::pause()
+	 * Pause consumption of the AMQP queue but do not mark the stream as closed
+	 */
+	public function pause()
+	{
+		if (!$this->listening) {
+			return;
+		}
+
+		$this->input->close();
+		$this->loop->cancelTimer($this->periodicTimer);
+		$this->listening = false;
+	}
+
+
+	/**
+	 * @see ReadableResourceStream::resume()
+	 * Register the consumer with the broker and add consumer again
+	 */
+	public function resume()
+	{
+		if ($this->listening || $this->closed) {
+			return;
+		}
+
+		$this->input->open();
+		$this->periodicTimer = $this->loop->addPeriodicTimer(
+			$this->interval->getInterval(),
+			function () {
+				if ($data = $this->read()) {
+					$this->emit('data', [$data]);
+				}
+			}
+		);
+
+		$this->listening = true;
+	}
+
+
+	/**
+	 * @param WritableStreamInterface $dest
+	 * @param array $options
+	 * @return WritableStreamInterface
+	 * @see ReadableResourceStream::pipe()
+	 */
+	public function pipe(WritableStreamInterface $dest, array $options = [])
+	{
+		return Util::pipe($this, $dest, $options);
+	}
+
+
+	/**
+	 * @see ReadableResourceStream::close()
+	 */
+	public function close()
+	{
+		if ($this->closed) {
+			return;
+		}
+
+		$this->closed = true;
+
+		$this->emit('close');
+		$this->pause();
+		$this->removeAllListeners();
+
+		$this->input->close();
+	}
+
+
+	private function read(): ?PayloadInterface
+	{
+		try {
+			return $this->input->select();
+		} catch (Throwable $t) {
+			$this->emit('error', [
+				new RuntimeException('Unable to read data from input: ' . $t->getMessage(), 0, $t),
+			]);
+			$this->close();
+		}
+
+		return null;
+	}
+
+}

src/ValueObject/PollingInterval.php

@@ -0,0 +1,29 @@
+<?php
+
+namespace JoopSchilder\React\Stream\NonBlockingInput\ValueObject;
+
+use InvalidArgumentException;
+
+final class PollingInterval
+{
+	private const DEFAULT_INTERVAL = 0.05;
+
+	private float $interval;
+
+
+	public function __construct(float $interval = self::DEFAULT_INTERVAL)
+	{
+		if ($interval < 0.0) {
+			throw new InvalidArgumentException('Interval must be greater than 0');
+		}
+
+		$this->interval = $interval;
+	}
+
+
+	public function getInterval(): float
+	{
+		return $this->interval;
+	}
+
+}