1
0
mirror of https://github.com/danog/amp.git synced 2024-11-27 04:24:42 +01:00
amp/lib/Listener.php

206 lines
5.5 KiB
PHP
Raw Normal View History

<?php
2016-08-16 06:46:26 +02:00
2016-05-24 18:47:14 +02:00
namespace Amp;
use AsyncInterop\Promise;
2016-08-11 21:35:58 +02:00
2016-06-01 18:37:12 +02:00
/**
2017-01-04 02:10:27 +01:00
* Asynchronous iterator that can be used within a coroutine to iterate over the emitted values from an Stream.
2016-06-01 18:37:12 +02:00
*
* Example:
2017-01-04 02:10:27 +01:00
* $listener = new Listener($stream); // $stream is an instance of \Amp\Stream
* while (yield $listener->advance()) {
* $emitted = $listener->getCurrent();
2016-06-01 18:37:12 +02:00
* }
2017-01-04 02:10:27 +01:00
* $result = $listener->getResult();
2016-06-01 18:37:12 +02:00
*/
2017-01-04 02:10:27 +01:00
class Listener {
/** @var \Amp\Stream */
private $stream;
2016-12-11 16:17:51 +01:00
2016-08-18 05:25:54 +02:00
/** @var mixed[] */
2016-05-30 06:49:44 +02:00
private $values = [];
2016-08-18 05:25:54 +02:00
/** @var \Amp\Deferred[] */
private $backPressure = [];
2016-05-29 18:35:09 +02:00
2016-08-18 05:25:54 +02:00
/** @var int */
2016-05-30 06:49:44 +02:00
private $position = -1;
2016-08-18 05:25:54 +02:00
/** @var \Amp\Deferred|null */
private $waiting;
2016-05-29 18:35:09 +02:00
2016-08-18 05:25:54 +02:00
/** @var bool */
2016-05-30 17:24:03 +02:00
private $resolved = false;
2016-05-29 18:35:09 +02:00
2016-08-18 05:25:54 +02:00
/** @var mixed */
2016-05-30 06:49:44 +02:00
private $result;
2016-05-29 18:35:09 +02:00
2016-08-18 05:25:54 +02:00
/** @var \Throwable|null */
2016-05-29 18:35:09 +02:00
private $exception;
2016-05-27 22:44:01 +02:00
/**
2017-01-04 02:10:27 +01:00
* @param \Amp\Stream $stream
2016-05-27 22:44:01 +02:00
*/
2017-01-04 02:10:27 +01:00
public function __construct(Stream $stream) {
$this->stream = $stream;
2016-12-11 16:17:51 +01:00
$waiting = &$this->waiting;
2016-11-14 20:59:21 +01:00
$values = &$this->values;
$backPressure = &$this->backPressure;
2016-11-14 20:59:21 +01:00
$resolved = &$this->resolved;
2016-05-29 18:35:09 +02:00
$this->stream->listen(static function ($value) use (&$waiting, &$values, &$backPressure, &$resolved) {
2016-05-30 06:49:44 +02:00
$values[] = $value;
$backPressure[] = $pressure = new Deferred;
2016-05-29 18:35:09 +02:00
if ($waiting !== null) {
$deferred = $waiting;
$waiting = null;
$deferred->resolve(true);
2016-05-30 06:49:44 +02:00
}
2016-05-29 18:35:09 +02:00
if ($resolved) {
return null;
}
2016-11-14 20:59:21 +01:00
return $pressure->promise();
2016-05-29 18:35:09 +02:00
});
2016-08-18 05:25:54 +02:00
$result = &$this->result;
2016-11-14 20:59:21 +01:00
$error = &$this->exception;
2016-05-29 18:35:09 +02:00
$this->stream->when(static function ($exception, $value) use (&$waiting, &$result, &$error, &$resolved) {
2016-05-30 17:24:03 +02:00
$resolved = true;
2016-05-29 18:35:09 +02:00
if ($exception) {
2016-05-30 06:49:44 +02:00
$result = null;
2016-05-29 18:35:09 +02:00
$error = $exception;
if ($waiting !== null) {
$waiting->fail($exception);
2016-05-29 18:35:09 +02:00
}
return;
}
2016-05-30 06:49:44 +02:00
$result = $value;
if ($waiting !== null) {
$waiting->resolve(false);
2016-05-29 18:35:09 +02:00
}
});
2016-05-27 22:44:01 +02:00
}
/**
2017-01-04 02:10:27 +01:00
* Marks the listener as resolved to relieve back-pressure on the stream.
2016-05-27 22:44:01 +02:00
*/
public function __destruct() {
2016-08-18 04:11:03 +02:00
$this->resolved = true;
2016-12-11 16:17:51 +01:00
foreach ($this->backPressure as $deferred) {
2016-08-18 05:25:54 +02:00
$deferred->resolve();
2016-05-29 18:35:09 +02:00
}
2016-05-27 22:44:01 +02:00
}
2016-12-11 16:17:51 +01:00
2016-05-24 18:47:14 +02:00
/**
2017-01-04 02:10:27 +01:00
* Succeeds with true if an emitted value is available by calling getCurrent() or false if the stream has
* resolved. If the stream fails, the returned promise will fail with the same exception.
2016-05-24 18:47:14 +02:00
*
* @return \AsyncInterop\Promise<bool>
*
* @throws \Error If the prior promise returned from this method has not resolved.
2016-05-24 18:47:14 +02:00
*/
public function advance(): Promise {
if ($this->waiting !== null) {
throw new \Error("The prior promise returned must resolve before invoking this method again");
}
if (isset($this->backPressure[$this->position])) {
$future = $this->backPressure[$this->position];
unset($this->values[$this->position], $this->backPressure[$this->position]);
2016-05-30 06:49:44 +02:00
$future->resolve();
}
++$this->position;
2016-08-23 20:18:56 +02:00
if (\array_key_exists($this->position, $this->values)) {
2016-05-30 06:49:44 +02:00
return new Success(true);
}
2016-05-30 17:24:03 +02:00
if ($this->resolved) {
2016-05-29 18:35:09 +02:00
if ($this->exception) {
return new Failure($this->exception);
}
return new Success(false);
}
$this->waiting = new Deferred;
return $this->waiting->promise();
2016-05-27 22:44:01 +02:00
}
2016-05-24 18:47:14 +02:00
/**
2017-01-04 02:10:27 +01:00
* Gets the last emitted value or throws an exception if the stream has completed.
2016-05-24 18:47:14 +02:00
*
2017-01-04 02:10:27 +01:00
* @return mixed Value emitted from stream.
2016-05-24 18:47:14 +02:00
*
2017-01-04 02:10:27 +01:00
* @throws \Error If the stream has resolved or advance() was not called before calling this method.
2016-05-24 18:47:14 +02:00
*/
2016-05-27 22:44:01 +02:00
public function getCurrent() {
2016-05-30 17:24:03 +02:00
if (empty($this->values) && $this->resolved) {
2017-01-04 02:10:27 +01:00
throw new \Error("The stream has resolved");
2016-05-29 18:35:09 +02:00
}
2016-08-23 20:18:56 +02:00
if (!\array_key_exists($this->position, $this->values)) {
throw new \Error("Promise returned from advance() must resolve before calling this method");
2016-05-29 18:35:09 +02:00
}
2016-05-30 06:49:44 +02:00
return $this->values[$this->position];
2016-05-27 22:44:01 +02:00
}
2016-05-24 18:47:14 +02:00
/**
2017-01-04 02:10:27 +01:00
* Gets the result of the stream or throws the failure reason. Also throws an exception if the stream has
2016-05-30 17:24:03 +02:00
* not completed.
2016-05-24 18:47:14 +02:00
*
2017-01-04 02:10:27 +01:00
* @return mixed Final return value of the stream.
2016-05-24 18:47:14 +02:00
*
2017-01-04 02:10:27 +01:00
* @throws \Error If the stream has not completed.
* @throws \Throwable The exception used to fail the stream.
2016-05-24 18:47:14 +02:00
*/
2016-05-30 17:24:03 +02:00
public function getResult() {
if (!$this->resolved) {
2017-01-04 02:10:27 +01:00
throw new \Error("The stream has not resolved");
2016-05-29 18:35:09 +02:00
}
if ($this->exception) {
throw $this->exception;
}
2016-05-30 06:49:44 +02:00
return $this->result;
2016-05-27 22:44:01 +02:00
}
2016-12-11 16:17:51 +01:00
2016-08-23 15:50:04 +02:00
/**
2017-01-04 02:10:27 +01:00
* Returns an array of values that were not consumed by the listener before the Stream completed.
2016-08-23 15:50:04 +02:00
*
* @return array Unconsumed emitted values.
*
2017-01-04 02:10:27 +01:00
* @throws \Error If the stream has not completed.
2016-08-23 15:50:04 +02:00
*/
2016-08-23 16:21:05 +02:00
public function drain(): array {
2016-08-23 15:50:04 +02:00
if (!$this->resolved) {
2017-01-04 02:10:27 +01:00
throw new \Error("The stream has not resolved");
2016-08-23 15:50:04 +02:00
}
2016-12-11 16:17:51 +01:00
unset($this->values[$this->position]);
2016-08-23 15:50:04 +02:00
$values = $this->values;
$this->values = [];
2016-12-11 16:17:51 +01:00
$deferreds = $this->backPressure;
$this->backPressure = [];
2016-08-23 15:50:04 +02:00
foreach ($deferreds as $deferred) {
$deferred->resolve();
}
2016-12-11 16:17:51 +01:00
2016-08-23 15:50:04 +02:00
return $values;
}
2016-05-24 18:47:14 +02:00
}