File Coverage

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

Criterion	Covered	Total	%
statement	29	79	36.7
branch	1	32	3.1
condition	0	22	0.0
subroutine	9	17	52.9
pod	5	5	100.0
total	44	155	28.3

#+##############################################################################
#                                                                              #
# File: Net/STOMP/Client/HeartBeat.pm                                          #
#                                                                              #
# Description: Heart-beat support for Net::STOMP::Client                       #
#                                                                              #
#-##############################################################################

#
# module definition
#

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

#
# used modules
#

use No::Worries::Die qw(dief);
use No::Worries::Export qw(export_control);
use Params::Validate qw(validate_pos :types);
use Time::HiRes qw();

#
# get/set the client heart-beat
#

sub client_heart_beat : method {
    my($self, $value);

    $self = shift(@_);
    return($self->{"client_heart_beat"}) if @_ == 0;
    $value = $_[0];
    if (@_ == 1 and defined($value) and ref($value) eq "") {
        $self->{"client_heart_beat"} = $value;
        return($self);
    }
    # otherwise complain...
    validate_pos(@_, { optional => 1, type => SCALAR });
}

#
# get/set the server heart-beat
#

sub server_heart_beat : method {
    my($self, $value);

    $self = shift(@_);
    return($self->{"server_heart_beat"}) if @_ == 0;
    $value = $_[0];
    if (@_ == 1 and defined($value) and ref($value) eq "") {
        $self->{"server_heart_beat"} = $value;
        return($self);
    }
    # otherwise complain...
    validate_pos(@_, { optional => 1, type => SCALAR });
}

#
# get the last time we received/read data
#

sub last_received : method {
    my($self) = @_;

    return(undef) unless $self->{"io"};
    return($self->{"io"}{"incoming_time"});
}

#
# get the last time we sent/wrote data
#

sub last_sent : method {
    my($self) = @_;

    return(undef) unless $self->{"io"};
    return($self->{"io"}{"outgoing_time"});
}

#
# send a NOOP frame only if needed wrt heart-beat settings
#

sub beat : method {
    my($self, %option) = @_;
    my($delta, $sent);

    # check if client heart-beats are expected
    $delta = $self->client_heart_beat();
    return($self) unless $delta;
    # check the last time we sent data
    $sent = $self->last_sent();
    return($self) if Time::HiRes::time() - $sent < $delta / 2;
    # send a NOOP frame
    return($self->noop(%option));
}

#
# setup
#

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

    # additional options for new()
    return(
        "client_heart_beat" => { optional => 1, type => SCALAR },
        "server_heart_beat" => { optional => 1, type => SCALAR },
    ) unless $self;
}

#
# hook for the CONNECT frame
#

sub _connect_hook ($$) {
    my($self, $frame) = @_;
    my($chb, $shb);

    # do not override what the user did put in the frame
    return if defined($frame->header("heart-beat"));
    # do nothing when only STOMP 1.0 is asked
    return unless grep($_ ne "1.0", $self->accept_version());
    # add the appropriate header (in milliseconds!)
    $chb = int(($self->client_heart_beat() || 0) * 1000.0 + 0.5);
    $shb = int(($self->server_heart_beat() || 0) * 1000.0 + 0.5);
    $frame->header("heart-beat", "$chb,$shb") if $chb or $shb;
}

#
# negotiation helper
#

sub _maxif ($$) {
    my($x, $y) = @_;

    return(0) unless $x and $y;
    return($x > $y ? $x : $y);
}

#
# hook for the CONNECTED frame
#

sub _connected_hook ($$) {
    my($self, $frame) = @_;
    my($value, $shb, $chb);

    $value = $frame->header("heart-beat");
    if (defined($value)) {
        # given specification: check
        if ($value =~ /^(\d+),(\d+)$/) {
            ($shb, $chb) = ($1 / 1000.0, $2 / 1000.0);
            $self->server_heart_beat(_maxif($self->server_heart_beat(), $shb));
            $self->client_heart_beat(_maxif($self->client_heart_beat(), $chb));
        } else {
            dief("unexpected heart-beat specification: %s", $value);
        }
    } else {
        # missing specification: disable
        $self->client_heart_beat(0);
        $self->server_heart_beat(0);
    }
}

#
# register the setup and hooks
#

{
    no warnings qw(once);
    $Net::STOMP::Client::Setup{"heart-beat"} = \&_setup;
    $Net::STOMP::Client::Hook{"CONNECT"}{"heart-beat"} = \&_connect_hook;
    $Net::STOMP::Client::Hook{"CONNECTED"}{"heart-beat"} = \&_connected_hook;
}

#
# export control
#

sub import : method {
    my($pkg, %exported);

    $pkg = shift(@_);
    grep($exported{$_}++, qw(beat));
    grep($exported{$_}++, map("${_}_heart_beat", qw(client server)));
    grep($exported{$_}++, map("last_${_}", qw(received sent)));
    export_control(scalar(caller()), $pkg, \%exported, @_);
}

1;

__END__