amp/lib/Internal/Producer.php

<?php

declare(strict_types=1);

namespace Amp\Internal;

use Amp\{ Deferred, Observable, Subscriber, Success};
use Interop\Async\{ Awaitable, Loop };

/**
 * Trait used by Observable implementations. Do not use this trait in your code, instead compose your class from one of
 * the available classes implementing \Amp\Observable.
 * Note that it is the responsibility of the user of this trait to ensure that subscribers have a chance to subscribe first
 * before emitting values.
 *
 * @internal
 */
trait Producer {
    use Placeholder {
        resolve as complete;
    }

    /**
     * @var callable[]
     */
    private $subscribers = [];

    /**
     * @var \Amp\Future[]
     */
    private $futures = [];

    /**
     * @var string
     */
    private $nextId = "a";

    /**
     * @var callable
     */
    private $unsubscribe;

    /**
     * Initializes the trait. Use as constructor or call within using class constructor.
     */
    private function init() {
		$this->unsubscribe = function ($id) {
			$this->unsubscribe($id);
		};
    }

    /**
     * @param callable $onNext
     *
     * @return \Amp\Subscriber
     */
    public function subscribe(callable $onNext): Subscriber {
        if ($this->resolved) {
            return new Subscriber($this->nextId++, function() {});
        }

        $id = $this->nextId++;
        $this->subscribers[$id] = $onNext;

        return new Subscriber($id, $this->unsubscribe);
    }

    /**
     * @param string $id
     */
    private function unsubscribe(string $id) {
        if (!isset($this->subscribers[$id])) {
            return;
        }

        unset($this->subscribers[$id]);
    }

    /**
     * Emits a value from the observable. The returned awaitable is resolved with the emitted value once all subscribers
     * have been invoked.
     *
     * @param mixed $value
     *
     * @return \Interop\Async\Awaitable
     *
     * @throws \Error If the observable has resolved.
     */
    private function emit($value): Awaitable {
        if ($this->resolved) {
            throw new \Error("The observable has been resolved; cannot emit more values");
        }

		if ($value instanceof Awaitable) {
			if ($value instanceof Observable) {
				$value->subscribe(function ($value) {
					return $this->emit($value);
				});
				return $value;
			}

			$value->when(function($e, $val) {
				if (!$e) {
					$this->emit($val);
				} elseif (!$this->resolved) {
					$this->fail($e);
				}
			});
			return $value;
		}

		$awaitables = [];

		foreach ($this->subscribers as $onNext) {
			try {
				$result = $onNext($value);
				if ($result instanceof Awaitable) {
					$awaitables[] = $result;
				}
			} catch (\Throwable $e) {
				Loop::defer(static function () use ($e) {
					throw $e;
				});
			}
		}

		if (!$awaitables) {
			return new Success($value);
		}

		$deferred = new Deferred;
		$count = count($awaitables);
		$f = function ($e) use ($deferred, $value, &$count) {
			if ($e) {
				Loop::defer(static function () use ($e) {
					throw $e;
				});
			}
			if (!--$count) {
				$deferred->resolve($value);
			}
		};

		foreach ($awaitables as $awaitable) {
			$awaitable->when($f);
		}

		return $deferred;
    }


    /**
     * Resolves the observable with the given value.
     *
     * @param mixed $value
     *
     * @throws \Error If the observable has already been resolved.
     */
    private function resolve($value = null) {
        $this->complete($value);

        $this->subscribers = [];
		$this->unsubscribe = null;
    }
}