File Coverage

blib/lib/PubNub/PubSub.pm

Criterion	Covered	Total	%
statement	47	135	34.8
branch	6	64	9.3
condition	8	43	18.6
subroutine	10	17	58.8
pod	5	5	100.0
total	76	264	28.7

package PubNub::PubSub;

use strict;
use warnings;
use v5.10;

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

use PubNub::PubSub::Message;

our $VERSION = '1.0.1';

sub new {    ## no critic (RequireArgUnpacking)
    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 {              ## no critic (RequireArgUnpacking)
    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);

    return 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 {    ## no critic (RequireArgUnpacking)
    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 {    ## no critic (RequireArgUnpacking)
    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 (undef, undef, $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 {    ## no critic (RequireArgUnpacking)
    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__