Async-Redis
view release on metacpan or search on metacpan
lib/Async/Redis/Subscription.pm view on Meta::CPAN
package Async::Redis::Subscription;
use strict;
use warnings;
use 5.018;
use Carp ();
use Future;
use Future::AsyncAwait;
use Future::IO;
use Scalar::Util qw(blessed refaddr weaken);
# Threshold for periodic event-loop yield inside the callback driver
# loop. Prevents stack growth when many messages are pre-queued and
# await on an already-ready Future returns synchronously.
use constant MAX_SYNC_DEPTH => 32;
sub new {
my ($class, %args) = @_;
return bless {
redis => $args{redis},
channels => {}, # channel => 1 (for regular subscribe)
patterns => {}, # pattern => 1 (for psubscribe)
sharded_channels => {}, # channel => 1 (for ssubscribe)
_pending_messages => [], # Queued messages for iterator consumers
_message_waiter => undef, # Future signalled when a message arrives
_slot_waiter => undef, # Future signalled when queue drains below depth
_fatal_error => undef, # Typed error set by _fail_fatal
_on_reconnect => undef, # Callback for reconnect notification
_on_message => undef, # Message-arrived callback (callback mode)
_on_error => undef, # Fatal-error callback
_driver_step => undef, # Running driver loop Future (owned by _tasks selector)
_closed => 0,
_paused => 0, # Set during reconnect; clears in _resume_after_reconnect
}, $class;
}
# Set/get reconnect callback
sub on_reconnect {
my ($self, $cb) = @_;
$self->{_on_reconnect} = $cb if @_ > 1;
return $self->{_on_reconnect};
}
# Set/get message-arrived callback. Once set, next() croaks — the
# subscription is in callback mode for the rest of its lifetime.
# $cb->($sub, $msg) receives the Subscription and the message hashref.
sub on_message {
my ($self, $cb) = @_;
if (@_ > 1) {
if (!$cb && $self->{_on_message}) {
Carp::croak(
"on_message is sticky; cannot clear once set "
. "(construct a new Subscription for iterator mode)"
);
}
$self->{_on_message} = $cb;
# If the subscription already has channels and is open, start
# the driver. If not, it'll be started when channels are added.
$self->_start_driver if $cb;
}
return $self->{_on_message};
}
# Set/get fatal-error callback. Fires once per fatal error; default
# (when unset) is to die so silent death is impossible.
# $cb->($sub, $err) receives the Subscription and the error.
sub on_error {
my ($self, $cb) = @_;
if (@_ > 1) {
$self->{_on_error} = $cb;
}
return $self->{_on_error};
}
lib/Async/Redis/Subscription.pm view on Meta::CPAN
};
}
elsif ($type eq 'smessage') {
$msg = {
type => 'smessage',
channel => $frame->[1],
pattern => undef,
data => $frame->[2],
};
}
else {
return undef; # non-message frame (subscribe confirmation, etc.)
}
# Queue the message for consumption by next() (iterator mode) or the
# callback driver loop (callback mode). The driver invokes _on_message;
# _dispatch_frame is intentionally agnostic about the consumption mode.
# This keeps backpressure uniform: the depth limit applies to both modes.
return if $self->{_closed};
my $redis = $self->{redis};
my $depth = ($redis && $redis->{message_queue_depth})
? $redis->{message_queue_depth}
: 0; # 0 = unbounded (default)
if ($depth && scalar(@{$self->{_pending_messages}}) >= $depth) {
# Queue full. Return a Future that queues the message once a slot
# opens (signalled by next() calling _slot_waiter->done).
$self->{_slot_waiter} //= Future->new;
my $slot = $self->{_slot_waiter};
weaken(my $weak = $self);
return $slot->then(sub {
return Future->done if !$weak || $weak->{_closed};
push @{$weak->{_pending_messages}}, $msg;
if (my $w = delete $weak->{_message_waiter}) {
$w->done unless $w->is_ready;
}
Future->done;
});
}
push @{$self->{_pending_messages}}, $msg;
if (my $w = delete $self->{_message_waiter}) {
$w->done unless $w->is_ready;
}
return undef;
}
# The callback-mode driver loop. Consumes from _pending_messages via
# _dequeue (populated by _run_reader's dispatch path), invokes the
# user's _on_message callback, and awaits its returned Future if any
# for consumer-opted backpressure.
#
# Exits cleanly when _dequeue returns undef (subscription closed or
# paused for reconnect). Dies with the typed error if _dequeue dies
# (fatal); _run_driver's Future failure is visible through the
# client's Future::Selector to any caller using run_until_ready.
#
# Periodic sleep(0) yield every MAX_SYNC_DEPTH iterations prevents
# stack growth when messages are pre-queued and await returns
# synchronously from an already-ready Future.
async sub _run_driver {
my ($self) = @_;
my $iter = 0;
while (!$self->{_closed} && !$self->{_paused}) {
my $msg;
my $deq_ok = eval { $msg = await $self->_dequeue(1); 1 };
unless ($deq_ok) {
my $err = $@;
# _fail_fatal already set _closed and fired on_error; don't
# double-fire. Any other propagation path routes through
# _handle_fatal_error.
return if $self->{_closed} || $self->{_paused};
$self->_handle_fatal_error($err);
return;
}
last unless defined $msg;
last if $self->{_closed} || $self->{_paused};
my $cb = $self->{_on_message} or last;
my $result = $self->_invoke_user_callback($cb, $msg);
if (blessed($result) && $result->isa('Future')) {
my $cb_ok = eval { await $result; 1 };
unless ($cb_ok) {
my $err = $@;
return if $self->{_closed} || $self->{_paused};
$self->_handle_fatal_error(
"on_message callback Future failed: $err"
);
return;
}
}
# Periodic yield prevents stack blowup when pre-queued messages
# resolve await synchronously.
await Future::IO->sleep(0) if ++$iter % MAX_SYNC_DEPTH == 0;
}
}
# Start the driver if not already running. Idempotent.
# Only starts when _on_message is set (callback mode). Iterator mode
# consumers call next() directly — no driver loop needed.
#
# Ownership: the driver Future is added to the client's Future::Selector
# ($redis->{_tasks}) and stored in $self->{_driver_step}. The selector
# owns the task; the slot is the dedup signal. on_ready clears the slot
# regardless of outcome. No ->retain.
sub _start_driver {
my ($self, $force) = @_;
return if $self->{_driver_step} && !$self->{_driver_step}->is_ready;
return unless $self->{_on_message}; # only callback mode needs a driver
return if $self->{_closed};
return if $self->{_paused};
return unless $self->channel_count > 0;
my $redis = $self->{redis} or return;
my $f = $self->_run_driver;
$self->{_driver_step} = $f;
$redis->{_tasks}->add(data => 'subscription-driver', f => $f);
$f->on_ready(sub { $self->{_driver_step} = undef });
return;
}
# Backward-compatible wrapper
async sub next_message {
my ($self) = @_;
my $msg = await $self->next();
return undef unless $msg;
# Convert new format to old format for compatibility
return {
channel => $msg->{channel},
message => $msg->{data},
pattern => $msg->{pattern},
type => $msg->{type},
};
}
# Intentional teardown: marks the subscription closed and wakes any
# blocked next() with undef. Clears the parent _subscription slot
# with an identity guard so a stale _close cannot evict a newer
# subscription object that reused the same slot.
sub _close {
my ($self) = @_;
return if $self->{_closed};
$self->{_closed} = 1;
$self->{_pending_messages} = [];
if (my $w = delete $self->{_message_waiter}) {
$w->done unless $w->is_ready;
}
if (my $w = delete $self->{_slot_waiter}) {
$w->done unless $w->is_ready;
lib/Async/Redis/Subscription.pm view on Meta::CPAN
If the returned Future fails, the failure is routed to C<on_error>.
=head2 Fatal error handling
$sub->on_error(sub {
my ($sub, $err) = @_;
...
});
C<on_error> fires when the underlying read encounters an error that
cannot be recovered by reconnect (e.g., reconnect is disabled, or
reconnect itself failed). After C<on_error> fires, the subscription is
closed and the driver stops.
B<If C<on_error> is not registered, fatal errors C<die>.> Silent death
of a pub/sub consumer is a debugging nightmare; loud-by-default
prevents it. If you genuinely want to swallow errors, register an
explicit no-op: C<< $sub->on_error(sub { }) >>.
Callback exceptions (dying inside C<on_message>) are also routed to
C<on_error>; the callback-died message is prepended to the error
string.
=head2 Ordering guarantee
Callbacks fire in the order frames arrive on the connection. No
concurrent invocation (Perl is single-threaded and the driver runs on
the event loop). After a reconnect, C<on_reconnect> always fires before
any post-reconnect C<on_message>.
=head2 Re-entrancy
Inside an C<on_message> callback you may safely:
=over 4
=item * Call C<< $sub->subscribe(...) >> — the new channel is added
cleanly; messages on it arrive via the same callback.
=item * Call C<< $sub->on_message($new_cb) >> — the current message is
dispatched to the previously-installed handler; the next frame uses
the new handler.
=item * C<die> — routed to C<on_error>.
=back
=head2 Backpressure and Redis server limits
Synchronous callbacks provide backpressure by blocking the driver loop:
while your callback runs, the driver doesn't read the next frame, so
TCP fills, Redis's output buffer grows. But Redis enforces
C<client-output-buffer-limit pubsub> (defaulting to S<32mb 8mb 60>
in recent versions) — if your subscriber cannot keep up for sustained
periods, B<Redis will disconnect you>. There is no amount of
client-side buffering that changes this: the limit is on the server.
If your processing is genuinely slow, return a Future from your
callback (enabling opt-in backpressure above) AND consider moving the
expensive work to a worker pool so the callback can return quickly.
Long synchronous processing in pub/sub callbacks is an anti-pattern at
scale regardless of client.
=head1 INTERNAL LIFECYCLE METHODS
The following methods are used by L<Async::Redis> to manage subscription
state. They are not part of the public API for end consumers, but are
documented here for maintainers.
=head2 _close
Intentional teardown. Marks the subscription closed and wakes any
blocked C<next()> with C<undef>. Clears the parent C<_subscription>
slot on the L<Async::Redis> object with an identity guard — a stale
C<_close> call from an earlier subscription object cannot evict a newer
one that has since taken the slot.
=head2 _fail_fatal($typed_error)
Unrecoverable failure. Marks the subscription closed with a typed error
object. Any blocked C<next()> call will C<die> with that error. The
error is preserved for callers who call C<next()> after the fact.
Routes through C<_close>'s identity guard for parent-slot clearing.
=head2 _pause_for_reconnect
Called before a reconnect attempt. Does B<not> mark the subscription
closed — the underlying reader has already exited due to the connection
drop. Channel/pattern tracking hashes are left intact for replay.
=head2 _resume_after_reconnect
Async. Replays all tracked C<SUBSCRIBE>, C<PSUBSCRIBE>, and
C<SSUBSCRIBE> commands on the freshly reconnected socket. Sets
C<in_pubsub=1> before sending replay commands so that racing message
frames classify correctly (mirrors the timing of the initial
subscribe). Fires C<on_reconnect> after replay. Callback-mode subscriptions
restart their callback driver; iterator-mode subscriptions continue to receive
frames through the parent Redis connection's reader.
=cut
( run in 0.576 second using v1.01-cache-2.11-cpan-9581c071862 )