File Coverage

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

Criterion	Covered	Total	%
statement	15	88	17.0
branch	0	20	0.0
condition			n/a
subroutine	5	15	33.3
pod	9	9	100.0
total	29	132	21.9

package XAS::Lib::Stomp::POE::Client;

our $VERSION = '0.04';

use POE;
use Try::Tiny;
use XAS::Lib::Stomp::Utils;
use XAS::Lib::Stomp::POE::Filter;

use XAS::Class
  debug     => 0,
  version   => $VERSION,
  base      => 'XAS::Lib::Net::POE::Client',
  mixins    => 'XAS::Lib::Mixins::Keepalive',
  utils     => 'trim',
  accessors => 'stomp',
  vars => {
    PARAMS => {
      -host     => { optional => 1, default => undef },
      -port     => { optional => 1, default => undef },
      -alias    => { optional => 1, default => 'stomp-client' },
      -login    => { optional => 1, default => 'guest' },
      -passcode => { optional => 1, default => 'guest' },
    }
  }
;

#use Data::Dumper;

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

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

sub session_initialize {
    my $self = shift;

    my $alias = $self->alias;

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

    # public events

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

    $poe_kernel->state('handle_noop',      $self);
    $poe_kernel->state('handle_error',     $self);
    $poe_kernel->state('handle_message',   $self);
    $poe_kernel->state('handle_receipt',   $self);
    $poe_kernel->state('handle_connected', $self);

    # walk the chain

    $self->SUPER::session_initialize();

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

}

sub session_shutdown {
    my $self = shift;

    my $alias = $self->alias;
    my $frame = $self->stomp->disconnect(
        -receipt => 'disconnecting'
    );

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

    $poe_kernel->call($alias, 'write_data', $frame);

    # walk the chain

    $self->SUPER::session_shutdown();

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

}

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

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

    my $alias = $self->alias;
    my $frame = $self->stomp->connect(
        -login    => $self->login,
        -passcode => $self->passcode
    );

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

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

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

}

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

    my $alias = $self->alias;

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

    if ($self->tcp_keepalive) {

        $self->log->info_msg('tcp_keepalive_enabled', $alias);

        $self->init_keepalive();
        $self->enable_keepalive($self->socket);

    }

    $self->log->info_msg('net_server_connected', $alias, $self->host, $self->port);

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

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

}

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

}

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

}

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

    my $message = '';
    my $message_id = '';
    my $alias = $self->alias;

    if ($frame->header->methods->has('message_id')) {

        $message_id = $frame->header->message_id;

    }

    if ($frame->header->methods->has('message')) {

        $message = $frame->header->message;

    }

    $self->log->error_msg('stomp_errors',
        $alias,
        trim($message_id),
        trim($message),
        trim($frame->body)
    );

}

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

    my $alias = $self->alias;

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

}

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

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

    my $alias = $self->alias;

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

    if ($frame->command eq 'CONNECTED') {

        $self->log->debug("$alias: received a \"CONNECTED\" message");
        $poe_kernel->post($alias, 'handle_connected', $frame);

    } elsif ($frame->command eq 'MESSAGE') {

        $self->log->debug("$alias: received a \"MESSAGE\" message");
        $poe_kernel->post($alias, 'handle_message', $frame);

    } elsif ($frame->command eq 'RECEIPT') {

        $self->log->debug("$alias: received a \"RECEIPT\" message");
        $poe_kernel->post($alias, 'handle_receipt', $frame);

    } elsif ($frame->command eq 'ERROR') {

        $self->log->debug("$alias: received an \"ERROR\" message");
        $poe_kernel->post($alias, 'handle_error', $frame);

    } elsif ($frame->command eq 'NOOP') {

        $self->log->debug("$alias: received an \"NOOP\" message");
        $poe_kernel->post($alias, 'handle_noop', $frame);

    } else {

        $self->log->warn_msg('stomp_unknown_type', $alias, $frame->command);

    }

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

}

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

sub init {
    my $class = shift;

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

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

        $self->{'host'} = $self->env->mqserver;

    }

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

        $self->{'port'} = $self->env->mqport;

    }

    $self->{'stomp'}  = XAS::Lib::Stomp::Utils->new();
    $self->{'filter'} = XAS::Lib::Stomp::POE::Filter->new();

    return $self;

}

1;

__END__