AnyEvent::Redis pubsub

anyevent_pubsub.pl

use strict;
use warnings;

use AnyEvent::Handle;
use AnyEvent::Redis;

my $redis_sub = AnyEvent::Redis->new(
  host => '127.0.0.1',
  port => 6379,
  encoding => 'utf8',
  on_error => sub { warn @_ },
);

my $redis_pub = AnyEvent::Redis->new(
  host => '127.0.0.1',
  port => 6379,
  encoding => 'utf8',
  on_error => sub { warn @_ },
);

my $channel = "subscribe_test";

my $cv_sub = $redis_sub->subscribe($channel, sub {
  my ($message, $channel) = @_;
  print "[$channel] $message\n";
});
$cv_sub->cb(sub {
  $cv_sub->recv;
});

my $cv = AE::cv;
my $handle = AnyEvent::Handle->new(
  fh => \*STDIN,
  on_read => sub {
    shift->push_read(
      line => sub {
        my ($h, $line) = @_;
        my $cv_pub = $redis_pub->publish($channel, $line);
        $cv_pub->cb(sub {
          $cv_pub->recv;
        });
      },
    );
  },
  on_eof => sub {
    my ($h) = @_;
    warn "got eof";
    $h->destroy;
    $cv->send;
  },
  on_error => sub {
    my ($h, $fatal, $msg) = @_;
    warn "got error $msg";
    $h->destroy;
    $cv->send;
  },
);
$cv->recv;