0

我正在使用使用 amphp 的 eventstore 客户端。我需要在我的应用程序中重用许多部分的连接。

所以我创建了一个连接提供者：

public function getConnection(): EventStoreConnection
{
    if ($this->connection) {
        return $this->connection;
    }
    $this->connection = $this->createConnection();
    wait($this->connection->connectAsync());

    return $this->connection;
}

然后我在很多地方使用这个连接：

\Amp\Loop::run(function () use ($eventStoreEvents, $streamName) {
    $connection = $this->connectionProvider->getConnection();

    // Creation of an event stream
    yield $connection->appendToStreamAsync($streamName, ExpectedVersion::ANY, $eventStoreEvents);
    // sleep(10); // This sleep does not work, code continue like nothing happend
});

\Amp\Loop::run(function () use ($streamName, $aggregateFqcn, &$aggregateRoot) {

    $start = 0;
    $count = \Prooph\EventStore\Internal\Consts::MAX_READ_SIZE;

    $connection = $this->connectionProvider->getConnection();

    do {
        $events = [];
        /** @var StreamEventsSlice $streamEventsSlice */
        $streamEventsSlice = yield $connection
            ->readStreamEventsForwardAsync(
                $streamName,
                $start,
                $count,
                true
            );

        if (!$streamEventsSlice->status()->equals(SliceReadStatus::success())) {
            dump($streamEventsSlice); // Event stream does not exist
            // Error here: the event stream doesn't exist at this point.
            throw new RuntimeGangxception('Impossible to generate the aggregate');
        }
    } while (! $streamEventsSlice->isEndOfStream());
});

问题：似乎第一个请求还没有结束，但第二个循环已经开始了。未注释的睡眠没有任何影响！

但是最终创建了事件流，其中包含相关事件，因此第一个请求有效。

如果我启动一个连接然后关闭然后启动一个新连接，它就可以工作。但由于每个新连接的握手开销，它很慢。

我用 Amphp 的 WebSocket 库尝试了一个类似的例子，它成功了。你看有什么不对吗？

这是我对 websocket 的测试：

$connection = \Amp\Promise\wait(connect('ws://localhost:8080'));
Amp\Loop::run(function () use ($connection) {
   /** @var Connection $connection */
   yield $connection->send("Hello...");
   sleep(10); // This sleep works!
});

Amp\Loop::run(function () use ($connection) {
   /** @var Connection $connection */
   yield $connection->send("... World !");
});

$connection->close();
4

2 回答 2

0

你试图做的事情毫无意义。您应该阅读amphp 的文档。

Amp 对事件循环使用全局访问器，因为每个应用程序只有一个事件循环。同时运行两个循环是没有意义的，因为它们只需要以繁忙的等待方式相互调度才能正确运行。

也就是说，实际上没有第二个循环。

于 2020-07-09T02:34:23.793 回答
0

Prooph eventstore 库基于 amphp 但不遵循所有原则：您不能等待连接准备好。如果你尝试大规模使用它会更糟，所以不要试图等待承诺完成。

作为替代方案，您可以为以后设置一个承诺并检查连接是否为空。这就是库在内部处理进一步步骤的实际操作。

在我这边，我决定停止使用这个库。但作为替代方案，您可以使用使用 HTTP 客户端的库，它也来自 prooph 团队。

于 2020-08-14T15:27:53.483 回答