view release on metacpan or  search on metacpan

socialcalc/third-party/lib/PocketIO/Connection.pm  view on Meta::CPAN

package PocketIO::Connection;

use strict;
use warnings;

use AnyEvent;
use JSON ();
use Encode ();
use Try::Tiny;

use PocketIO::Pool;

use constant DEBUG => $ENV{POCKETIO_CONNECTION_DEBUG};

sub new {
    my $class = shift;

    my $self = {@_};
    bless $self, $class;

    $self->{connect_timeout}   ||= 15;
    $self->{reconnect_timeout} ||= 15;

    $self->{max_messages_to_stage} ||= 32;
    $self->{messages} = [];

    $self->{on_connect_failed}   ||= sub { };
    $self->{on_connect}          ||= sub { };
    $self->{on_reconnect}        ||= sub { };
    $self->{on_reconnect_failed} ||= sub { };
    $self->{on_message}          ||= sub { };
    $self->{on_disconnect}       ||= sub { };
    $self->{on_error}            ||= sub { };

    $self->{data} = '';

    DEBUG && warn "Connection created\n";

    return $self;
}

sub type { shift->{type} }

sub is_connected { $_[0]->{is_connected} }

sub connecting {
    my $self = shift;

    DEBUG && warn "State 'connecting'\n";

    $self->{connect_timer} = AnyEvent->timer(
        after => $self->{connect_timeout},
        cb    => sub {
            DEBUG && warn "Timeout 'connect_timeout'";

            $self->on('connect_failed')->($self);
        }
    );
}

sub reconnecting {
    my $self = shift;

    DEBUG && warn "State 'reconnecting'\n";

    $self->{reconnect_timer} = AnyEvent->timer(
        after => $self->{reconnect_timeout},
        cb    => sub {
            DEBUG && warn "Timeout 'reconnect_timeout'\n";

            $self->on('reconnect_failed')->($self);
        }
    );
}

sub connected {
    my $self = shift;

    DEBUG && warn "State 'connected'\n";

    delete $self->{connect_timer};

    $self->{is_connected} = 1;

    $self->on('connect')->($self);

    return $self;
}

sub reconnected {
    my $self = shift;

    DEBUG && warn "State 'reconnected'\n";

    delete $self->{reconnect_timer};

    $self->on('reconnect')->($self);

    return $self;
}

sub disconnected {
    my $self = shift;

    DEBUG && warn "State 'disconnected'\n";

    delete $self->{connect_timer};
    delete $self->{reconnect_timer};

    $self->{data}     = '';
    $self->{messages} = [];

    $self->{is_connected} = 0;

    $self->{disconnect_timer} = AnyEvent->timer(
        after => 0,
        cb    => sub {
            $self->on('disconnect')->($self);
        }
    );

    return $self;
}

sub id {
    my $self = shift;

    $self->{id} ||= $self->_generate_id;

    return $self->{id};
}

sub on_message    { shift->on(message    => @_) }
sub on_disconnect { shift->on(disconnect => @_) }
sub on_error      { shift->on(error      => @_) }
sub on_write      { shift->on(write      => @_) }

sub on {
    my $self = shift;
    my $event = shift;

    my $name = "on_$event";

    unless (@_) {
        DEBUG && warn "Event 'on_$event'\n";

        return $self->{$name};
    }

    $self->{$name} = $_[0];

    return $self;
}

sub read {
    my $self = shift;
    my ($data) = @_;

    return $self unless defined $data;

    $self->{data} .= Encode::decode('UTF-8', $data);

    while (my $message = $self->_parse_data) {
        $self->on('message')->($self, $message);
    }

    return $self;
}

sub send_heartbeat {
    my $self = shift;

    $self->{heartbeat}++;

    return $self->send_message('~h~' . $self->{heartbeat});