view release on metacpan or  search on metacpan

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

# Track a channel subscription
sub _add_channel {
    my ($self, $channel) = @_;
    $self->{channels}{$channel} = 1;
    $self->_start_driver;
}

sub _add_pattern {
    my ($self, $pattern) = @_;
    $self->{patterns}{$pattern} = 1;
    $self->_start_driver;
}

sub _add_sharded_channel {
    my ($self, $channel) = @_;
    $self->{sharded_channels}{$channel} = 1;
    $self->_start_driver;
}

sub _remove_channel {
    my ($self, $channel) = @_;
    delete $self->{channels}{$channel};
}

sub _remove_pattern {
    my ($self, $pattern) = @_;
    delete $self->{patterns}{$pattern};
}

sub _remove_sharded_channel {
    my ($self, $channel) = @_;
    delete $self->{sharded_channels}{$channel};
}

# List subscribed channels/patterns
sub channels { keys %{shift->{channels}} }
sub patterns { keys %{shift->{patterns}} }
sub sharded_channels { keys %{shift->{sharded_channels}} }

sub channel_count {
    my ($self) = @_;
    return scalar(keys %{$self->{channels}})
         + scalar(keys %{$self->{patterns}})
         + scalar(keys %{$self->{sharded_channels}});
}

# Internal dequeue: wait for a message from the queue, dequeue it, and
# signal _slot_waiter so any pending _dispatch_frame can proceed. Used by
# both next() (iterator mode) and _start_driver (callback mode driver loop).
# 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};
    return undef if $self->{_closed} && !@{$self->{_pending_messages}};
    return undef
        if $exit_on_pause && $self->{_paused} && !@{$self->{_pending_messages}};

    my $msg = shift @{$self->{_pending_messages}};
    if (my $w = delete $self->{_slot_waiter}) {
        $w->done unless $w->is_ready;
    }
    return $msg;
}

# Receive next message (async iterator pattern). Waits on the queue
# populated by _run_reader via _dispatch_frame. Returns undef on clean
# close; dies with the typed error on fatal close.
async sub next {
    my ($self) = @_;

    # Exclusivity check: callback mode disables iterator mode.
    if ($self->{_on_message}) {
        Carp::croak("Cannot call next() on a callback-driven subscription");
    }

    # In iterator mode the unified reader (_run_reader) feeds the queue.
    # The reader is already running (started by _pubsub_command during
    # subscribe). No separate driver start is needed.

    return await $self->_dequeue;
}

# Read one pub/sub frame from the underlying connection. On transient
# read error, attempt reconnect if enabled and fire on_reconnect on
# success; on unrecoverable failure, propagate the error.
# Returns a Future resolving to the raw frame (arrayref) or undef if
# the connection is gone and no more frames are available.
# Shared by next() and the callback driver loop added in a later task.
async sub _read_frame_with_reconnect {
    my ($self) = @_;
    my $redis = $self->{redis};

    while (1) {
        my $frame;
        my $ok = eval {
            $frame = await $redis->_read_pubsub_frame;
            1;
        };

        unless ($ok) {
            my $error = $@;
            if ($redis->{reconnect} && $self->channel_count > 0) {
                my $reconnect_error;