view release on metacpan or  search on metacpan

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

Set or get callback-driven delivery. See L</CALLBACK-DRIVEN DELIVERY>.

=head2 on_error

    $sub->on_error(sub {
        my ($sub, $err) = @_;
        ...
    });

Set or get the fatal-error callback used by callback-driven delivery.

=head2 unsubscribe

    await $sub->unsubscribe('channel1');
    await $sub->unsubscribe;  # all regular channels

Unsubscribe regular channels. With no arguments, unsubscribes all regular
channels tracked by this subscription.

=head2 punsubscribe

    await $sub->punsubscribe('prefix:*');
    await $sub->punsubscribe;  # all patterns

Unsubscribe pattern subscriptions.

=head2 sunsubscribe

    await $sub->sunsubscribe('shard-channel');
    await $sub->sunsubscribe;  # all sharded channels

Unsubscribe sharded pub/sub channels.

=head2 channels / patterns / sharded_channels

Return the currently tracked regular channels, patterns, or sharded channels.

=head2 channel_count

Return the total number of tracked regular, pattern, and sharded subscriptions.

=head2 is_closed

Return true after the subscription has been closed.

=head1 MESSAGE STRUCTURE

    {
        type    => 'message',      # or 'pmessage', 'smessage'
        channel => 'channel_name',
        pattern => 'pattern',      # defined for pmessage, undef otherwise
        data    => 'payload',
    }

The C<pattern> key is always present. It is defined for C<pmessage>
frames (the matching glob pattern) and C<undef> for C<message> and
C<smessage> frames. Consumers do not need C<exists $msg-E<gt>{pattern}>
checks.

C<next()> always returns real pub/sub messages. Reconnection is transparent.

=head1 RECONNECTION

When C<reconnect> is enabled on the Redis connection, subscriptions are
automatically re-established after a connection drop. To be notified:

    $sub->on_reconnect(sub {
        my ($sub) = @_;
        warn "Reconnected, may have lost messages";
        # re-poll state, log, etc.
    });

Messages published while the connection was down are lost (Redis pub/sub
has no persistence).

=head1 CALLBACK-DRIVEN DELIVERY

As an alternative to the C<await $sub-E<gt>next> iterator, you can
register a callback to receive messages:

    my $sub = await $redis->subscribe('chat');
    $sub->on_message(sub {
        my ($sub, $msg) = @_;
        # $msg has the same shape as next() returns:
        #   { type => 'message'|'pmessage'|'smessage',
        #     channel => ...,
        #     pattern => ...,  # defined for pmessage, undef otherwise
        #     data    => ... }
    });

Callback mode is designed for fire-and-forget listeners — background
dispatchers, websocket gateways, channel-layer middleware — where the
iterator pattern's requirement to be inside an awaited async sub is
awkward or triggers Future::AsyncAwait "lost its returning future"
warnings.

=head2 Exclusivity

Once C<on_message> is set on a Subscription, it is callback-mode for
the rest of its lifetime. Calls to C<< $sub->next >> will C<croak>.
This is sticky — there is no way to switch back. If you need iterator
mode, construct a new Subscription.

=head2 Signature

    $sub->on_message(sub {
        my ($subscription, $message) = @_;
        ...
    });

The callback receives the C<$subscription> itself as its first argument
(consistent with C<on_reconnect>), and the message hashref as its
second. The return value is normally ignored; if the return is a
C<Future>, see L</Backpressure>.

=head2 Backpressure

If your callback returns a C<Future>, the driver waits for that Future
to resolve before reading the next frame: