File Coverage

blib/lib/Inferno/RegMgr/TCP.pm

Criterion	Covered	Total	%
statement	16	18	88.8
branch			n/a
condition			n/a
subroutine	6	6	100.0
pod			n/a
total	22	24	91.6

package Inferno::RegMgr::TCP;

use warnings;
use strict;
use Carp;

# update POD & Changes & README
use version; our $VERSION = qv('0.1.3');

# update DEPENDENCIES in POD & Makefile.PL & README
use Scalar::Util qw( weaken );
use IO::Stream;
use Inferno::RegMgr::Utils qw( run_callback quote attr parse_svc );


use constant PORT_NEW   => 6701;
use constant PORT_FIND  => 6702;
use constant PORT_EVENT => 6703;
use constant KiB        => 1024;    ## no critic (Capitalization)


sub new {
    my ($class, $opt) = @_;
    croak '{host} required' if !defined $opt->{host};
    my $self = {
        host        => $opt->{host},
        port_new    => $opt->{port_new}   || PORT_NEW,
        port_find   => $opt->{port_find}  || PORT_FIND,
        port_event  => $opt->{port_event} || PORT_EVENT,
    };
    return bless $self, $class;
}

sub open_event {
    my ($self, $opt) = @_;
    croak '{cb} required' if !defined $opt->{cb};
    my $io = IO::Stream->new({
        host        => $self->{host},
        port        => $self->{port_event},
        cb          => \&_cb_event,
        wait_for    => IN|EOF,
        User_cb     => [ $opt->{cb}, $opt->{method} ],
        Is_connected=> 0,
    });
    return $io;
}

sub _cb_event {
    my ($io, $e, $err) = @_;
    if (!$io->{Is_connected}) {
        if ($e & IN) {
            if ($io->{in_buf} =~ s/\AREADY\n//xms) {
                $io->{Is_connected} = 1;
                $e |= CONNECTED;
            }
            else {
                $err = "Bug in registry: expected 'READY\\n', got '$io->{in_buf}'";
            }
        }
    }
    else {
        if ($err) {
            $e |= EOF;
        }
    }
    if ($e & IN && $io->{in_buf} !~ s/.*\n//xms) {
        $e &= ~IN;
        if (length $io->{in_buf} > KiB) {
            $err = 'Bug in registry: got 1 KiB without \\n';
        }
    }
    if ($e & EOF || $err) {
        $io->close();
    }
    if ($e || $err) {
        run_callback( @{ $io->{User_cb} }, $e, $err );
    }
    return;
}

sub open_new {
    my ($self, $opt) = @_;
    croak '{name} required' if !defined $opt->{name};
    croak '{cb} required'   if !defined $opt->{cb};
    my $io = IO::Stream->new({
        host        => $self->{host},
        port        => $self->{port_new},
        cb          => \&_cb_new,
        wait_for    => 0,
        User_cb     => [ $opt->{cb}, $opt->{method} ],
    });
    $io->write(sprintf "%s %s\n",  quote($opt->{name}), attr($opt->{attr}));
    return $io;
}

sub _cb_new {
    my ($io, $e, $err) = @_;
    $io->close();
    run_callback( @{ $io->{User_cb} }, $err );
    return;
}

sub update {
    my ($self, $io, $attr) = @_;
    $io->write(sprintf "%s\n", attr($attr));
    return;
}

sub open_find {
    my ($self, $opt) = @_;
    croak '{cb} required'   if !defined $opt->{cb};
    my $io = IO::Stream->new({
        host        => $self->{host},
        port        => $self->{port_find},
        cb          => \&_cb_find,
        wait_for    => EOF,
        in_buf_limit=> KiB*KiB,
        User_cb     => [ $opt->{cb}, $opt->{method} ],
    });
    $io->write(sprintf "%s\n", attr($opt->{attr}));
    return $io;
}

sub _cb_find {
    my ($io, $e, $err) = @_;
    my $svc;
    if ($e & EOF && !$err) {
        ($svc, $err) = parse_svc($io->{in_buf});
    }
    $io->close();
    run_callback( @{ $io->{User_cb} }, $svc, $err );
    return;
}


1; # Magic true value required at end of module
__END__