acrylic-origami / ampreactor Goto Github PK

In the case that hot consumers are synchronous, the first hot consumer will get all the values and the rest won't get any. This is because internal to Internal\Producer::advance, hot consumers implicitly race to get control from the Internal\Producer::$waiting promise. However, promises handle multiple subscribers via a queue, not randomly. So, when the first consumer gets control and handles the item synchronously, it will continue to receive control first in this race and consume all the elements, starving the rest of the consumers.

See for yourself:

<?php
use AmpReactor\InteractiveProducer;
use Amp\Promise;
use Amp\Coroutine;
use function Amp\Promise\all;
use function Amp\Promise\wait;
$source = InteractiveProducer::from_producerish(function($emitter) { for($i = 0; $i < 10; $i++) $emitter($i); if(false) yield null; });
wait(all([
  new Coroutine((function() use ($source) { while(yield $source->advance()) echo 'Consumer 1: '.$source->getCurrent()."\n"; })()),
  new Coroutine((function() use ($source) { while(yield $source->advance()) echo 'Consumer 2: '.$source->getCurrent()."\n"; })())
]));