File Coverage

blib/lib/Mojolicious/Plugin/Pubsub.pm

Criterion	Covered	Total	%
statement	114	125	91.2
branch	22	34	64.7
condition	1	3	33.3
subroutine	17	20	85.0
pod	1	1	100.0
total	155	183	84.7

use strict;
use warnings;
package Mojolicious::Plugin::Pubsub;
#ABSTRACT: Pubsub plugin for Mojolicious
$Mojolicious::Plugin::Pubsub::VERSION = '0.006';
use Mojo::Base 'Mojolicious::Plugin';

use Mojo::IOLoop;
use Mojo::JSON qw( decode_json encode_json );
use Mojo::Util qw( b64_decode b64_encode deprecated );
use IO::Socket::UNIX;

my $client;
my $conf;

sub register {
  my ($self, $app, $cfg) = @_;

  $cfg->{cbs} = [];
  push @{ $cfg->{subs} }, $cfg->{cb} if exists $cfg->{cb};
  $cfg->{socket} = $app->home->child($app->moniker . '.pubsub') unless exists $cfg->{socket};
  $conf = $cfg;

  my $loop = Mojo::IOLoop->singleton;

  pipe my $in, my $out or die "Could not open pipe pair: $!";

  my $pid = fork();
  die "Could not fork: $!" if not defined $pid;

  if ($pid) {
    close $out;
    chomp(my $result = readline $in);
    close $in;

    die "Could not establish pubsub socket: $result" if $result ne 'DONE';
  } else {
    # now in fork
    close $in;
    $loop->reset;

    my @streams;

    unless (-e $conf->{socket} and IO::Socket::UNIX->new(Peer => $conf->{socket})) {
      my $server = eval { $loop->server(
        {path => $conf->{socket}} => sub {
          my (undef, $stream) = @_;
          push @streams, $stream;

          my $msg;
          $stream->on(
            read => sub {
              my ($stream, $bytes) = @_;
              $msg .= $bytes;

              while (length $msg) {
                if ($msg =~ s/^(.+\n)//) {
                  my $line = $1;
                  foreach my $str (@streams) { $str->write($line); }
                } else {
                  return;
                }
              }
            }
          );

          $stream->on(
            close => sub {
              @streams = grep $_ ne $_[0], @streams;
              $loop->stop unless @streams;
            }
          );

          $stream->timeout(0);
        }
      ); };

      if (not defined $server) {
        print $out $@;
        close $out;
        exit;
      }
    }

    print $out "DONE\n";
    close $out;

    $loop->start unless $loop->is_running;
    exit;
  }

  $loop->next_tick(sub { _connect() });

  $app->helper(
    'pubsub.publish' => sub {
      my $self = shift;
      my $msg = b64_encode(encode_json([@_]), "");

      _send($msg . "\n");

      return $self;
    }
  );

  $app->helper(
    'pubsub.subscribe' => sub {
      my $self = shift;
      my $cb = shift;

      push @{ $conf->{subs} }, $cb;

      return $self;
    }
  );

  $app->helper(
    'pubsub.unsubscribe' => sub {
      my $self = shift;
      my $cb = shift;

      @{ $conf->{subs} } = grep { $_ != $cb } @{ $conf->{subs} };

      return $self;
    }
  );

  $app->helper(
    publish => sub {
      deprecated '->publish is deprecated in favour of ->pubsub->publish';
      shift->pubsub->publish(@_);
    });
  $app->helper(
    subscribe => sub {
      deprecated '->subscribe is deprecated in favour of ->pubsub->subscribe';
      shift->pubsub->subscribe(@_);
    });
  $app->helper(
    unsubscribe => sub {
      deprecated '->unsubscribe is deprecated in favour of ->pubsub->unsubscribe';
      shift->pubsub->unsubscribe(@_);
    });

}

sub _send {
  my ($msg) = @_;

  if (not defined $client) {
    return _connect(sub { $_[0]->write($msg); });
  }

  $client->write($msg);
}

sub _connect {

  my $cb = shift;

  Mojo::IOLoop->singleton->client(
    { path => $conf->{socket} } => sub {
      my ($loop, $err, $stream) = @_;
      die sprintf "Could not connect to %s: %s", $conf->{socket}, $err if defined $err;

      if (defined $client) {
        $stream->close();
        $cb->($client) if defined $cb;

        return;
      }

      $client = $stream;

      my $msg;
      $stream->on(read => sub {
        my ($stream, $bytes) = @_;

        $msg .= $bytes;

        while (length $msg) {
          if ($msg =~ s/^(.+)\n//) {
            my $b64 = $1;
            my $args = decode_json(b64_decode($b64));
            foreach my $subscriber (@{ $conf->{subs} }) {
              $subscriber->(@{ $args });
            }
          }
          else {
            return
          }

        }
      });

      $stream->timeout(0);

      $cb->($stream) if defined $cb;

    }
  );
}

1;

__END__