1
0
mirror of https://github.com/danog/amp.git synced 2024-12-11 08:59:46 +01:00
amp/lib/Internal/Producer.php
Aaron Piotrowski 0eceb48fad
Refactor internal traits as classes
Trait tests should test Deferred and Emitter instead, will update with other tests.
2020-09-26 23:14:17 -05:00

210 lines
5.6 KiB
PHP

<?php
namespace Amp\Internal;
use Amp\Deferred;
use Amp\Failure;
use Amp\Promise;
use Amp\Success;
use React\Promise\PromiseInterface as ReactPromise;
/**
* 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.
* 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.
*
* @internal
* @template-covariant TValue
*/
final class Producer
{
private ?Promise $complete = null;
private array $values = [];
private array $backPressure = [];
private int $consumePosition = -1;
private int $emitPosition = -1;
private ?Deferred $waiting = null;
private ?array $resolutionTrace = null;
/**
* {@inheritdoc}
*
* @return Promise<bool>
*/
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;
}
$this->waiting = new Deferred;
return $this->waiting->promise();
}
/**
* {@inheritdoc}
*
* @return TValue
*/
public function getCurrent()
{
if (empty($this->values) && $this->complete) {
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];
}
public function isComplete(): bool
{
return (bool) $this->complete;
}
/**
* Emits a value from the iterator. The returned promise is resolved once the emitted value has been consumed.
*
* @param mixed $value
*
* @return Promise
* @psalm-return Promise<null>
*
* @throws \Error If the iterator has completed.
*/
public function emit($value): Promise
{
if ($this->complete) {
throw new \Error("Iterators cannot emit values after calling complete");
}
if ($value instanceof ReactPromise) {
$value = Promise\adapt($value);
}
if ($value instanceof Promise) {
$deferred = new Deferred;
$value->onResolve(function ($e, $v) use ($deferred) {
if ($this->complete) {
$deferred->fail(
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));
});
return $deferred->promise();
}
$position = ++$this->emitPosition;
$this->values[$position] = $value;
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.
}
$this->backPressure[$position] = $pressure = new Deferred;
return $pressure->promise();
}
/**
* Completes the iterator.
*
* @return void
*
* @throws \Error If the iterator has already been completed.
*/
public 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 {
// @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 () {
if (isDebugEnabled()) {
$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
*/
public function fail(\Throwable $exception)
{
$this->complete = new Failure($exception);
if ($this->waiting !== null) {
$waiting = $this->waiting;
$this->waiting = null;
$waiting->resolve($this->complete);
}
}
}