Playing with Redis Streams

gistfile1.txt

use Redis::Async;

my $r = Redis::Async.new("localhost:6379", :0timeout);

sub xread(*@streams, Int :$timeout = 0, Bool :$add-stream, Bool :$add-id)
{
    supply {
        my %streams = @streams.map: * => '$';
        loop {
            with $r.XREAD('block', $timeout, 'streams',
                          %streams.keys, %streams.values)
            {
                for $_ -> @ ($stream, @messages)
                {
                    for @messages -> @ ($id, @fields)
                    {
                        %streams{$stream} = $id;
                        my $obj = %( @fields );
                        $obj<_stream> = $stream if $add-stream;
                        $obj<_id> = $id if $add-id;
                        emit $obj;
                    }
                }
            }
            else
            {
                warn "Timeout!";
                last
            }
        }
    }
}

react
{
    whenever xread(<mystream1 mystream2>, :add-stream)
    {
        .say
    }
}