1
0
mirror of https://github.com/danog/amp.git synced 2024-12-13 01:47:33 +01:00
amp/lib/Internal/Producer.php

205 lines
5.4 KiB
PHP
Raw Normal View History

<?php
2016-08-16 06:46:26 +02:00
2016-05-27 01:20:05 +02:00
namespace Amp\Internal;
use Amp\Deferred;
use Amp\Failure;
use Amp\Promise;
use Amp\Success;
2017-02-20 21:53:58 +01:00
use React\Promise\PromiseInterface as ReactPromise;
2016-05-27 01:20:05 +02:00
2016-06-01 18:37:12 +02:00
/**
* Trait used by Iterator implementations. Do not use this trait in your code, instead compose your class from one of
* the available classes implementing \Amp\Iterator.
2017-01-04 02:10:27 +01:00
* Note that it is the responsibility of the user of this trait to ensure that listeners have a chance to listen first
* before emitting values.
2016-06-01 18:37:12 +02:00
*
* @internal
2020-03-28 14:25:39 +01:00
* @template-covariant TValue
2016-06-01 18:37:12 +02:00
*/
2018-06-18 20:00:01 +02:00
trait Producer
{
2020-09-24 18:52:22 +02:00
private ?Promise $complete;
2020-09-24 18:52:22 +02:00
private array $values = [];
2020-09-24 18:52:22 +02:00
private array $backPressure = [];
2020-09-24 18:52:22 +02:00
private int $consumePosition = -1;
2020-09-24 18:52:22 +02:00
private int $emitPosition = -1;
2020-09-24 18:52:22 +02:00
private ?Deferred $waiting;
2016-05-29 18:35:09 +02:00
2020-09-24 18:52:22 +02:00
private ?array $resolutionTrace;
2016-05-27 01:20:05 +02:00
/**
* {@inheritdoc}
2020-03-28 14:25:39 +01:00
*
* @return Promise<bool>
2016-05-27 01:20:05 +02:00
*/
2018-06-18 20:00:01 +02:00
public function advance(): Promise
{
if ($this->waiting !== null) {
throw new \Error("The prior promise returned must resolve before invoking this method again");
}
unset($this->values[$this->consumePosition]);
$position = ++$this->consumePosition;
if (\array_key_exists($position, $this->values)) {
\assert(isset($this->backPressure[$position]));
$deferred = $this->backPressure[$position];
unset($this->backPressure[$position]);
$deferred->resolve();
return new Success(true);
}
if ($this->complete) {
return $this->complete;
2016-05-29 18:35:09 +02:00
}
$this->waiting = new Deferred;
return $this->waiting->promise();
}
/**
* {@inheritdoc}
2020-03-28 14:25:39 +01:00
*
* @return TValue
*/
2018-06-18 20:00:01 +02:00
public function getCurrent()
{
if (empty($this->values) && $this->complete) {
2017-04-28 07:51:59 +02:00
throw new \Error("The iterator has completed");
}
if (!\array_key_exists($this->consumePosition, $this->values)) {
throw new \Error("Promise returned from advance() must resolve before calling this method");
}
return $this->values[$this->consumePosition];
2016-05-27 01:20:05 +02:00
}
2016-05-29 18:35:09 +02:00
/**
* Emits a value from the iterator. The returned promise is resolved once the emitted value has been consumed.
2016-05-27 22:44:01 +02:00
*
* @param mixed $value
*
* @return Promise
* @psalm-return Promise<null>
2016-05-29 18:35:09 +02:00
*
2017-04-28 07:51:59 +02:00
* @throws \Error If the iterator has completed.
2016-05-27 01:20:05 +02:00
*/
2018-06-18 20:00:01 +02:00
private function emit($value): Promise
{
if ($this->complete) {
2017-04-28 07:51:59 +02:00
throw new \Error("Iterators cannot emit values after calling complete");
2016-05-27 01:20:05 +02:00
}
2017-02-20 21:53:58 +01:00
if ($value instanceof ReactPromise) {
$value = Promise\adapt($value);
2017-02-20 21:53:58 +01:00
}
2016-11-14 20:59:21 +01:00
if ($value instanceof Promise) {
$deferred = new Deferred;
$value->onResolve(function ($e, $v) use ($deferred) {
if ($this->complete) {
$deferred->fail(
2017-04-28 07:51:59 +02:00
new \Error("The iterator was completed before the promise result could be emitted")
);
return;
}
if ($e) {
$this->fail($e);
$deferred->fail($e);
return;
}
$deferred->resolve($this->emit($v));
2016-08-16 09:28:47 +02:00
});
2016-11-14 20:59:21 +01:00
return $deferred->promise();
2016-08-16 09:28:47 +02:00
}
$position = ++$this->emitPosition;
$this->values[$position] = $value;
2016-08-16 09:28:47 +02:00
if ($this->waiting !== null) {
$waiting = $this->waiting;
$this->waiting = null;
$waiting->resolve(true);
return new Success; // Consumer was already waiting for a new value, so back-pressure is unnecessary.
2016-08-16 09:28:47 +02:00
}
$this->backPressure[$position] = $pressure = new Deferred;
return $pressure->promise();
2016-05-27 01:20:05 +02:00
}
/**
* Completes the iterator.
2016-05-27 22:44:01 +02:00
*
* @return void
*
2017-04-28 07:51:59 +02:00
* @throws \Error If the iterator has already been completed.
2016-05-27 01:20:05 +02:00
*/
2018-06-18 20:00:01 +02:00
private function complete()
{
if ($this->complete) {
$message = "Iterator has already been completed";
if (isset($this->resolutionTrace)) {
$trace = formatStacktrace($this->resolutionTrace);
$message .= ". Previous completion trace:\n\n{$trace}\n\n";
} else {
2017-12-02 16:54:28 +01:00
// @codeCoverageIgnoreStart
$message .= ", define environment variable AMP_DEBUG or const AMP_DEBUG = true and enable assertions "
. "for a stacktrace of the previous resolution.";
// @codeCoverageIgnoreEnd
}
throw new \Error($message);
}
\assert((function () {
2020-07-11 17:13:51 +02:00
if (isDebugEnabled()) {
2017-12-02 16:54:28 +01:00
$trace = \debug_backtrace(\DEBUG_BACKTRACE_IGNORE_ARGS);
\array_shift($trace); // remove current closure
$this->resolutionTrace = $trace;
}
return true;
})());
$this->complete = new Success(false);
if ($this->waiting !== null) {
$waiting = $this->waiting;
$this->waiting = null;
$waiting->resolve($this->complete);
}
}
/**
* @param \Throwable $exception
*
* @return void
*/
2018-06-18 20:00:01 +02:00
private function fail(\Throwable $exception)
{
$this->complete = new Failure($exception);
if ($this->waiting !== null) {
$waiting = $this->waiting;
$this->waiting = null;
$waiting->resolve($this->complete);
}
2016-05-27 01:20:05 +02:00
}
}