File Coverage

blib/lib/XAS/Lib/Net/POE/Client.pm

Criterion	Covered	Total	%
statement	24	146	16.4
branch	0	12	0.0
condition	0	2	0.0
subroutine	8	25	32.0
pod	11	11	100.0
total	43	196	21.9

package XAS::Lib::Net::POE::Client;

our $VERSION = '0.02';

use POE;
use Try::Tiny;
use Socket ':all';
use Errno ':POSIX';
use POE::Filter::Line;
use POE::Wheel::ReadWrite;
use POE::Wheel::SocketFactory;

use XAS::Class
  debug     => 0,
  version   => $VERSION,
  base      => 'XAS::Lib::POE::Service',
  mixin     => 'XAS::Lib::Mixins::Keepalive',
  accessors => 'wheel host port listener socket',
  utils     => 'dotid',
  vars => {
    PARAMS => {
      -port            => 1,
      -retry_reconnect => { optional => 1, default => 1 },
      -tcp_keepalive   => { optional => 1, default => 0 },
      -filter          => { optional => 1, default => undef },
      -alias           => { optional => 1, default => 'client' },
      -eol             => { optional => 1, default => "\012\015" },
      -host            => { optional => 1, default => 'localhost'},
    }
  }
;

our @ERRORS = (0, EPIPE, ETIMEDOUT, ECONNRESET, ECONNREFUSED, ENETUNREACH, ENETDOWN, ENETRESET);
our @RECONNECTIONS = (60, 120, 240, 480, 960, 1920, 3840);

#use Data::Dumper;

# ----------------------------------------------------------------------
# Public Events
# ----------------------------------------------------------------------

# ---------------------------------------------------------------------
# Public Methods
# ---------------------------------------------------------------------

sub session_initialize {
    my $self = shift;

    my $alias = $self->alias;

    $self->log->debug("$alias: entering session_initialize()");

    # private events

    $self->log->debug("$alias: doing private events");

    # private events

    $poe_kernel->state('server_error',     $self, '_server_error');
    $poe_kernel->state('server_message',   $self, '_server_message');
    $poe_kernel->state('server_connect',   $self, '_server_connect');
    $poe_kernel->state('server_connected', $self, '_server_connected');
    $poe_kernel->state('server_reconnect', $self, '_server_reconnect');
    $poe_kernel->state('server_connection_failed', $self, '_server_connection_failed');

    # public events

    $self->log->debug("$alias: doing public events");

    $poe_kernel->state('read_data',         $self);
    $poe_kernel->state('write_data',        $self);
    $poe_kernel->state('connection_up',     $self);
    $poe_kernel->state('connection_down',   $self);
    $poe_kernel->state('handle_connection', $self);

    # walk the chain

    $self->SUPER::session_initialize();

    $self->log->debug("$alias: leaving session_initialize()");

}

sub session_startup {
    my $self = shift;

    my $alias = $self->alias;

    $self->log->debug("$alias: entering session_startup");

    $poe_kernel->post($alias, 'server_connect');

    # walk the chain

    $self->SUPER::session_startup();

    $self->log->debug("$alias: leaving session_startup");

}

sub session_shutdown {
    my $self = shift;

    my $alias = $self->alias;

    $self->log->debug("$alias: entering session_shutdown");

    $self->{'socket'}   = undef;
    $self->{'wheel'}    = undef;
    $self->{'listener'} = undef;

    # walk the chain

    $self->SUPER::session_shutdown();

    $self->log->debug("$alias: leaving session_shutdown");

}

sub session_pause {
    my ($self) = $_[OBJECT];

    my $alias = $self->alias;

    $self->log->debug("$alias: entering session_pause");

    $poe_kernel->call($alias, 'connection_down');

    # walk the chain

    $self->SUPER::session_pause();

    $self->log->debug("$alias: leaving session_pause");

}

sub session_resume {
    my ($self) = $_[OBJECT];

    my $alias = $self->alias;

    $self->log->debug("$alias: entering session_resume");

    $poe_kernel->call($alias, 'connection_up');

    # walk the chain

    $self->SUPER::session_resume();

    $self->log->debug("$alias: leaving session_resume");

}

# ---------------------------------------------------------------------
# Public Events
# ---------------------------------------------------------------------

sub handle_connection {
    my ($self) = $_[OBJECT];

}

sub connection_down {
    my ($self) = $_[OBJECT];

}

sub connection_up {
    my ($self) = $_[OBJECT];

}

sub read_data {
    my ($self, $data) = @_[OBJECT, ARG0];

    my $alias = $self->alias;

    $poe_kernel->post($alias, 'write_data', $data);

}

sub write_data {
    my ($self, $data) = @_[OBJECT, ARG0];

    my @packet;
    my $alias = $self->alias;

    if (my $wheel = $self->wheel) {

        push(@packet, $data);
        $wheel->put(@packet);

    } else {

        $self->throw_msg(
            dotid($self->class) . '.write_data.nowheel',
            'net_server_nowheel',
            $alias
        );

    }

}

# ---------------------------------------------------------------------
# Private Events
# ---------------------------------------------------------------------

sub _server_message {
    my ($self, $data, $wheel_id) = @_[OBJECT, ARG0, ARG1];

    my $alias = $self->alias;

    $self->log->debug("$alias: _server_message()");

    $poe_kernel->post($alias, 'read_data', $data);

}

sub _server_connected {
    my ($self, $socket, $peeraddr, $peerport, $wheel_id) = @_[OBJECT,ARG0..ARG3];

    my $alias = $self->alias;

    $self->log->debug("$alias: _server_connected()");

    my $wheel = POE::Wheel::ReadWrite->new(
        Handle     => $socket,
        Filter     => $self->filter,
        InputEvent => 'server_message',
        ErrorEvent => 'server_error',
    );

    my $host = gethostbyaddr($peeraddr, AF_INET);

    $self->{'host'}     = $host;
    $self->{'port'}     = $peerport;
    $self->{'wheel'}    = $wheel;
    $self->{'socket'}   = $socket;
    $self->{'attempts'} = 0;

    $poe_kernel->post($alias, 'handle_connection');

}

sub _server_connect {
    my ($self) = $_[OBJECT];

    my $alias = $self->alias;

    $self->log->debug("$alias: _server_connect()");

    $self->{'listner'} = POE::Wheel::SocketFactory->new(
        RemoteAddress  => $self->host,
        RemotePort     => $self->port,
        SocketType     => SOCK_STREAM,
        SocketDomain   => AF_INET,
        Reuse          => 'no',
        SocketProtocol => 'tcp',
        SuccessEvent   => 'server_connected',
        FailureEvent   => 'server_connection_failed',
    );

}

sub _server_connection_failed {
    my ($self, $operation, $errnum, $errstr, $wheel_id) = @_[OBJECT,ARG0..ARG3];

    my $alias = $self->alias;

    $self->log->debug("$alias: _server_connection_failed()");
    $self->log->error_msg('net_server_connection_failed', $alias, $operation, $errnum, $errstr);

    delete $self->{'socket'};
    delete $self->{'listner'};
    delete $self->{'wheel'};

    foreach my $error (@ERRORS) {

        if ($errnum == $error) {

            $poe_kernel->post($alias, 'server_reconnect');
            last;

        }

    }

}

sub _server_error {
    my ($self, $operation, $errnum, $errstr, $wheel_id) = @_[OBJECT,ARG0..ARG3];

    my $alias = $self->alias;

    $self->log->debug("$alias: _server_error()");
    $self->log->error_msg('net_server_error', $alias, $operation, $errnum, $errstr);

    delete $self->{'socket'};
    delete $self->{'listner'};
    delete $self->{'wheel'};

    $poe_kernel->post($alias, 'connection_down');

    foreach my $error (@ERRORS) {

        if ($errnum == $error) {

            $poe_kernel->post($alias, 'server_reconnect');
            last;

        }

    }

}

sub _server_reconnect {
    my ($self) = $_[OBJECT];

    my $retry;
    my $alias = $self->alias;

    $self->log->warn_msg('net_server_reconnect', $alias, $self->{'attempts'}, $self->{'count'});

    if ($self->{'attempts'} < $self->{'count'}) {

        my $delay = $RECONNECTIONS[$self->{'attempts'}];
        $self->log->warn_msg('net_server_attempts', $alias, $self->{'attempts'}, $delay);
        $self->{'attempts'} += 1;
        $poe_kernel->delay('server_connect', $delay);

    } else {

        $retry = $self->retry_reconnect || 0;

        if ($retry) {

            $self->log->warn_msg('net_server_recycle', $alias);
            $self->{'attempts'} = 0;
            $poe_kernel->post($alias, 'server_connect');

        } else {

            $self->log->warn_msg('net_server_shutdown', $alias);
            $poe_kernel->post($alias, 'session_shutdown');

        }

    }

}

# ---------------------------------------------------------------------
# Private Methods
# ---------------------------------------------------------------------

sub init {
    my $class = shift;

    my $self = $class->SUPER::init(@_);

    $self->{'attempts'} = 0;
    $self->{'count'} = scalar(@RECONNECTIONS);

    unless (defined($self->{'filter'})) {

        $self->{'filter'} = POE::Filter::Line->new(
            InputLiteral  => $self->eol,
            OutputLiteral => $self->eol,
        );

    }

    return $self;

}

1;

__END__