Fungsi pada async amphp tidak mau lanjut ke fungsi berikutnya jika pake `Emitter` dan lanjut jika pake `Amp\Iterator\map`

.env

DB_HOST=127.0.0.1
DB_USER=user
DB_PASSWD=password
DB_NAME=test

AmpDb.php

<?php
declare(strict_types=1);

require __DIR__ . '/vendor/autoload.php';

use Amp\Mysql\ConnectionConfig;
use Dotenv\Dotenv;

final class AmpDb
{
    private $conn = null;

    public function __construct(string $dotName = '.env')
    {
        $dotenv = Dotenv::createImmutable(__DIR__, $dotName);
        $dotenv->safeLoad();
        return $this;
    }

    public function getPool()
    {
        return $this->conn;
    }

    public function createConn()
    {
        $connStr = 'host={host} user={user} password={password} db={dbname}';
        $dsn = strtr($connStr, [
            '{host}' => $_SERVER['DB_HOST'],
            '{user}' => $_SERVER['DB_USER'],
            '{password}' => $_SERVER['DB_PASSWD'],
            '{dbname}' => $_SERVER['DB_NAME']
        ]);
        $config = ConnectionConfig::fromString($dsn);

        $this->conn = Amp\Mysql\pool($config);
        return $this;
    }
}

emitter.php

<?php
declare(strict_types=1);

require __DIR__ . '/vendor/autoload.php';
require __DIR__ . '/AmpDb.php';

use Amp\Emitter;
use Amp\Loop;
use function Amp\call;

final class Hello
{
    public function run()
    {
        Loop::run([$this, 'main']);
    }
    
    public function main()
    {
        // $pool adalah hasil koneksi db        
        $pool = (new AmpDb())->createConn()->getPool();
        $user = yield $this->getUsers($pool);
        while (yield $user->advance()) {
            print_r($user->getCurrent());
        }
        
        $this->iters();
    }
    
    public function iters()
    {
        return call(function () {
            try {
                $emitter = new Emitter();
                $data = [1,2,3];
                foreach ($data as $val) {
                    $emitter->emit($val);
                }

                $iters = $emitter->iterate();
                while (yield $iters->advance()) {
                    echo $iters->getCurrent(), PHP_EOL;
                }

            } catch (\Throwable $ex) {
                \printf("Exception: %s\n", (string)$ex);
            }
        });
    }

    public function getUsers($pool)
    {
        return call(function () use ($pool) {
            $emitter = new Emitter();
            $stm = yield $pool->prepare('SELECT * FROM users limit 5');

            $result = yield $stm->execute([]);
            while (yield $result->advance()) {
                $row = $result->getCurrent();
                $emitter->emit($row['user_id']);
            }
            return $emitter->iterate();
        });
    }
}

(new Hello())->run();

map-iterator.php

<?php
declare(strict_types=1);

require __DIR__ . '/vendor/autoload.php';
require __DIR__ . '/AmpDb.php';

use Amp\Emitter;
use Amp\Loop;
use function Amp\call;

final class Hello
{
    public function run()
    {
      Loop::run([$this, 'main']);
    }
    
    public function main()
    {
        // $pool adalah hasil koneksi db
        $pool = (new AmpDb())->createConn()->getPool();
        $user = yield $this->getUsers($pool);
        while (yield $user->advance()) {
            print_r($user->getCurrent());
        }
        
        $this->iters();
    }
    
    public function iters()
    {
        return call(function () {
            try {
                $emitter = new Emitter();
                $data = [1,2,3];
                foreach ($data as $val) {
                    $emitter->emit($val);
                }

                $iters = $emitter->iterate();
                while (yield $iters->advance()) {
                    echo $iters->getCurrent(), PHP_EOL;
                }

            } catch (\Throwable $ex) {
                \printf("Exception: %s\n", (string)$ex);
            }
        });
    }

    public function getUsers($pool)
    {
        return call(function () use ($pool) {
            $stm = yield $pool->prepare('SELECT * FROM users limit 5');
            $result = yield $stm->execute([]);
            return \Amp\Iterator\map($result, function ($value) {
                return $value['user_id'];
            });            
        });
    }
}

(new Hello())->run();

Perbedaannya hanya pada penggunaan klas Emitter dan fungsi map

return \Amp\Iterator\map($result, function ($value) {
    return $value['user_id'];
});

Penyebabnya belum saya ketahui