File Coverage

blib/lib/Spread/Messaging/Transport.pm

Criterion	Covered	Total	%
statement	10	12	83.3
branch			n/a
condition			n/a
subroutine	4	4	100.0
pod			n/a
total	14	16	87.5

package Spread::Messaging::Transport;

use 5.008;
use strict;
use warnings;

use Spread;
use Spread::Messaging::Exception;

require Exporter;

our @ISA = qw(Exporter);

# Items to export into callers namespace by default. Note: do not export
# names by default without a very good reason. Use EXPORT_OK instead.
# Do not simply export all your public functions/methods/constants.

# This allows declaration	use Spread::Transport ':all';
# If you do not need this, moving things directly into @EXPORT or @EXPORT_OK
# will save memory.

our %EXPORT_TAGS = ( 'all' => [ qw(

) ] );

our @EXPORT_OK = ( @{ $EXPORT_TAGS{'all'} } );

our @EXPORT = qw(
    UNRELIABLE_MESS
    RELIABLE_MESS
    FIFO_MESS
    CAUSAL_MESS
    AGREED_MESS
    SAFE_MESS
    REGULAR_MESS
    SELF_DISCARD
    DROP_RECV
    REG_MEMB_MESS
    TRANSITION_MESS
    CAUSED_BY_JOIN
    CAUSED_BY_LEAVE
    CAUSED_BY_DISCONNECT
    CAUSED_BY_NETWORK
    MEMBERSHIP_MESS
    ACCEPT_SESSION
    ILLEGAL_GROUP
    ILLEGAL_MESSAGE
    ILLEGAL_SERVICE
    ILLEGAL_SESSION
    ILLEGAL_SPREAD
    CONNECTION_CLOSED
    COULD_NOT_CONNECT
    BUFFER_TOO_SHORT
    GROUPS_TOO_SHORT
    MESSAGE_TOO_LONG
    REJECT_ILLEGAL_NAME
    REJECT_NOT_UNIQUE
    REJECT_NO_NAME
    REJECT_QUOTA
    REJECT_VERSION
);

our $VERSION = '0.03';

# -------------------------------------------------------------------------
# public methods
# -------------------------------------------------------------------------

sub new {
    my $proto = shift;
    my %params = @_;

    my $self = {};
    my @foo = split(/\//, $0);
    my $class = ref ($proto) || $proto;

    # Initialize variables - these are defaults.

    $self->{port} = '4803';
    $self->{timeout} = '5';
    $self->{host} = 'localhost';
    $self->{service_type} = SAFE_MESS;
    $self->{private_name} = sprintf("%0s%05d",
                                    substr($ENV{USER} || pop @foo, 0, 5), $$);

    # Parse named parameters, these may overwrite, or suppliment the above.

    my ($k, $v);
    local $_;

    if (defined ($v = delete $params{'-port'})) {

        $self->{port} = $v;

    }

    if (defined ($v = delete $params{'-host'})) {

        $self->{host} = $v;

    }

    if (defined ($v = delete $params{'-private_name'})) {

        $self->{private_name} = $v;

    }

    if (defined ($v = delete $params{'-timeout'})) {

        $self->{timeout} = $v;

    }

    if (defined ($v = delete $params{'-service_type'})) {

        $self->{service_type} = $v;

    }

    bless($self, $class);

    # Open the connection

    $self->connect();

    return $self;

}

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

    my $connection;

    $connection->{spread_name} = $self->{port} . '@' . $self->{host};
    $connection->{private_name} = $self->{private_name};

    undef $sperrno;
    ($self->{mailbox}, $self->{private_group}) = Spread::connect($connection);
    if ($sperrno) {

        Spread::Messaging::Exception->throw(
            errno => $sperrno + 0,
            errstr => $sperrno
        );

    }

}

sub join_group {
    my ($self, $groups) = @_;

    my @groupss = split(',', $groups);

    foreach my $group (@groupss) {

        undef $sperrno;
        Spread::join($self->{mailbox}, $group);
        if ($sperrno) {

            Spread::Messaging::Exception->throw(
                errno => $sperrno + 0,
                errstr => $sperrno
            );

        }

    }

}

sub leave_group {
    my ($self, $groups) = @_;

    my @groupss = split(',', $groups);

    foreach my $group (@groupss) {

        undef $sperrno;
        Spread::leave($self->{mailbox}, $group);
        if ($sperrno) {

            Spread::Messaging::Exception->throw(
                errno => $sperrno + 0,
                errstr => $sperrno
            );

        }

    }

}

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

    undef $sperrno;
    my $size = Spread::poll($self->{mailbox});
    if ($sperrno) {

        Spread::Messaging::Exception->throw(
            errno => $sperrno + 0,
            errstr => $sperrno
        );

    }

    return $size;

}

sub send {
    my ($self, $group, $message, $type) = @_;

    undef $sperrno;
    Spread::multicast($self->{mailbox}, $self->{service_type}, $group, $type, $message);
    if ($sperrno) {

        Spread::Messaging::Exception->throw(
            errno => $sperrno + 0,
            errstr => $sperrno
        );

    }

}

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

    my ($data, $content);

    undef $sperrno;
    my ($service_type, $sender, $groups, $msg_type, $endian, $message) =
        Spread::receive($self->{mailbox}, $self->{timeout});
    if ($sperrno) {

        Spread::Messaging::Exception->throw(
            errno => $sperrno + 0,
            errstr => $sperrno
        );

    }

    $data = $message;

    if ($service_type & MEMBERSHIP_MESS) {

        if ($service_type & REG_MEMB_MESS) {

            $data = _decode_reg_memb_mess($self, $service_type, $message);

        } elsif ($service_type & TRANSITION_MESS) {

            $data = _decode_transition_mess($self, $service_type, $message);

        }

    }

    return $service_type, $sender, $groups, $msg_type, $endian, $data;

}

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

    undef $sperrno;
    Spread::disconnect($self->{mailbox});
    if ($sperrno) {

        Spread::Messaging::Exception->throw(
            errno => $sperrno + 0,
            errstr => $sperrno
        );

    }

}

# -------------------------------------------------------------------------
# public accessors
# -------------------------------------------------------------------------

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

    return($self->{mailbox});

}

sub service_type {
    my ($self, $p) = @_;

    $self->{service_type} = $p if ((defined $p) &&
                                   (($p == UNRELIABLE_MESS) ||
                                    ($p == RELIABLE_MESS) ||
                                    ($p == FIFO_MESS) ||
                                    ($p == CAUSAL_MESS) ||
                                    ($p == AGREED_MESS) ||
                                    ($p == SAFE_MESS)));
    return($self->{service_type});

}

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

    $self->{timeout} = $p if defined $p;
    return($self->{timeout});

}

sub host {
    my ($self, $p) = @_;

    $self->{host} = $p if defined $p;
    return($self->{host});

}

sub port {
    my ($self, $p) = @_;

    $self->{port} = $p if defined $p;
    return($self->{port});

}

sub private_group {
    my ($self, $p) = @_;

    $self->{private_group} = $p if defined $p;
    return($self->{private_group});

}

sub private_name {
    my ($self, $p) = @_;

    $self->{private_name} = $p if defined $p;
    return($self->{private_name});

}

DESTROY {
    my ($self) = @_;

    Spread::disconnect($self->{mailbox}) if defined $self->{mailbox};

}

# -------------------------------------------------------------------------
# Private methods
# -------------------------------------------------------------------------

sub _decode_reg_memb_mess {
    my ($self, $service_type, $message) = @_;

    # Try to decode the message buffer. This really should
    # be done in the xs code, not here in perl. And the returned
    # buffer doesn't match what the Spread C API documentation says
    # should be there. Arrgh...
    #
    # The first 12 bytes is the group ID, the remainer of the
    # buffer is the current private group of this session.
    #
    # It would have been really nice if the buffer had been broken up into
    # delimited fields, of which the first three fields would be the group ID,
    # the next field the private group, the next field a count
    # and the rest the fields were the actual names of the group memebers,
    # the number of whom would have matched the count. This would have
    # been a nice approximation of the actual C data structure.
    #
    # Oh well, such is life.

    my @data;

    if (($service_type & CAUSED_BY_JOIN) ||
        ($service_type & CAUSED_BY_LEAVE) ||
        ($service_type & CAUSED_BY_DISCONNECT)) {

        @data = unpack("I[3]A*", $message);

    } elsif ($service_type & CAUSED_BY_NETWORK) {

        @data = unpack("I[3]A*", $message);

    }

    return \@data;

}

sub _decode_transition_mess {
    my ($self, $service_type, $message) = @_;

    # Try to decode the message buffer. This really should
    # be done in the xs code, not here in perl.

    my @data = unpack("I3", $message);

    return \@data;

}

1;

__END__