view release on metacpan or  search on metacpan

lib/Async/Redis/Subscription.pm  view on Meta::CPAN

# Returns undef on clean close; dies with typed error on fatal close.
async sub _dequeue {
    my ($self, $exit_on_pause) = @_;

    # Iterator mode (default): pause is transient. Block through it and
    # return real messages once the driver resumes.
    #
    # Callback driver (exit_on_pause=1): exit cleanly on pause so the
    # driver task can be restarted after reconnect without two drivers
    # racing. _pause_for_reconnect wakes any in-flight _message_waiter
    # via done() so this path can exit promptly.
    while (!@{$self->{_pending_messages}}) {
        die $self->{_fatal_error} if $self->{_fatal_error};
        return undef if $self->{_closed};
        return undef if $exit_on_pause && $self->{_paused};
        $self->{_message_waiter} //= Future->new;
        await $self->{_message_waiter};
        delete $self->{_message_waiter};
    }

    die $self->{_fatal_error} if $self->{_fatal_error};