File Coverage

blib/lib/Net/STOMP/Client/IO.pm

Criterion	Covered	Total	%
statement	29	135	21.4
branch	0	70	0.0
condition	0	33	0.0
subroutine	10	16	62.5
pod	4	4	100.0
total	43	258	16.6

#+##############################################################################
#                                                                              #
# File: Net/STOMP/Client/IO.pm                                                 #
#                                                                              #
# Description: Input/Output support for Net::STOMP::Client                     #
#                                                                              #
#-##############################################################################

#
# module definition
#

package Net::STOMP::Client::IO;
use 5.005; # need the four-argument form of substr()
use strict;
use warnings;
our $VERSION  = "2.4";
our $REVISION = sprintf("%d.%02d", q$Revision: 2.2 $ =~ /(\d+)\.(\d+)/);

#
# used modules
#

use List::Util qw(min);
use No::Worries::Die qw(dief);
use No::Worries::Log qw(log_debug);
use POSIX qw(:errno_h);
use Time::HiRes qw();

#
# constants
#

use constant READ_LENGTH  => 32_768;  # chunk size for sysread()
use constant WRITE_LENGTH => 32_768;  # minimum length for syswrite()

#+++############################################################################
#                                                                              #
# private helpers                                                              #
#                                                                              #
#---############################################################################

#
# attempt to read data from the socket to the buffer
#
# note: we read at least once even if the buffer contains enough data
#
# common scenarios:
#  - timeout=undef minlen=undef: loop until we successfully read once
#  - timeout=undef minlen=N: loop until we read at least N bytes
#  - timeout=0     minlen=undef: read only once (successful or not)
#  - timeout=0     minlen=N: loop until we read >=N bytes or fail once
#  - timeout=T     minlen=undef: loop until timeout
#  - timeout=T     minlen=N: loop until we read >=N bytes or timeout
#

sub _try_to_read ($$$) {  ## no critic 'ProhibitExcessComplexity'
    my($self, $timeout, $minlen) = @_;
    my($maxtime, $total, $count, $sleeptime, $remaining);

    $self->{incoming_buflen} = length($self->{incoming_buffer});
    # boundary conditions
    if ($timeout) {
        return(0) unless $timeout > 0;
        # timer starts now
        $maxtime = Time::HiRes::time() + $timeout;
    }
    # try to read, in a loop, until we are done
    $total = 0;
    while (1) {
        # attempt to read once
        $count = sysread($self->{socket}, $self->{incoming_buffer},
                         READ_LENGTH, $self->{incoming_buflen});
        if (defined($count)) {
            # we could read this time
            unless ($count) {
                # ... but we hit the EOF
                $self->{error} = "cannot sysread(): EOF";
                return($total);
            }
            # this is a normal successful read
            $self->{incoming_time} = Time::HiRes::time();
            $self->{incoming_buflen} += $count;
            $total += $count;
            # check if we have worked enough
            return($total) unless $minlen and $total < $minlen;
        } else {
            # we could not read this time
            if ($! != EAGAIN and $! != EWOULDBLOCK) {
                # unexpected error
                $self->{error} = "cannot sysread(): $!";
                return(undef);
            }
        }
        # check time
        if (not defined($timeout)) {
            # timeout = undef => loop forever until we are done
            $sleeptime = 0.01;
        } elsif ($timeout) {
            # timeout > 0 => try again only if not too late
            $remaining = $maxtime - Time::HiRes::time();
            return($total) unless $remaining > 0;
            $sleeptime = min($remaining, 0.01);
        } else {
            # timeout = 0 => try again unless last read failed
            return($total) unless $count;
        }
        # sleep a bit...
        Time::HiRes::sleep($sleeptime) unless $count;
    }
}

#
# attempt to write data from the queue and buffer to the socket
#
# common scenarios:
#  - timeout=undef minlen=undef: loop until we successfully write once
#  - timeout=undef minlen=N: loop until we write at least N bytes
#  - timeout=0     minlen=undef: write only once (successful or not)
#  - timeout=0     minlen=N: loop until we write >=N bytes or fail once
#  - timeout=T     minlen=undef: loop until timeout
#  - timeout=T     minlen=N: loop until we write >=N bytes or timeout
#

sub _try_to_write ($$$) {  ## no critic 'ProhibitExcessComplexity'
    my($self, $timeout, $minlen) = @_;
    my($maxtime, $total, $count, $sleeptime, $remaining, $data);

    $self->{outgoing_buflen} = length($self->{outgoing_buffer});
    # boundary conditions
    return(0) unless $self->{outgoing_buflen} or @{ $self->{outgoing_queue} };
    if ($timeout) {
        return(0) unless $timeout > 0;
        # timer starts now
        $maxtime = Time::HiRes::time() + $timeout;
    }
    # try to write, in a loop, until we are done
    $total = 0;
    while (1) {
        # make sure there is enough data in the outgoing buffer
        while ($self->{outgoing_buflen} < WRITE_LENGTH
               and @{ $self->{outgoing_queue} }) {
            $data = shift(@{ $self->{outgoing_queue} });
            $self->{outgoing_buffer} .= ${ $data };
            $self->{outgoing_buflen} += length(${ $data });
        }
        return($total) unless $self->{outgoing_buflen};
        # attempt to write once
        $count = syswrite($self->{socket}, $self->{outgoing_buffer},
                          $self->{outgoing_buflen});
        if (defined($count)) {
            # we could write this time
            if ($count) {
                # this is a normal successful write
                $self->{outgoing_time} = Time::HiRes::time();
                $self->{outgoing_buflen} -= $count;
                $total += $count;
                substr($self->{outgoing_buffer}, 0, $count, "");
                $self->{outgoing_length} -= $count;
                # check if we have worked enough
                return($total) unless $self->{outgoing_buflen}
                                or @{ $self->{outgoing_queue} };
                return($total) unless $minlen and $total < $minlen;
            }
        } else {
            # we could not write this time
            if ($! != EAGAIN and $! != EWOULDBLOCK) {
                # unexpected error
                $self->{error} = "cannot syswrite(): $!";
                return(undef);
            }
        }
        # check time
        if (not defined($timeout)) {
            # timeout = undef => loop forever until we are done
            $sleeptime = 0.01;
        } elsif ($timeout) {
            # timeout > 0 => try again only if not too late
            $remaining = $maxtime - Time::HiRes::time();
            return($total) unless $remaining > 0;
            $sleeptime = min($remaining, 0.01);
        } else {
            # timeout = 0 => try again unless last write failed
            return($total) unless $count;
        }
        # sleep a bit...
        Time::HiRes::sleep($sleeptime) unless $count;
    }
}

#+++############################################################################
#                                                                              #
# object oriented interface                                                    #
#                                                                              #
#---############################################################################

#
# constructor
#

sub new : method {
    my($class, $socket) = @_;
    my($self);

    dief("missing or invalid socket")
        unless $socket and ref($socket) and $socket->isa("IO::Socket");
    $socket->blocking(0);
    $self = {};
    $self->{socket} = $socket;
    $self->{incoming_buffer} = "";
    $self->{incoming_buflen} = 0;
    $self->{outgoing_buffer} = "";
    $self->{outgoing_buflen} = 0; # buffer length only
    $self->{outgoing_queue} = [];
    $self->{outgoing_length} = 0; # buffer + queue length
    return(bless($self, $class));
}

#
# queue the given data (a scalar reference!)
#

sub queue_data : method {
    my($self, $data) = @_;
    my($length);

    dief("unexpected data: %s", $data) unless ref($data) eq "SCALAR";
    $length = length(${ $data });
    if ($length) {
        push(@{ $self->{outgoing_queue} }, $data);
        $self->{outgoing_length} += $length;
    }
    return($self->{outgoing_length});
}

#
# send the queued data
#

sub send_data : method {
    my($self, %option) = @_;
    my($minlen, $count);

    unless ($self->{error}) {
        # send some data
        $minlen = $self->{outgoing_length};
        $count = _try_to_write($self, $option{timeout}, $minlen);
    }
    dief($self->{error}) unless defined($count);
    # so far so good
    log_debug("sent %d bytes", $count)
        if $option{debug} and $option{debug} =~ /\b(io|all)\b/;
    return($count);
}

#
# receive some data
#

sub receive_data : method {
    my($self, %option) = @_;
    my($minlen, $count);

    unless ($self->{error}) {
        # receive some data
        $minlen = $option{timeout} ? 1 : undef;
        $count = _try_to_read($self, $option{timeout}, $minlen);
    }
    dief($self->{error}) unless defined($count);
    # so far so good
    log_debug("received %d bytes", $count)
        if $option{debug} and $option{debug} =~ /\b(io|all)\b/;
    return($count);
}

1;

__END__