File Coverage

blib/lib/Spread/Client.pm

Criterion	Covered	Total	%
statement	13	15	86.6
branch			n/a
condition			n/a
subroutine	5	5	100.0
pod			n/a
total	18	20	90.0

package Spread::Client;

use 5.006;
use strict;
use warnings;

use bytes;
use Spread::Client::Constant ':all';
use Spread::Client::Frame;
use List::Util qw(reduce);
use Socket;

our $VERSION = '0.03_02';

sub connect {
    my %args = @_;

    my ($port, $host) = split /@/, $args{spread_name};

    die "Bad/missing SPREAD NAME.\n"
        unless defined $port;

    my $connect_class = $args{connect_class} ? "Spread::Client::Connection::$args{connect_class}"
                                             : "Spread::Client::Connection::Sync";

    eval "require $connect_class";

    die $@
        if $@;

    my $conn = $connect_class->new( @_ )
                or die "Could not connect to spread daemon: $!\n";

    my $sock = $conn->sock;

    # In case connection class sets socket to non-blocking
    # we will set this to non-blocking again below
    unless( defined $sock->blocking( 1 ) ) {
         die "could not set blocking on handle: $!\n";
    }

    # CONNECT ROUTINE
    unless( $conn->is_unix ) {

        my $addr = sockaddr_in($port, inet_aton( $host ));
        connect( $sock, $addr)
            or die "Could not connect to host: $host\n";
    }

    # TODO revisit this at a later date
    my $connect_message = build_connect_message( private_name => $args{private_name} || 0);

    # Send connect message
    $conn->write( \$connect_message );

    # read buffer
    my $buffer;

    # Get auth methods length, to find what auth methods we have available
    $buffer = $conn->read( 1 );
    my $authlen = ord( $$buffer );

    die "Bad authentication length: $authlen\n"
        if $authlen == -1 or $authlen >= 128;

    # Get auth methods
    my $auth_method = $conn->read( $authlen );

    # Send auth message
    my $auth_method_message = build_auth_message( auth_method => $$auth_method );
    $conn->write( \$auth_method_message );

    # Pull down accept and versions and grouplen
    $buffer = $conn->read( 5 );

    my (@versions, $accept, $full_version);

    # Get accept
    $accept = unpack('c', $$buffer);

    die "Did not get accept: $accept\n"
        unless $accept == 1;
    #

    $versions[0] = substr( $$buffer, 1, 1); # Major
    $versions[1] = substr( $$buffer, 2, 1); # Minor
    $versions[2] = substr( $$buffer, 3, 1); # Patch

    {
        no warnings 'once';
        $full_version = unpack('c', reduce { $a | $b } (@versions));
    }

    die "Full version is a problem: $full_version\n"
        if $full_version == -1;

    # Get private group info
    # Get group length
    my $grouplen = unpack('c', substr($$buffer, 4, 1));

    die "We had an issue with the group length\n"
        if $grouplen == -1;

    # Get actual name private group name
    $buffer = $conn->read( $grouplen );
    my $private_group = unpack("a*", $$buffer);
    $conn->private_group( $private_group );

    if( $conn->is_async ) {
        $sock->blocking( 0 );
        $conn->listen_for_messages;
        $conn->session_connected( 1 );
    }

    return $conn;
}

sub receive {
    my %args = @_;

    if( $args{conn}->is_async ) {

        $args{conn}->receive;

        unless( $args{conn}->message_callback ) {
            return $args{conn}->get_queued_messages;
        }
    }
    else {
        my $buffer;

        # pull down message header
        $buffer = $args{conn}->read( 48 );

        # define variables we need
        my ($mess_type, $endian);

        # unpack header data
        my ($service_type, $sender, $num_members, $hint, $data_len) = parse_message_header( $buffer );

        # Get group data
        my $pull_bytes = $num_members * 32;
        $buffer = $args{conn}->read( $pull_bytes );

        my $groups = parse_message_groups( $buffer, $num_members);

        # pull down message body
        $buffer = $args{conn}->read( $data_len );

        my $message = parse_message_body( $buffer, $service_type);

        return ($service_type, $sender, $groups, $mess_type, $endian, $message);
    }
}

sub multicast {
    my %args = @_;

    my $private_group = $args{conn}->private_group;

    my $multi_message = build_standard_message( type       => $args{type} || DEFAULT_SEND_MESS,
                                                from_group => $private_group,
                                                to_groups  => $args{groups},
                                                data       => $args{message},
                                             );

    my $ret = $args{conn}->write( \$multi_message );

    return $ret;
}

sub join {
    my %args = @_;

    my $private_group = $args{conn}->private_group;

    my $join_message = build_standard_message( type       => JOIN_MESS,
                                               from_group => $private_group,
                                               to_groups  => $args{groups},
                                            );

    my $ret = $args{conn}->write( \$join_message );

    return $ret;
}

sub leave {
    my %args = @_;

    my $private_group = $args{conn}->private_group;

    my $leave_message = build_standard_message( type       => LEAVE_MESS,
                                                from_group => $private_group,
                                                to_groups  => $args{groups},
                                             );

    my $ret = $args{conn}->write( \$leave_message );

    return $ret;
}

sub disconnect {
    my %args = @_;

    my $private_group = $args{conn}->private_group;

    my $disconnect_message = build_standard_message( type       => KILL_MESS,
                                                     from_group => $private_group,
                                                     to_groups  => [ $private_group ],
                                                  );
    # write close message to daemon
    $args{conn}->write( \$disconnect_message );

    # close socket
    $args{conn}->close();
}

1;
__END__