prefork pubsub server using redis

pubsub_server_using_redis.pl

use strict;
use warnings;

use IO::Socket::INET;
use Parallel::Prefork;
use AnyEvent::Handle;
use AnyEvent::Redis;

$| = 1;

sub RedisChannel { 'chat' };
sub MaxRequestsPerChild { 100 };

my $listen = IO::Socket::INET->new(
  Listen    => 5,
  LocalHost => '0.0.0.0',
  LocalPort => '1986',
  Proto     => 'tcp',
) or die $!;

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

my $pm = Parallel::Prefork->new({
  max_workers  => 5,
  trap_signals => {
    TERM => 'TERM',
    HUP  => 'TERM',
  },
});

while ($pm->signal_received ne 'TERM') {
  $pm->start and next;
  my $reqs_remaining = MaxRequestsPerChild;
  $SIG{TERM} = sub {
    $reqs_remaining = 0;
  };
  while ($reqs_remaining-- > 0) {
    my $s = $listen->accept();
    print $s "Welcome, enter your name.\n";
    my $user_name = undef;
    
    my $cv = AE::cv;

    my $redis_sub = AnyEvent::Redis->new(
      host => '127.0.0.1',
      port => 6379,
      encoding => 'utf8',
      on_error => sub { warn @_ },
    );
    $redis_sub->subscribe(RedisChannel, sub {
        my ($message, $channel) = @_;
        print $s "$message\n" if defined $user_name;
    });

    my $handle = AnyEvent::Handle->new(
      fh => $s,
      on_read => sub {
        shift->push_read(line => sub {
          my ($h, $line) = @_;
          if ($line eq ':exit') {
            print $s "Bye Bye\n";
            if (defined $user_name) {
              $redis_pub->publish(RedisChannel, "-- $user_name has left.");
              $cv->cb(sub { $cv->recv });
            }
            $cv->send;
            return;
          }
          unless (defined $user_name) {
            unless ($line =~ /\S+/) {
              print $s "Enter valid name.\n";
            } else {
              $user_name = $line;
              print $s "Welcome, $user_name.\n";
              $redis_pub->publish(RedisChannel, "-- $user_name has joined.");
              $cv->cb(sub { $cv->recv });
            }
          } else {
            $redis_pub->publish(RedisChannel, "$user_name: $line");
            $cv->cb(sub { $cv->recv });
          }
        });
      },
      on_eof => sub {
        my ($h) = @_;
        warn "got eof";
        if (defined $user_name) {
          $redis_pub->publish(RedisChannel, "-- $user_name has left.");
          $cv->cb(sub { $cv->recv });
        }
        $h->destroy;
        $cv->send;
      },
      on_error => sub {
        my ($h, $fatal, $msg) = @_;
        warn "got error: $msg";
        $h->destroy;
        $cv->send;
      },
    );

    $cv->recv;
    close($s);
  }
  $pm->finish;
}

$pm->wait_all_children;