File Coverage

blib/lib/Net/Stomp/Receipt.pm

Criterion	Covered	Total	%
statement	9	46	19.5
branch	0	6	0.0
condition	0	3	0.0
subroutine	3	7	42.8
pod	2	3	66.6
total	14	65	21.5

package Net::Stomp::Receipt;

use strict;
use warnings;

#
# Subclass of Net::Stomp for adding "transactional sends"
# with receipt and commit.
#
# Author: Hugo Salgado
#
use base 'Net::Stomp';

#
# I start the version in sync with Net::Stomp
# (and hopes to keep it in that way)
our $VERSION = '0.36';


# Added a new configuration variable on creation,
# so we need to subclass the constructor
sub new {
    my ($class, $conf) = @_;

    my $self = $class->SUPER::new({
        hostname => $conf->{'hostname'},
        port => $conf->{'port'},
    });

    # to keep the session id, as will be given from the server
    $self->{'sessionid'} = undef;
    # to keep an incremental for the transaction and receipt ids
    $self->{'serial'}    = 0;

    # and we add the persistent feature on constructor
    $self->{'PERSISTENT'} = 1 if $conf->{'PERSISTENT'};

    bless $self, $class;

    return $self;
}

# We need to set the sessionid on connection time
sub connect {
    my ( $self, $conf ) = @_;

    my $frame = $self->SUPER::connect( $conf );

    # Setting initial values for session id, as given from
    # the stomp server
    $self->{'sessionid'} = $frame->headers->{'session'};

    return $frame;
}

# Internal method for autoincremental serial id
sub _getSerial {
    my $self = shift;

    $self->{'serial'}++;

    return $self->{'serial'};
}

# The new method. We don't override the original "send", so
# we can use one or another.
sub send_safe {
    my ( $self, $conf ) = @_;
    my $body = $conf->{body};
    delete $conf->{body};

    # Transaction begins
    $conf->{transaction} = $self->{'sessionid'} .  '-' .  $self->_getSerial;
    my $frame = Net::Stomp::Frame->new(
        { command => 'BEGIN', headers => $conf } );
    $self->send_frame($frame);
    undef $frame;

    # Sending the message, with receipt header
    my $receipt_id = $self->{'sessionid'} .  '-' .  $self->_getSerial;
    $conf->{receipt} = $receipt_id;
    $conf->{persistent} = 'true' if $self->{'PERSISTENT'};
    $frame = Net::Stomp::Frame->new(
        { command => 'SEND', headers => $conf, body => $body } );
    $self->send_frame($frame);
    undef $frame;
    delete $conf->{receipt};
    delete $conf->{persistent};

    # Checking the server for the right receipt
    # If it's OK -> commit the transaction
    $frame = $self->receive_frame;
    if (($frame->command eq 'RECEIPT') and
        ($frame->headers->{'receipt-id'} eq $receipt_id)) {
        my $frame_commit = Net::Stomp::Frame->new(
            { command => 'COMMIT', headers => $conf } );
        $self->send_frame($frame_commit);

        return 1;
    }

    # whatever else, abort transaction
    my $frame_abort = Net::Stomp::Frame->new(
        { command => 'ABORT', headers => $conf } );
    $self->send_frame($frame_abort);

    return 0;
}


1;

__END__