File Coverage

blib/lib/PubNub/PubSub.pm

Criterion	Covered	Total	%
statement	45	133	33.8
branch	6	64	9.3
condition	8	43	18.6
subroutine	9	16	56.2
pod	5	5	100.0
total	73	261	27.9

package PubNub::PubSub;

use strict;
use v5.10;
our $VERSION = '1.1.0';

use Carp;
use Mojo::JSON qw/encode_json/;
use Mojo::UserAgent;
use Mojo::Util qw/url_escape/;

use PubNub::PubSub::Message;

sub new {
    my $class = shift;
    my %args  = @_ % 2 ? %{$_[0]} : @_;

    $args{host} ||= 'pubsub.pubnub.com';
    $args{port} ||= 80;
    $args{timeout} ||= 60; # for ua timeout
    $args{publish_queue} ||= [];

    my $proto = ($args{port} == 443) ? 'https://' : 'http://';
    $args{web_host} ||= $proto . $args{host};

    return bless \%args, $class;
}

sub __ua {
    my $self = shift;

    return $self->{ua} if exists $self->{ua};

    my $ua = Mojo::UserAgent->new;
    $ua->max_redirects(3);
    $ua->inactivity_timeout($self->{timeout});
    $ua->proxy->detect; # env proxy
    $ua->cookie_jar(0);
    $ua->max_connections(100);
    $self->{ua} = $ua;

    return $ua;
}

sub publish {
    my $self = shift;

    my %params = @_ % 2 ? %{$_[0]} : @_;
    my $callback = $params{callback} || $self->{publish_callback};

    my $ua = $self->__ua;

    my @steps = map {
             my $ref = $_;
             my $url = $ref->{url};
             sub {
                 my $delay = shift;
                 my $end = $delay->begin;
                 $ua->get($url => sub {
                    $callback->($_[1]->res, $ref->{message}) if $callback;
                    $end->();
                  });
             }
    } $self->__construct_publish_urls(%params);

    Mojo::IOLoop->delay(@steps)->wait;
}

sub __construct_publish_urls {
    my ($self, %params) = @_;

    my $pub_key = $params{pub_key} || $self->{pub_key};
    $pub_key or croak "pub_key is required.";
    my $sub_key = $params{sub_key} || $self->{sub_key};
    $sub_key or croak "sub_key is required.";
    my $channel = $params{channel} || $self->{channel};
    $channel or croak "channel is required.";
    $params{messages} or croak "messages is required.";

    return map {
        my $json = $_->json;
        my $uri = Mojo::URL->new( $self->{web_host} . qq~/publish/$pub_key/$sub_key/0/$channel/0/~ . url_escape($json) );
        $uri->query($_->query_params(\%params));
        { url => $uri->to_string, message => $_ };
    } map { PubNub::PubSub::Message->new($_) } @{$params{messages}};
}

sub subscribe {
    my $self = shift;
    my %params = @_ % 2 ? %{$_[0]} : @_;

    my $sub_key = $params{sub_key} || $self->{sub_key};
    $sub_key or croak "sub_key is required.";
    my $channel = $params{channel} || $self->{channel};
    $channel or croak "channel is required.";

    my $callback = $params{callback} or croak "callback is required.";
    my $timetoken = $params{timetoken} || '0';

    my $ua = $self->__ua;

    my $tx = $ua->get($self->{web_host} . "/subscribe/$sub_key/$channel/0/$timetoken");
    unless ($tx->success) {
        # for example $tx->error->{message} =~ /Inactivity timeout/

        # This is not a traditional goto. Instead it exits this function
        # and re-enters with @ as params.
        #
        # see goto docs, this is basically a method call which exits the current
        # function first.  So no extra call stack depth.
        sleep 1;
        @_ = ($self, %params, timetoken => $timetoken);
        goto &subscribe;
    }
    my $json = $tx->res->json;
    my @cb_args = $params{raw_msg}? ($json) : (@{$json->[0]});

    my $rtn = $callback ? $callback->(@cb_args) : 1;
    return unless $rtn;

    $timetoken = $json->[1];
    return $self->subscribe(%params, timetoken => $timetoken);
}

sub subscribe_multi {
    my $self = shift;
    my %params = @_ % 2 ? %{$_[0]} : @_;
    croak 'channels must be an arrayref'
         unless ref($params{channels}) =~ /ARRAY/;
    croak 'callback must be a hashref or coderef'
         unless ref($params{callback}) =~ /(HASH|CODE)/;

    my $callback;
    if (ref($params{callback}) =~ /HASH/){
       for (keys %{$params{callback}}) {
           croak "Non-coderef value found for callback key $_"
                unless ref($params{callback}->{$_}) =~ /CODE/;
       }
       $callback = sub {
           my ($obj) = @_;
           my ($msg, $timetoken, $channel) = @$obj;
           my $cb_dispatch = $params{callback};
           unless ($channel) { # on connect messages
              goto $cb_dispatch->{on_connect}
                   if exists $cb_dispatch->{on_connect};
              return 1;
           }
           if (exists $cb_dispatch->{$channel}){

              # these are verified coderefs, so replacing the current stack
              # frame with a call to the function.  They will *not* jump to
              # a label or other points.  Basically this just lets us pretend
              # that this was called directly by subscribe above.
              goto $cb_dispatch->{$channel};
           } elsif (exists $cb_dispatch->{'_default'}){
              goto $cb_dispatch->{_default};
           } else {
              warn 'Using callback dispatch table, cannot find channel callback'
                   . ' and _default callback not specified';
              return;
           }
       };
    }
    $callback = $params{callback} unless ref $callback;

    my $channel_string = join ',', @{$params{channels}};
    return $self->subscribe(channel => $channel_string, callback => $callback,
                           raw_msg => 1);
}

sub history {
    my $self = shift;

    if (scalar(@_) == 1 and ref($_[0]) ne 'HASH' and $_[0] =~ /^\d+$/) {
        @_ = (count => $_[0]);
        warn "->history(\$num) is deprecated and will be removed in next few releases.\n";
    }

    my %params = @_ % 2 ? %{$_[0]} : @_;

    my $sub_key = delete $params{sub_key} || $self->{sub_key};
    $sub_key or croak "sub_key is required.";
    my $channel = delete $params{channel} || $self->{channel};
    $channel or croak "channel is required.";

    my $ua = $self->__ua;

    my $tx = $ua->get($self->{web_host} . "/v2/history/sub-key/$sub_key/channel/$channel" => form => \%params);
    return [$tx->error->{message}] unless $tx->success;
    return $tx->res->json;
}

1;
__END__