File Coverage

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

Criterion	Covered	Total	%
statement	64	141	45.3
branch	16	78	20.5
condition	0	7	0.0
subroutine	14	17	82.3
pod	1	1	100.0
total	95	244	38.9

#+##############################################################################
#                                                                              #
# File: Net/STOMP/Client/Connection.pm                                         #
#                                                                              #
# Description: Connection support for Net::STOMP::Client                       #
#                                                                              #
#-##############################################################################

#
# module definition
#

package Net::STOMP::Client::Connection;
use strict;
use warnings;
our $VERSION  = "2.3";
our $REVISION = sprintf("%d.%02d", q$Revision: 2.6 $ =~ /(\d+)\.(\d+)/);

#
# used modules
#

use IO::Socket::INET qw();
use List::Util qw(shuffle);
use Net::STOMP::Client::Peer qw();
use No::Worries::Die qw(dief);
use No::Worries::File qw(file_read);
use No::Worries::Log qw(log_debug);
use Params::Validate qw(validate :types);
use Time::HiRes qw();

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

#
# convert a URI (with no options) to a peer object
#

sub _uri2peer ($) {
    my($uri) = @_;

    if ($uri =~ m{ ^ (tcp|ssl|stomp|stomp\+ssl)
                     \:\/\/ ([_a-z0-9\.\-]+) \: (\d+) \/? $ }ix) {
        return(Net::STOMP::Client::Peer->new(
            proto => $1,
            host  => $2,
            port  => $3,
        ));
    } else {
        dief("unexpected server uri: %s", $uri);
    }
}

#
# set the default connection options
#

sub _default_options ($) {
    my($option) = @_;

    $option->{randomize}  =  1;
    $option->{sleep}      =  0.01;
    $option->{max_sleep}  = 30;
    $option->{multiplier} =  2;
}

#
# parse an option string (we do not complain about unknown options)
#

sub _parse_options ($$) {
    my($option, $string) = @_;

    if ($string =~ /\b(backOffMultiplier=(\d+(\.\d+)?))\b/) {
        $option->{multiplier} = $2;
    }
    if ($string =~ /\b(useExponentialBackOff=false)\b/) {
        $option->{multiplier} = 0;
    }
    if ($string =~ /\b(randomize=false)\b/) {
        $option->{randomize} = 0;
    }
    if ($string =~ /\b(initialReconnectDelay=(\d+))\b/) {
        $option->{sleep} = $2 / 1000;
    }
    if ($string =~ /\b(maxReconnectDelay=(\d+))\b/) {
        $option->{max_sleep} = $2 / 1000;
    }
    if ($string =~ /\b(maxReconnectAttempts=(\d+))\b/) {
        $option->{max_attempt} = $2 + 1;
    }
}

#
# parse a connection URI
#
# supported URIs:
#  - tcp://foo:12
#  - file:/foo/bar
#  - ActiveMQ failover URIs
#

sub _parse_uri ($) {
    my($uri) = @_;
    my(@peers, %option, @list);

    while (1) {
        if ($uri =~ /^file:(.+)$/) {
            # list of URIs stored in a file, one per line
            @list = ();
            foreach my $line (split(/\n/, file_read($1))) {
                $line =~ s/\#.*//;
                $line =~ s/\s+//g;
                push(@list, $line) if length($line);
            }
            if (@list == 1) {
                # if only one, allow failover syntax for it
                $uri = shift(@list);
            } else {
                # otherwise, they must be simple URIs
                @peers = map(_uri2peer($_), @list);
                last;
            }
        } elsif ($uri =~ m{ ^ failover \: (?:\/\/)? \( ([_a-z0-9\.\-\:\/\,]+) \)
                            ( \? [_a-z0-9\.\=\-\&]+ ) ? $ }ix) {
            # failover with options
            _default_options(\%option);
            _parse_options(\%option, $2) if $2;
            @peers = map(_uri2peer($_), split(/,/, $1));
            last;
        } elsif ($uri =~ m{ ^ failover \: ([_a-z0-9\.\-\:\/\,]+) $ }ix) {
            # failover without options
            _default_options(\%option);
            @peers = map(_uri2peer($_), split(/,/, $1));
            last;
        } else {
            # otherwise this must be a simple URI
            @peers = (_uri2peer($uri));
            last;
        }
    }
    dief("empty server uri: %s", $uri) unless @peers;
    return(\@peers, \%option);
}

#
# attempt to connect to one peer (low level)
#

sub _attempt ($%) {
    my($peer, %sockopt) = @_;
    my($socket);

    # options sanity
    $sockopt{Proto} = "tcp"; # yes, even SSL is TCP...
    $sockopt{PeerAddr} = $peer->host();
    $sockopt{PeerPort} = $peer->port();
    # try to connect
    if ($peer->proto() =~ /\b(ssl)\b/) {
        # with SSL
        unless ($IO::Socket::SSL::VERSION) {
            eval { require IO::Socket::SSL };
            return(sprintf("cannot load IO::Socket::SSL: %s", $@)) if $@;
        }
        $socket = IO::Socket::SSL->new(%sockopt);
        return(sprintf("cannot SSL connect to %s:%d: %s", $peer->host(),
                       $peer->port(), IO::Socket::SSL::errstr()))
            unless $socket;
    } else {
        # with plain TCP
        $socket = IO::Socket::INET->new(%sockopt);
        return(sprintf("cannot connect to %s:%d: %s", $peer->host(),
                       $peer->port(), $!))
            unless $socket;
        return(sprintf("cannot binmode(socket): %s", $!))
            unless binmode($socket);
    }
    # so far so good...
    @{ $peer }[3] = $socket->peerhost();
    return($socket);
}

#
# try to connect to a list of peers (high level)
#

sub _try ($$$$) {
    my($peers, $peeropt, $sockopt, $debug) = @_;
    my(@list, $count, $result);

    dief("no peers given!") unless @{ $peers };
    $count = 0;
    while (1) {
        @list = $peeropt->{randomize} ? shuffle(@{ $peers }) : @{ $peers };
        foreach my $peer (@list) {
            $result = _attempt($peer, %{ $sockopt });
            if (ref($result)) {
                log_debug("connect to %s ok: %s", $peer->uri(), $peer->addr())
                    if $debug =~ /\b(connection|all)\b/;
                return($result, $peer);
            } else {
                log_debug("connect to %s failed: %s", $peer->uri(), $result)
                    if $debug =~ /\b(connection|all)\b/;
            }
            $count++;
            if (defined($peeropt->{max_attempt})) {
                last if $count >= $peeropt->{max_attempt};
            }
            if ($peeropt->{sleep}) {
                Time::HiRes::sleep($peeropt->{sleep});
                if ($peeropt->{multiplier}) {
                    $peeropt->{sleep} *= $peeropt->{multiplier};
                    if ($peeropt->{max_sleep} and
                        $peeropt->{sleep} > $peeropt->{max_sleep}) {
                        $peeropt->{sleep} = $peeropt->{max_sleep};
                        delete($peeropt->{multiplier});
                    }
                }
            }
        }
        if (defined($peeropt->{max_attempt})) {
            last if $count >= $peeropt->{max_attempt};
        }
        last unless keys(%{ $peeropt });
    }
    # in case of failure, we only report the last error message...
    dief($result);
}

#+++############################################################################
#                                                                              #
# public function                                                              #
#                                                                              #
#---############################################################################

my %new_options = (
    "host" => {
        optional => 1,
        type     => SCALAR,
        regex    => qr/^[a-z0-9\.\-]+$/,
    },
    "port" => {
        optional => 1,
        type     => SCALAR,
        regex    => qr/^\d+$/,
    },
    "uri" => {
        optional => 1,
        type     => SCALAR,
    },
    "sockopt" => {
        optional => 1,
        type     => HASHREF,
    },
    "debug" => {
        optional => 1,
        type     => UNDEF|SCALAR,
    },
);

sub new (@) {
    my(%option, $proto, $peers, $peeropt);

    %option = validate(@_, \%new_options);
    $option{sockopt} ||= {};
    $option{debug} ||= "";
    if (defined($option{uri})) {
        # by URI
        dief("unexpected server host: %s", $option{host})
            if defined($option{host});
        dief("unexpected server port: %s", $option{port})
            if defined($option{port});
        ($peers, $peeropt) = _parse_uri($option{uri});
    } else {
        # by host + port
        dief("missing server host") unless defined($option{host});
        dief("missing server port") unless defined($option{port});
        $proto = "tcp";
        $proto = "ssl" if grep(/^SSL_/, keys(%{ $option{sockopt} }));
        $peers = [
            Net::STOMP::Client::Peer->new(
                proto => $proto,
                host  => $option{host},
                port  => $option{port},
            ) ];
        $peeropt = {};
    }
    return(_try($peers, $peeropt, $option{sockopt}, $option{debug}));
}

1;

__END__