<commit_before><?php

namespace Amp\Internal;

use Amp\Coroutine;
use Amp\Future;
use Amp\Observable;
use Amp\Subscriber;
use Amp\Success;
use Amp\UnsubscribedException;
use Interop\Async\Awaitable;
use Interop\Async\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.
 *
 * @internal
 */
trait Producer {
    use Placeholder {
        resolve as complete;
    }

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

    /**
     * @var \Amp\Future|null
     */
    private $waiting;

    /**
     * @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.
     */
    public function init()
    {
        $this->waiting = new Future;
        $this->unsubscribe = function ($id, $exception = null) {
            $this->unsubscribe($id, $exception);
        };
    }

    /**
     * @param callable $onNext
     *
     * @return \Amp\Subscriber
     */
    public function subscribe(callable $onNext) {
        if ($this->result !== null) {
            return new Subscriber(
                $this->nextId++,
                $this->result instanceof Awaitable ? $this->result : new Success($this->result),
                $this->unsubscribe
            );
        }

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

        if ($this->waiting !== null) {
            $waiting = $this->waiting;
            $this->waiting = null;
            $waiting->resolve();
        }

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

    /**
     * @param string $id
     * @param \Throwable|\Exception|null $exception
     */
    protected function unsubscribe($id, $exception = null) {
        if (!isset($this->futures[$id])) {
            return;
        }

        $future = $this->futures[$id];
        unset($this->subscribers[$id], $this->futures[$id]);

        if (empty($this->subscribers)) {
            $this->waiting = new Future;
        }<commit_msg>Fix bug when emitting failed awaitable and waiting for a subscriber<commit_after><?php

namespace Amp\Internal;

use Amp\Coroutine;
use Amp\Future;
use Amp\Observable;
use Amp\Subscriber;
use Amp\Success;
use Amp\UnsubscribedException;
use Interop\Async\Awaitable;
use Interop\Async\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.
 *
 * @internal
 */
trait Producer {
    use Placeholder {
        resolve as complete;
    }

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

    /**
     * @var \Amp\Future|null
     */
    private $waiting;

    /**
     * @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.
     */
    public function init()
    {
        $this->waiting = new Future;
        $this->unsubscribe = function ($id, $exception = null) {
            $this->unsubscribe($id, $exception);
        };
    }

    /**
     * @param callable $onNext
     *
     * @return \Amp\Subscriber
     */
    public function subscribe(callable $onNext) {
        if ($this->result !== null) {
            return new Subscriber(
                $this->nextId++,
                $this->result instanceof Awaitable ? $this->result : new Success($this->result),
                $this->unsubscribe
            );
        }

        $id = $this->nextId++;
        $this->futures[$id] = $future = new Future;
        $this->subscribers[$id] = $onNext;

        if ($this->waiting !== null) {
            $waiting = $this->waiting;
            $this->waiting = null;
            $waiting->resolve();
        }

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

    /**
     * @param string $id
     * @param \Throwable|\Exception|null $exception
     */
    protected function unsubscribe($id, $exception = null) {
        if (!isset($this->futures[$id])) {
            return;
        }

        $future = $this->futures[$id];
        unset($this->subscribers[$id], $this->futures[$id]);

        if (empty($this->subscribers)) {
            $this->waiting = new Future;
        }
{'fraction_non_alphanumeric': 0.1445246179966044, 'fraction_numerical': 0.0, 'mean_word_length': 1.960427135678392, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 50, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3679186', 'n_tokens_mistral': 1447, 'n_tokens_neox': 1411, 'n_words': 339}