*/ final class UnboundedReceiver implements ReceiverInterface { /** * @use ChannelSideTrait> */ use ChannelSideTrait; private null|Suspension $suspension = null; /** * @param UnboundedChannelState $state */ public function __construct(UnboundedChannelState $state) { $this->state = $state; } /** * {@inheritDoc} */ public function receive(): mixed { if ($this->suspension) { $this->suspension = $suspension = EventLoop::getSuspension(); $this->state->waitForMessage($suspension); $suspension->suspend(); } try { return $this->state->receive(); } catch (Exception\EmptyChannelException) { $this->suspension = $suspension = EventLoop::getSuspension(); $this->state->waitForMessage($suspension); $suspension->suspend(); /** @psalm-suppress MissingThrowsDocblock */ return $this->state->receive(); } finally { $this->suspension = null; } } /** * {@inheritDoc} */ public function tryReceive(): mixed { return $this->state->receive(); } }