File Coverage

blib/lib/XAS/Lib/Stomp/Utils.pm

Criterion	Covered	Total	%
statement	12	162	7.4
branch	0	80	0.0
condition	0	3	0.0
subroutine	4	18	22.2
pod	13	13	100.0
total	29	276	10.5

package XAS::Lib::Stomp::Utils;

our $VERSION = '0.03';

use XAS::Lib::Stomp::Frame;
use XAS::Constants ':stomp';

use XAS::Class
  debug   => 0,
  version => $VERSION,
  base    => 'XAS::Base',
  utils   => ':validation',
  vars => {
    PARAMS => {
      -target  => { optional => 1, default => undef, regex => STOMP_LEVELS },
    }
  }
;

#use Data::Dumper;

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

sub connect {
    my $self = shift;
    my $p = validate_params(\@_, {
        -login      => { optional => 1, default => undef },
        -passcode   => { optional => 1, default => undef },
        -host       => { optional => 1, default => 'localhost' },
        -heart_beat => { optional => 1, default => '0,0', regex => qr/\d+,\d+/ },
        -acceptable => { optional => 1, default => '1.0,1.1,1.2',
            callbacks => {
                'valid target' => \&_match
            }
        }
    });

    my $frame;
    my $header = {};

    $header->{'login'}    = $p->{'login'}    if (defined($p->{'login'}));
    $header->{'passcode'} = $p->{'passcode'} if (defined($p->{'passcode'}));

    if ($self->target > 1.0) {

        $header->{'host'}           = $p->{'host'};
        $header->{'heart-beat'}     = $p->{'heart_beat'};
        $header->{'accept-version'} = $p->{'acceptable'};

    }

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'CONNECT',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub stomp {
    my $self = shift;
    my $p = validate_params(\@_, {
        -login      => { optional => 1, default => undef },
        -passcode   => { optional => 1, default => undef },
        -prefetch   => { optional => 1, default => undef },
        -host       => { optional => 1, default => 'localhost' },
        -heart_beat => { optional => 1, default => '0,0', regex => qr/\d+,\d+/ },
        -acceptable => {
            optional => 1,
            default  => '1.0,1.1,1.2',
            callbacks => {
                'valid target' => \&_match
            }
        }
    });

    my $frame;
    my $header = {};

    if ($self->target == 1.0) {

        $self->throw_msg(
            'xas.lib.stomp.utils.stomp.nosup',
            'stomp_no_support',
            $self->target,
            'stomp'
        );

    }

    $header->{'login'}         = $p->{'login'}    if (defined($p->{'login'}));
    $header->{'passcode'}      = $p->{'passcode'} if (defined($p->{'passcode'}));
    $header->{'prefetch-size'} = $p->{'prefetch'} if (defined($p->{'prefetch'}));

    if ($self->target > 1.0) {

        $header->{'host'}           = $p->{'host'};
        $header->{'heart-beat'}     = $p->{'heart_beat'};
        $header->{'accept-version'} = $p->{'acceptable'};

    }

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'STOMP',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub subscribe {
    my $self = shift;
    my $p = validate_params(\@_, {
        -destination  => 1,
        -prefetch     => { optional => 1, default => 0 },
        -id           => { optional => 1, default => undef },
        -receipt      => { optional => 1, default => undef },
        -ack          => { optional => 1, default => 'auto', regex => qr/auto|client|client\-individual/ },
    });

    my $frame;
    my $header = {};

    $header->{'ack'}            = $p->{'ack'};
    $header->{'prefetch-count'} = $p->{'prefetch'};
    $header->{'destination'}    = $p->{'destination'};
    $header->{'receipt'}        = $p->{'receipt'} if (defined($p->{'receipt'}));

    if (defined($p->{'id'})) {

        $header->{'id'} = $p->{'id'};

    } else {

        # v1.1 and greater must have an id header

        if ($self->target > 1.0) {

            $self->throw_msg(
                'xas.lib.stomp.utils.subscribe',
                'stomp_no_id',
                $self->target

            );

        }

    }

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'SUBSCRIBE',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub unsubscribe {
    my $self = shift;
    my $p = validate_params(\@_, {
        -id           => { optional => 1, default => undef },
        -destination  => { optional => 1, default => undef },
        -receipt      => { optional => 1, default => undef },
    });

    my $frame;
    my $header = {};

    $header->{'receipt'} = $p->{'receipt'} if (defined($p->{'receipt'}));

    # v1.0 should have either a destination and/or id header
    # v1.1 and greater may have a destination header

    if (defined($p->{'destination'}) && defined($p->{'id'})) {

        $header->{'id'}          = $p->{'id'};
        $header->{'destination'} = $p->{'destination'};

    } elsif (defined($p->{'destination'})) {

        $header->{'destination'} = $p->{'destination'};

    } elsif (defined($p->{'id'})) {

        $header->{'id'} = $p->{'id'};

    } else {

        $self->throw_msg(
            'xas.lib.stomp.utils.unsubscribe.invparams',
            'stomp_invalid_params',
            $self->target
        );

    }

    if ($self->target > 1.0) {

        # v1.1 and greater must have an id header

        unless (defined($header->{'id'})) {

            $self->throw_msg(
                'xas.lib.stomp.utils.unsubscribe.noid',
                'stomp_no_id',
                $self->target
            );

        }

    }

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'UNSUBSCRIBE',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub begin {
    my $self = shift;
    my $p = validate_params(\@_, {
        -transaction => 1,
        -receipt     => { optional => 1, default => undef },
    });

    my $frame;
    my $header = {};

    $header->{'transaction'} = $p->{'transaction'};
    $header->{'receipt'}     = $p->{'receipt'} if (defined($p->{'receipt'}));

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'BEGIN',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub commit {
    my $self = shift;
    my $p = validate_params(\@_, {
        -transaction => 1,
        -receipt     => { optional => 1, default => undef },
    });

    my $frame;
    my $header = {};

    $header->{'transaction'} = $p->{'transaction'};
    $header->{'receipt'}     = $p->{'receipt'} if (defined($p->{'receipt'}));

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'COMMIT',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub abort {
    my $self = shift;
    my $p = validate_params(\@_, {
        -transaction => 1,
        -receipt     => { optional => 1, default => undef },
    });

    my $frame;
    my $header = {};

    $header->{'transaction'} = $p->{'transaction'};
    $header->{'receipt'}     = $p->{'receipt'} if (defined($p->{'receipt'}));

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'ABORT',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub ack {
    my $self = shift;
    my $p = validate_params(\@_, {
        -message_id   => 1,
        -subscription => { optional => 1, default => undef },
        -receipt      => { optional => 1, default => undef },
        -transaction  => { optional => 1, default => undef },
    });

    my $frame;
    my $header = {};

    $header->{'receipt'}     = $p->{'receipt'}     if (defined($p->{'receipt'}));
    $header->{'transaction'} = $p->{'transaction'} if (defined($p->{'transaction'}));

    if ($self->target < 1.2) {

        $header->{'message-id'} = $p->{'message_id'};

    } else {

        $header->{'id'} = $p->{'message_id'};

    }

    if (defined($p->{'subscription'})) {

        $header->{'subscription'} = $p->{'subscription'};

    } else {

        if ($self->target > 1.0) {

            $self->throw_msg(
                'xas.lib.stomp.utils.ack.nosup',
                'stomp_no_subscription',
                $self->target
            );

        }

    }

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'ACK',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub nack {
    my $self = shift;
    my $p = validate_params(\@_, {
        -message_id   => 1,
        -receipt      => { optional => 1, default => undef },
        -subscription => { optional => 1, default => undef },
        -transaction  => { optional => 1, default => undef },
    });

    my $frame;
    my $header = {};

    $header->{'receipt'}     = $p->{'receipt'}     if (defined($p->{'receipt'}));
    $header->{'transaction'} = $p->{'transaction'} if (defined($p->{'transaction'}));

    if ($self->target == 1.0) {

        $self->throw_msg(
            'xas.lib.stomp.utils.nack',
            'stomp_no_support',
            $self->target,
            'nack'
        );

    }

    if ($self->target < 1.2) {

        $header->{'message-id'} = $p->{'message_id'};

    } else {

        $header->{'id'} = $p->{'message_id'};

    }

    if (defined($p->{'subscription'})) {

        $header->{'subscription'} = $p->{'subscription'};

    } else {

        if ($self->target > 1.0) {

            $self->throw_msg(
                'xas.lib.stomp.utils.nact',
                'stomp_no_support',
                $self->target,
                'nack'
            );

        }

    }

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'NACK',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub disconnect {
    my $self = shift;
    my $p = validate_params(\@_, {
        -receipt => { optional => 1, default => undef }
    });

    my $frame;
    my $header = {};

    $header->{'receipt'} = $p->{'receipt'} if (defined($p->{'receipt'}));

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'DISCONNECT',
        -headers => $header,
        -body    => ''
    );

    return $frame;

}

sub send {
    my $self = shift;
    my $p = validate_params(\@_, {
        -destination => 1,
        -message     => 1,
        -receipt     => { optional => 1, default => undef },
        -persistent  => { optional => 1, default => undef },
        -transaction => { optional => 1, default => undef },
        -length      => { optional => 1, default => undef },
        -type        => { optional => 1, default => 'text/plain' },
    });

    my $frame;
    my $header = {};
    my $body = $p->{'message'};

    $header->{'destination'} = $p->{'destination'};
    $header->{'receipt'}     = $p->{'receipt'}     if (defined($p->{'receipt'}));
    $header->{'persistent'}  = $p->{'persistent'}  if (defined($p->{'presistent'}));
    $header->{'transaction'} = $p->{'transaction'} if (defined($p->{'transaction'}));
    {
        use bytes;
        $header->{'content-length'} = defined($p->{'length'}) ? $p->{'length'} : length($body);
    }

    if ($self->target > 1.0) {

        $header->{'content-type'} = $p->{'type'};

    }

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'SEND',
        -headers => $header,
        -body    => $body
    );

    return $frame;

}

sub noop {
    my $self = shift;

    my $frame;

    if ($self->target == 1.0) {

        $self->throw_msg(
            'xas.lib.stomp.utils.noop.nosup',
            'stomp_no_support',
            $self->target,
            'noop'
        );

    }

    $frame = XAS::Lib::Stomp::Frame->new(
        -command => 'NOOP',
        -headers => {},
        -body    => ''
    );

    return $frame;

}

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

sub init {
    my $class = shift;

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

    unless (defined($self->{target})) {

        $self->{target} = $self->env->mqlevel;

    }

    return $self;

}

sub _match {
    my $buffer = shift;

    foreach my $item (split(',', $buffer)) {

        return 0 if ($item !~ m/\d\.\d/);

    }

    return 1;

}

1;

__END__