File Coverage

blib/lib/Net/Stomp.pm

Criterion	Covered	Total	%
statement	272	308	88.3
branch	88	116	75.8
condition	27	42	64.2
subroutine	32	34	94.1
pod	13	13	100.0
total	432	513	84.2

package Net::Stomp;
use strict;
use warnings;
use IO::Select;
use Net::Stomp::Frame;
use Carp qw(longmess);
use base 'Class::Accessor::Fast';
use Log::Any;
our $VERSION = '0.60';

__PACKAGE__->mk_accessors( qw(
    current_host failover hostname hosts port select serial session_id socket ssl
    ssl_options socket_options subscriptions _connect_headers bufsize
    reconnect_on_fork logger connect_delay
    reconnect_attempts initial_reconnect_attempts timeout receipt_timeout
) );

sub _logconfess {
    my ($self,@etc) = @_;
    my $m = longmess(@etc);
    $self->logger->fatal($m);
    die $m;
}
sub _logdie {
    my ($self,@etc) = @_;
    $self->logger->fatal(@etc);
    die "@etc";
}

sub new {
    my $class = shift;
    my $self  = $class->SUPER::new(@_);

    $self->bufsize(8192) unless $self->bufsize;
    $self->connect_delay(5) unless defined $self->connect_delay;
    $self->reconnect_on_fork(1) unless defined $self->reconnect_on_fork;
    $self->reconnect_attempts(0) unless defined $self->reconnect_attempts;
    $self->initial_reconnect_attempts(1) unless defined $self->initial_reconnect_attempts;
    $self->socket_options({}) unless defined $self->socket_options;

    $self->logger(Log::Any->get_logger) unless $self->logger;

    $self->{_framebuf} = "";

    # We are not subscribed to anything at the start
    $self->subscriptions( {} );

    $self->select( IO::Select->new );
    my @hosts = ();

    # failover://tcp://primary:61616
    # failover:(tcp://primary:61616,tcp://secondary:61616)?randomize=false

    if ($self->failover) {
        my ($uris, $opts) = $self->failover =~ m{^failover:(?://)? \(? (.*?) \)? (?: \? (.*?) ) ?$}ix;

        $self->_logconfess("Unable to parse failover uri: " . $self->failover)
            unless $uris;

        foreach my $host (split(/,/,$uris)) {
            $host =~ m{^\w+://([a-zA-Z0-9\-./]+):([0-9]+)$} || $self->_logconfess("Unable to parse failover component: '$host'");
            my ($hostname, $port) = ($1, $2);

            push(@hosts, {hostname => $hostname, port => $port});
        }
    } elsif ($self->hosts) {
        ## @hosts is used inside the while loop later to decide whether we have
        ## cycled through all setup hosts.
        @hosts = @{$self->hosts};
    }
    $self->hosts(\@hosts) if @hosts;

    $self->_get_connection_retrying(1);

    return $self;
}

sub _get_connection_retrying {
    my ($self,$initial) = @_;

    my $tries=0;
    while(not eval { $self->_get_connection; 1 }) {
        my $err = $@;$err =~ s{\n\z}{}sm;
        ++$tries;
        if($self->_should_stop_trying($initial,$tries)) {
            # We've cycled enough. Die now.
            $self->_logdie("Failed to connect: $err; giving up");
        }
        $self->logger->warn("Failed to connect: $err; retrying");
        sleep($self->connect_delay);
    }
}

sub _should_stop_trying {
    my ($self,$initial,$tries) = @_;

    my $max_tries = $initial
        ? $self->initial_reconnect_attempts
        : $self->reconnect_attempts;

    return unless $max_tries > 0; # 0 means forever

    if (defined $self->hosts) {
        $max_tries *= @{$self->hosts}; # try at least once per host
    }
    return $tries >= $max_tries;
}

my $socket_class;
sub _get_connection {
    my $self = shift;
    if (my $hosts = $self->hosts) {
        if (defined $self->current_host && ($self->current_host < $#{$hosts} ) ) {
            $self->current_host($self->current_host+1);
        } else {
            $self->current_host(0);
        }
        my $h = $hosts->[$self->current_host];
        $self->hostname($h->{hostname});
        $self->port($h->{port});
        $self->ssl($h->{ssl});
        $self->ssl_options($h->{ssl_options} || {});
    }
    my $socket = $self->_get_socket;
    $self->_logdie("Error connecting to " . $self->hostname . ':' . $self->port . ": $!")
        unless $socket;

    $self->select->remove($self->socket);

    $self->select->add($socket);
    $self->socket($socket);
    $self->{_pid} = $$;
}

sub _get_socket {
    my ($self) = @_;
    my $socket;

    my $timeout = $self->timeout;
    $timeout = 5 unless defined $timeout;

    my %sockopts = (
        Timeout  => $timeout,
        %{ $self->socket_options },
        PeerAddr => $self->hostname,
        PeerPort => $self->port,
        Proto    => 'tcp',
    );
    my $keep_alive = delete $sockopts{keep_alive};

    if ( $self->ssl ) {
        eval { require IO::Socket::SSL };
        $self->_logdie(
            "You should install the IO::Socket::SSL module for SSL support in Net::Stomp"
        ) if $@;
        %sockopts = ( %sockopts, %{ $self->ssl_options || {} } );
        $self->logger->trace('opening IO::Socket::SSL',\%sockopts);
        $socket = IO::Socket::SSL->new(%sockopts);
    } else {
        $socket_class ||= eval { require IO::Socket::IP; IO::Socket::IP->VERSION('0.20'); "IO::Socket::IP" }
            || do { require IO::Socket::INET; "IO::Socket::INET" };
        $self->logger->trace("opening $socket_class",\%sockopts);
        $socket = $socket_class->new(%sockopts);
        binmode($socket) if $socket;
    }
    if ($keep_alive) {
        require Socket;
        if (Socket->can('SO_KEEPALIVE')) {
            $socket->setsockopt(Socket::SOL_SOCKET(),Socket::SO_KEEPALIVE(),1);
        }
        else {
            $self->logger->warn(q{TCP keep-alive was requested, but the Socket module does not export the SO_KEEPALIVE constant, so we couldn't enable it});
        }
    }

    return $socket;
}

sub connect {
    my ( $self, $conf ) = @_;

    $self->logger->trace('connecting');
    my $frame = Net::Stomp::Frame->new(
        { command => 'CONNECT', headers => $conf } );
    $self->send_frame($frame);
    $frame = $self->receive_frame;

    if ($frame && $frame->command eq 'CONNECTED') {
        $self->logger->trace('connected');
        # Setting initial values for session id, as given from
        # the stomp server
        $self->session_id( $frame->headers->{session} );
        $self->_connect_headers( $conf );
    }
    else {
        $self->logger->warn('failed to connect',{ %{$frame} });
    }

    return $frame;
}

sub _close_socket {
    my ($self) = @_;
    return unless $self->socket;
    $self->logger->trace('closing socket');
    $self->socket->close;
    $self->select->remove($self->socket);
}

sub disconnect {
    my $self = shift;
    $self->logger->trace('disconnecting');
    my $frame = Net::Stomp::Frame->new( { command => 'DISCONNECT' } );
    $self->send_frame($frame);
    $self->_close_socket;
    return 1;
}

sub _reconnect {
    my $self = shift;
    $self->_close_socket;

    $self->logger->warn("reconnecting");
    $self->_get_connection_retrying(0);
    # Both ->connect and ->subscribe can call _reconnect. It *should*
    # work out fine in the end, worst scenario we send a few subscribe
    # frame more than once
    $self->connect( $self->_connect_headers );
    for my $sub(keys %{$self->subscriptions}) {
        $self->subscribe($self->subscriptions->{$sub});
    }
}

sub can_read {
    my ( $self, $conf ) = @_;

    # If there is any data left in the framebuffer that we haven't read, return
    # 'true'. But we don't want to spin endlessly, so only return true the
    # first time. (Anything touching the _framebuf should update this flag when
    # it does something.
    if ( $self->{_framebuf_changed} && length $self->{_framebuf} ) {
        $self->{_framebuf_changed} = 0;
        return 1;
    }

    $conf ||= {};
    my $timeout = exists $conf->{timeout} ? $conf->{timeout} : $self->timeout;
    return $self->select->can_read($timeout) || 0;
}

sub send {
    my ( $self, $conf ) = @_;
    $conf = { %$conf };
    $self->logger->trace('sending',$conf);
    my $body = $conf->{body};
    delete $conf->{body};
    my $frame = Net::Stomp::Frame->new(
        { command => 'SEND', headers => $conf, body => $body } );
    $self->send_frame($frame);
    return 1;
}

sub send_with_receipt {
    my ( $self, $conf ) = @_;
    $conf = { %$conf };

    # send the message
    my $receipt_id = $self->_get_next_transaction;
    $self->logger->debug('sending with receipt',{ receipt => $receipt_id });
    $conf->{receipt} = $receipt_id;
    my $receipt_timeout = exists $conf->{timeout} ? delete $conf->{timeout} : $self->receipt_timeout;
    $self->send($conf);

    $self->logger->trace('waiting for receipt',$conf);
    # check the receipt
    my $receipt_frame = $self->receive_frame({
        ( defined $receipt_timeout ?
              ( timeout => $receipt_timeout )
              : () ),
    });

    if (@_ > 2) {
        $_[2] = $receipt_frame;
    }

    if (   $receipt_frame
        && $receipt_frame->command eq 'RECEIPT'
        && $receipt_frame->headers->{'receipt-id'} eq $receipt_id )
    {
        $self->logger->debug('got good receipt',{ %{$receipt_frame} });
        return 1;
    } else {
        $self->logger->debug('got bad receipt',{ %{$receipt_frame || {} } });
        return 0;
    }
}

sub send_transactional {
    my ( $self, $conf ) = @_;

    $conf = { %$conf };
    # begin the transaction
    my $transaction_id = $self->_get_next_transaction;
    $self->logger->debug('starting transaction',{ transaction => $transaction_id });
    my $begin_frame
        = Net::Stomp::Frame->new(
        { command => 'BEGIN', headers => { transaction => $transaction_id } }
        );
    $self->send_frame($begin_frame);

    $conf->{transaction} = $transaction_id;
    my $receipt_frame;
    my $ret = $self->send_with_receipt($conf,$receipt_frame);

    if (@_ > 2) {
        $_[2] = $receipt_frame;
    }

    if ( $ret ) {
        # success, commit the transaction
        $self->logger->debug('committing transaction',{ transaction => $transaction_id });
        my $frame_commit = Net::Stomp::Frame->new(
            {   command => 'COMMIT',
                headers => { transaction => $transaction_id }
            }
        );
        $self->send_frame($frame_commit);
    } else {
        $self->logger->debug('rolling back transaction',{ transaction => $transaction_id });
        # some failure, abort transaction
        my $frame_abort = Net::Stomp::Frame->new(
            {   command => 'ABORT',
                headers => { transaction => $transaction_id }
            }
        );
        $self->send_frame($frame_abort);
    }
    return $ret;
}

sub _sub_key {
    my ($conf) = @_;

    if ($conf->{id}) { return "id-".$conf->{id} }
    return "dest-".$conf->{destination}
}

sub subscribe {
    my ( $self, $conf ) = @_;
    $self->logger->trace('subscribing',$conf);
    my $frame = Net::Stomp::Frame->new(
        { command => 'SUBSCRIBE', headers => $conf } );
    $self->send_frame($frame);
    my $subs = $self->subscriptions;
    $subs->{_sub_key($conf)} = $conf;
    return 1;
}

sub unsubscribe {
    my ( $self, $conf ) = @_;
    $self->logger->trace('unsubscribing',$conf);
    my $frame = Net::Stomp::Frame->new(
        { command => 'UNSUBSCRIBE', headers => $conf } );
    $self->send_frame($frame);
    my $subs = $self->subscriptions;
    delete $subs->{_sub_key($conf)};
    return 1;
}

sub ack {
    my ( $self, $conf ) = @_;
    $conf = { %$conf };
    my $id    = $conf->{frame}->headers->{'message-id'};
    delete $conf->{frame};
    $self->logger->trace('acking',{ 'message-id' => $id, %$conf });
    my $frame = Net::Stomp::Frame->new(
        { command => 'ACK', headers => { 'message-id' => $id, %$conf } } );
    $self->send_frame($frame);
    return 1;
}

sub nack {
    my ( $self, $conf ) = @_;
    $conf = { %$conf };
    my $id    = $conf->{frame}->headers->{'message-id'};
    $self->logger->trace('nacking',{ 'message-id' => $id, %$conf });
    delete $conf->{frame};
    my $frame = Net::Stomp::Frame->new(
        { command => 'NACK', headers => { 'message-id' => $id, %$conf } } );
    $self->send_frame($frame);
    return 1;
}

sub send_frame {
    my ( $self, $frame ) = @_;
    # see if we're connected before we try to syswrite()
    if (not $self->_connected) {
        $self->_reconnect;
        if (not $self->_connected) {
            $self->_logdie(q{wasn't connected; couldn't _reconnect()});
        }
    }
    # keep writing until we finish, or get an error
    my $to_write = my $frame_string = $frame->as_string;
    my $written;
    while (length($to_write)) {
        local $SIG{PIPE}='IGNORE'; # just in case writing to a closed
                                   # socket kills us
        $written = $self->socket->syswrite($to_write);
        last unless defined $written;
        substr($to_write,0,$written,'');
    }
    if (not defined $written) {
        $self->logger->warn("error writing frame <<$frame_string>>: $!");
    }
    unless (defined $written && $self->_connected) {
        $self->_reconnect;
        $self->send_frame($frame);
    }
    return;
}

sub _read_data {
    my ($self, $timeout) = @_;

    return unless $self->select->can_read($timeout);
    my $len = $self->socket->sysread($self->{_framebuf},
                                     $self->bufsize,
                                     length($self->{_framebuf} || ''));

    if (defined $len && $len>0) {
        $self->{_framebuf_changed} = 1;
    }
    else {
        if (!defined $len) {
            $self->logger->warn("error reading frame: $!");
        }
        # EOF or error detected - connection is gone. We have to reset
        # the framebuf in case we had a partial frame in there that
        # will never arrive.
        $self->_close_socket;
        $self->{_framebuf} = "";
        delete $self->{_command};
        delete $self->{_headers};
    }
    return $len;
}

sub _read_headers {
    my ($self) = @_;

    return 1 if $self->{_headers};
    if ($self->{_framebuf} =~ s/^\n*([^\n].*?)\n\n//s) {
        $self->{_framebuf_changed} = 1;
        my $raw_headers = $1;
        if ($raw_headers =~ s/^(.+)\n//) {
            $self->{_command} = $1;
        }
        foreach my $line (split(/\n/, $raw_headers)) {
            my ($key, $value) = split(/\s*:\s*/, $line, 2);
            $self->{_headers}{$key} = $value
                unless defined $self->{_headers}{$key};
        }
        return 1;
    }
    return 0;
}

sub _read_body {
    my ($self) = @_;

    my $h = $self->{_headers};
    if ($h->{'content-length'}) {
        if (length($self->{_framebuf}) > $h->{'content-length'}) {
            $self->{_framebuf_changed} = 1;
            my $body = substr($self->{_framebuf},
                              0,
                              $h->{'content-length'},
                              '' );

            # Trim the trailer off the frame.
            $self->{_framebuf} =~ s/^.*?\000\n*//s;
            return Net::Stomp::Frame->new({
                command => delete $self->{_command},
                headers => delete $self->{_headers},
                body => $body
            });
        }
    } elsif ($self->{_framebuf} =~ s/^(.*?)\000\n*//s) {
        # No content-length header.

        my $body = $1;
        $self->{_framebuf_changed} = 1;
        return Net::Stomp::Frame->new({
              command => delete $self->{_command},
              headers => delete $self->{_headers},
              body => $body });
    }

    return 0;
}

# this method is to stop the pointless warnings being thrown when trying to
# call peername() on a closed socket, i.e.
#   getpeername() on closed socket GEN125 at
#   /opt/xt/xt-perl/lib/5.12.3/x86_64-linux/IO/Socket.pm line 258.
#
# solution taken from:
# http://objectmix.com/perl/80545-warning-getpeername.html
sub _connected {
    my $self = shift;

    return if $self->{_pid} != $$ and $self->reconnect_on_fork;

    my $connected;
    {
        local $^W = 0;
        $connected = $self->socket->connected;
    }
    return $connected;
}

sub receive_frame {
    my ($self, $conf) = @_;

    $self->logger->trace('waiting to receive frame',$conf);
    my $timeout = exists $conf->{timeout} ? $conf->{timeout} :  $self->timeout;

    unless ($self->_connected) {
        $self->_reconnect;
    }

    my $done = 0;
    while ( not $done = $self->_read_headers ) {
        return undef unless $self->_read_data($timeout);
    }
    while ( not $done = $self->_read_body ) {
        return undef unless $self->_read_data($timeout);
    }

    return $done;
}

sub _get_next_transaction {
    my $self = shift;
    my $serial = $self->serial || 0;
    $serial++;
    $self->serial($serial);

    return ($self->session_id||'nosession') . '-' . $serial;
}

1;

__END__