The Perl and Raku Conference 2025: Greenville, South Carolina - June 27-29 Learn more

TEAM / Net-Async-Redis-6.004 / lib / Net / Async / Redis / Multi.pm 

            

              
1
2
3
4
5
6
—
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
—
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
—
125
126
127
128
129
130
131
132
133

              
package Net::Async::Redis::Multi;
use strict;
use warnings;
our $VERSION = '6.004'; # VERSION
=head1 NAME
Net::Async::Redis::Multi - represents multiple operations in a single Redis transaction
=head1 DESCRIPTION
Instances are generated by L<Net::Async::Redis/multi>.
=cut
use Syntax::Keyword::Try;
use Syntax::Keyword::Dynamically;
use Future::AsyncAwait;
use Scalar::Util qw(weaken blessed);
use Log::Any qw($log);
sub new {
    my ($class, %args) = @_;
    $args{queued_requests} //= [];
    weaken($args{redis} // die 'Must be provided a Net::Async::Redis instance');
    bless \%args, $class;
}
async sub exec {
    my ($self, $code) = @_;
    try {
        my $f = $self->$code;
        $f->retain if blessed($f) and $f->isa('Future');
        $log->tracef('MULTI exec');
        dynamically $self->redis->{_is_multi} = $self->redis->{multi_queue};
        my ($exec_result) = await $self->redis->exec;
        $self->redis->{multi_queue}->finish;
        my @reply = $exec_result->@*;
        my $success = 0;
        my $failure = 0;
        while(@reply) {
            try {
                my $reply = shift @reply;
                my $queued = shift @{$self->{queued_requests}} or die 'invalid queued request';
                $queued->done($reply) unless $queued->is_ready;
                ++$success
            } catch {
                $log->warnf("Failure during transaction: %s", $@);
                ++$failure
            }
        }
        return $success, $failure;
    } catch {
        my $err = $@;
        $log->errorf('Failed to complete multi - %s', $err);
        for my $queued (splice @{$self->{queued_requests}}) {
            try {
                $queued->fail("Transaction failed", redis => 'transaction_failure') unless $queued->is_ready;
            } catch {
                $log->warnf("Failure during transaction: %s", $@);
            }
        }
        die $@;
    }
}
=head2 redis
Accessor for the L<Net::Async::Redis> instance.
=cut
sub redis { shift->{redis} }
use Sub::Util qw(set_subname);
sub AUTOLOAD {
    my ($method) = our $AUTOLOAD =~ m{::([^:]+)$};
    # We only need to check this once
    die "Unknown method $method" unless Net::Async::Redis::Commands->can($method) or Net::Async::Redis->can($method);
    my $code = async sub {
        my ($self, @args) = @_;
        my $f = $self->redis->future->set_label($method);
        push @{$self->{queued_requests}}, $f;
        my $ff = do {
            # $self->redis->{_is_multi} //= 0;
            dynamically $self->redis->{_is_multi} = $self->redis->{multi_queue};
            $self->redis->$method(@args);
        };
        my ($resp) = await $ff;
        return await $f if $resp eq 'QUEUED';
        # my $addr = refaddr($f);
        # extract_by { $addr == refaddr($_) } @{$self->{queued_requests}};
        $f->fail($resp);
        die $resp;
    };
    set_subname $method => $code;
    { no strict 'refs'; *$method = sub { $code->(@_)->retain } }
    $code->(@_)->retain;
}
sub DESTROY {
    my ($self) = @_;
    return if ${^GLOBAL_PHASE} eq 'DESTRUCT' or not $self->{queued_requests};
    for my $queued (splice @{$self->{queued_requests}}) {
        try {
            $queued->cancel;
        } catch {
            $log->warnf("Failure during cleanup: %s", $@);
        }
    }
}
1;
__END__
=head1 AUTHOR
Tom Molesworth C<< <TEAM@cpan.org> >> plus contributors as mentioned in
L<Net::Async::Redis/CONTRIBUTORS>.
=head1 LICENSE
Copyright Tom Molesworth 2015-2024. Licensed under the same terms as Perl itself.