File Coverage

blib/lib/POE/Component/IKC/ClientLite.pm

Criterion	Covered	Total	%
statement	32	343	9.3
branch	1	176	0.5
condition	0	71	0.0
subroutine	12	35	34.2
pod	11	12	91.6
total	56	637	8.7

package POE::Component::IKC::ClientLite;

############################################################
# $Id: ClientLite.pm 1247 2014-07-07 09:06:34Z fil $
# By Philp Gwyn
#
# Copyright 1999-2014 Philip Gwyn.  All rights reserved.
# This program is free software; you can redistribute it and/or modify
# it under the same terms as Perl itself.
#
# Contributed portions of IKC may be copyright by their respective
# contributors.

use strict;
use vars qw($VERSION @ISA @EXPORT @EXPORT_OK $error $request);

use Socket;
use IO::Socket;
use IO::Select;
use POE::Component::IKC::Specifier;
use POE::Component::IKC::Protocol;
use Data::Dumper;
use POSIX qw(:errno_h);
use Carp;

require Exporter;
@ISA = qw(Exporter);
@EXPORT = qw(create_ikc_client);
$VERSION = '0.2402';

sub DEBUG { 0 }

$request=0;

###############################################################################
sub spawn
{
    my( $package, %parms ) = @_;

#    $parms{on_connect}||=sub{};         # would be silly for this to be blank
    $parms{ip}||='localhost';
    $parms{port}||=603;                 # POE! (almost :)
    $parms{name}||="Client$$";
    $parms{connect_timeout} ||= $parms{timeout} || 30;
    $parms{timeout}||=30;
    $parms{serialiser}||=_default_freezer();
    $parms{block_size} ||= 65535;
    $parms{protocol} ||= 'IKC0';

    my %self;
    @self{qw(ip port name serialiser timeout connect_timeout block_size protocol)}=
            @parms{qw(ip port name serialiser timeout connect_timeout block_size protocol)};

    eval {
        @{$self{remote}}{qw(freeze thaw)}=_get_freezer($self{serialiser});
    };

    if($@) {
        $self{error}=$error=$@;
        return;
    }
    my $self=bless \%self, $package;
    $self->{remote}{aliases}={};
    $self->{remote}{name}="$self->{ip}:$self->{port}";

    $self->connect and return $self;
    return;
}

sub create_ikc_client
{
    my(%parms)=@_;
    my $package = $parms{package} || __PACKAGE__;
    carp "create_ikc_client is deprecated; use $package->spawn instead";
    $package->spawn( %parms );
}

sub name { $_[0]->{name}; }

#----------------------------------------------------
sub connect
{
    my($self)=@_;
    return 1 if($self->{remote}{connected} and $self->{remote}{socket} and
                $self->ping);  # are we already connected?

    my $remote=$self->{remote};
    delete $remote->{socket};
    delete $remote->{connected};

    my $name=$remote->{name};
    DEBUG && print "Connecting to $name...\n";
    my( $sock, $resp );

    my $DONE = 0;
    eval {
        local $SIG{__DIE__}='DEFAULT';
        local $SIG{__WARN__};
        local $SIG{ALRM} = sub { die "alarm\n" }; # NB: \n required
        $sock=IO::Socket::INET->new( PeerAddr=>$self->{ip},
                                     PeerPort=>$self->{port},
                                     # Proto=>'tcp',
                                     Timeout=>$self->{connect_timeout}
                                    );
        die "Unable to connect to $name: $!\n" unless $sock;
        $sock->autoflush(1);
        local $/="\cM\cJ";
        local $\="\cM\cJ";

        # Attempt IKC0 protocol
        if( $self->{protocol} eq 'IKC0' ) {
            if( $self->_protocol_IKC0( $sock ) ) {
                $DONE = 1;
                return;
            }
        }

        # Fallback to IKC protocol
        $sock->print('HELLO');
        my $resp;

        alarm( $self->{connect_timeout} );
        while (defined($resp=$sock->getline))       # phase 000
        {
            chomp($resp);
            last if $resp eq 'DONE';
            die "Invalid IAM response from $name: $resp\n"
                unless $resp=~/^IAM\s+([-:.\w]+)$/;
            $remote->{name}||=$1;
            $self->{ping}||="poe://$1/IKC/ping";
            $remote->{aliases}->{$1}=1;
            $sock->print('OK');
        }
        die "Phase 000: $!\n" unless defined $resp;


        alarm( $self->{connect_timeout} );
        $sock->print("IAM $self->{name}");          # phase 001
        chomp($resp=$sock->getline);
        die "Phase 001: $!\n" unless defined $resp;
        die "Didn't get OK from $name\n" unless $resp eq 'OK';
        $sock->print("DONE");

        alarm( $self->{connect_timeout} );
        $sock->print("FREEZER $self->{serialiser}");# phase 002
        chomp($resp=$sock->getline);
        die "Phase 002: $!\n" unless defined $resp;
        die "$name refused $self->{serialiser}\n" unless $resp eq 'OK';

        alarm( $self->{connect_timeout} );
        $sock->print('WORLD');                      # phase 003
        chomp($resp=$sock->getline);
        die "Phase 003: $!\n" unless defined $resp;
        die "Didn't get UP from $name\n" unless $resp eq 'UP';
        $DONE = 1;
    };
    alarm( 0 );
    if($@)
    {
        $self->{error}=$error=$@;
        if( $error eq "alarm\n" ) {
            $self->{error}=$error="Timeout connecting to $self->{ip}:$self->{port}";
        }
        return;
    }
    $remote->{socket}=$sock;
    $remote->{connected}=1;
    return 1;
}

#----------------------------------------------------
sub _protocol_IKC0
{
    my( $self, $sock ) = @_;

    my $remote=$self->{remote};
    my $name=$remote->{name};
    my $resp;

    my $setup = POE::Component::IKC::Protocol::__build_setup(
                            [ $self->{name} ], [ $self->{serialiser} ] );
    $sock->print( $setup );
    alarm( $self->{connect_timeout} );
    while (defined($resp=$sock->getline))       # phase 010
    {
        chomp($resp);
        return if $resp eq 'NOT';                 # move to phase 000
        die "Phase 010: Invalid response from $name: $resp\n"
                                unless $resp =~ /^SETUP (.+)$/;
        my $neg = POE::Component::IKC::Protocol::__neg_setup( $1 );
        if( $neg->{bad} ) {
            $sock->print( 'NOT' );
            next;
        }
        die "Phase 010: Refused $self->{serialiser}, wants $neg->{freezer}[0]"
                            unless $neg->{freezer}[0] eq $self->{serialiser};
        $remote->{name} = $neg->{kernel}[0];
        foreach my $a ( @{ $neg->{kernel} } ) {
            $remote->{aliases}{$a} = 1;
        }
        return 1;
    }
}


#----------------------------------------------------
sub error
{
    return $_[0]->{error} if @_==1;
    return $error;
}
#----------------------------------------------------
sub ping
{
    my($self)=@_;
    my $ret=eval {
        my $rsvp={kernel=>$self->{name},
                  session=>'IKC', state=>'pong'
                 };
        my $r=$self->_send_msg({event=>$self->{ping}, params=>'PING',
                                rsvp=>$rsvp});
        return unless $r;
        my $pong=$self->_response($rsvp);
        return 1 if $pong and $pong eq 'PONG';
    };
    $self->{error}=$error=$@ if $@;
    $self->{remote}{connected}=$ret;
    return $ret;
}

#----------------------------------------------------
sub disconnect
{
    my($self)=@_;
    # 2001/01 why did we try to unregister ourselves?  unregister wouldn't
    # be safe for remote kernels anyway
    # $self->call('IKC/unregister', $self->{name}) if $self->{remote};
    delete @{$self->{remote}}{qw(socket connected name aliases)};
    $self->{remote}={};
}

sub DESTROY
{
    my($self)=@_;
    $self->disconnect;
}
sub END
{
    DEBUG and print "end\n";
}

#----------------------------------------------------
# Post an event, maybe waits for a response and throws it away
#
sub post
{
    my($self, $spec, $params)=@_;
    unless(ref $spec or $spec=~m(^poe:)) {

        unless($self->{remote}{name}) {
            $self->{error}=$error="Attempting to post $spec to unknown kernel";
            # carp $error;
            return;
        }

        $spec="poe://$self->{remote}{name}/$spec";
    }

    my $ret=eval {
        return 0 if(0==$self->_try_send({event=>$spec, params=>$params}));
        1;
    };
    if($@) {
        $self->{error}=$error=$@;
        return;
    }
    return $ret;
}

#----------------------------------------------------
# posts an event, waits for the response, returns the response
sub call
{
    my($self, $spec, $params)=@_;
    $spec="poe://$self->{remote}{name}/$spec" unless ref $spec or $spec=~m(^poe:);

    my $rsvp={kernel=>$self->{name}, session=>'IKCLite',
              state=>'response'.$request++};

    my $req={event=>$spec, params=>$params,
             rsvp=>$rsvp, 'wantarray'=>wantarray(),
            };
    my @ret=eval {
        return unless $self->_try_send($req);
        DEBUG && print "Waiting for response...\n";
        return $self->_response($rsvp, $req->{wantarray});
    };
    if($@) {
        $self->{error}=$error=$@;
        return;
    }
    return @ret if $req->{wantarray};
    return $ret[0];
}

#----------------------------------------------------
# posts an event, waits for the response, returns the response
# this differs from call() in that the foreign server may
# need many states before getting a response
sub post_respond
{
    my($self, $spec, $params)=@_;
    $spec="poe://$self->{remote}{name}/$spec" unless ref $spec or $spec=~m(^poe:);

    my $ret;
    my $rsvp={kernel=>$self->{name}, session=>'IKCLite',
              state=>'response'.$request++};
    $ret=eval {
        return unless $self->_try_send({event=>$spec,
                                        params=>[$params, $rsvp],
                                       });
        DEBUG && print "Waiting for response...\n";
        return $self->_response($rsvp);
    };
    if($@) {
        $self->{error}=$error=$@;
        return;
    }
    return $ret;
}

#----------------------------------------------------
sub responded
{
    my( $self, $state ) = @_;

    my $wantarray = wantarray;
    my $rsvp = { kernel=>$self->{name},
                 session=>'IKCLite',
                 state=>$state
               };
    my @ret = eval {
            DEBUG && print "Waiting for response...\n";
            return $self->_response($rsvp, $wantarray);
        };
    if($@) {
        $self->{error}=$error=$@;
        return;
    }
    return @ret if wantarray;
    return $ret[0];
}



#----------------------------------------------------
sub _from
{
    my( $self ) = @_;
    return { kernel => $self->{name},
             session => 'IKCLite',
             # state   => 'IKC:lite'
           }
}

#----------------------------------------------------
sub _try_send
{
    my($self, $msg)=@_;
    return unless $self->{remote}{connected} or $self->connect();

    $msg->{from} ||= $self->_from;

    my $ret=$self->_send_msg($msg);
    DEBUG && print "Sending message...\n";
    if(defined $ret and $ret==0) {
        return 0 unless $self->connect();
        DEBUG && print "Retry message...\n";
        $ret=$self->_send_msg($msg);
    }
    return $ret;
}

#----------------------------------------------------
sub _send_msg
{
    my($self, $msg)=@_;

    my $e=$msg->{rsvp} ? 'call' : 'post';

    my $to=specifier_parse($msg->{event});
    unless($to) {
        croak "Bad message ", Dumper $msg;
    }
    unless($to) {
        warn "Bad or missing 'to' parameter '$msg->{event}' to poe:/IKC/$e\n";
        return;
    }
    unless($to->{session}) {
        warn "Need a session name in poe:/IKC/$e";
        return;
    }
    unless($to->{state})   {
        carp "Need a state name in poe:IKC/$e";
        return;
    }

    my $frozen = $self->{remote}{freeze}->($msg);
    my $raw=length($frozen) . "\0" . $frozen;

    unless($self->{remote}{socket}->opened()) {
        $self->{connected}=0;
        $self->{error}=$error="Socket not open";
        return 0;
    }
    unless($self->{remote}{socket}->syswrite($raw, length $raw)) {
        $self->{connected}=0;
        return 0 if($!==EPIPE);
        $self->{error}=$error="Error writing: $!\n";
        return 0;
    }
    return 1;
}


#----------------------------------------------------
sub _response
{
    my($self, $rsvp, $wantarray)=@_;

    $rsvp=specifier_parse($rsvp);
    my $remote=$self->{remote};

    my $start = time;
    my $stopon = $start + $self->{timeout};

    my $select=IO::Select->new() or die $!;     # create the select object
    $select->add($remote->{socket});

    my(@ready, $s, $raw, $frozen, $ret, $l, $need);
    $raw='';

    my $blocks = 0;
    do {{
        my $timeout = $stopon-time;
        if( $timeout <= 0 ) {
            $timeout = 1;
        }
#        Torture::my_warn( "timeout=$timeout" );
        @ready=$select->can_read( $timeout );   # this is the select
        unless( @ready ) {                     # nothing ready == timeout
            # Torture::my_warn( 'select hates me' );
            last;
        }

        foreach $s (@ready)                     # let's see what's ready...
        {
            die "Hey!  $s isn't $remote->{socket}"
                unless $s eq $remote->{socket};
        }
        DEBUG && print "Got something...\n";

                                    # read in another chunk
        $l = $remote->{socket}->sysread($raw, $self->{block_size},
                                                length($raw));

        unless(defined $l) {                    # disconnect, maybe?
            $remote->{connected}=0 if $!==EPIPE;
            die "Error reading: $!\n";
        }
        $blocks ++;

        if(not $need and $raw=~s/(\d+)\0//s) {  # look for a marker?
            $need=$1 ;
            DEBUG && print "Need $need bytes...\n";
        }

        next unless $need;                      # still looking...

        if(length($raw) >= $need)               # do we have all we want?
        {
            # Torture::my_warn( 'Got it all' );
            DEBUG && print "Got it all...\n";

            $frozen=substr($raw, 0, $need);     # seems so...
            substr($raw, 0, $need)='';
            my $msg=$self->{remote}{thaw}->($frozen);   # thaw the message
            DEBUG && print "msg=", Dumper $msg;
            my $to=specifier_parse($msg->{event});

            die "$msg->{params}\n" if($msg->{is_error});    # throw an error out
            DEBUG && print "Not an error...\n";

                # make sure it's what we're waiting for...
            if($to->{session} ne 'IKC' and $to->{session} ne 'IKCLite')
            {
                warn "Unknown session $to->{session}\n";
                DEBUG && print "Not for us!  ($to->{session})...\n";
                next;
            }
            if($to->{session} ne $rsvp->{session} or
               $to->{state} ne $rsvp->{state})
            {
                warn specifier_name($to). " received, expecting " .
                     specifier_name($rsvp). "\n";
                DEBUG && print "Not for us!  ($to->{session}/$to->{state})...\n";
                next;
            }

            DEBUG and print "wantarray=$wantarray\n";
            if( $wantarray ) {
                DEBUG and print "Wanted an array\n";
                return @{$msg->{params}} if ref $msg->{params} eq 'ARRAY';
            }
            return $msg->{params};              # finaly!
        }
        # Torture::my_warn( "blocks=$blocks l=$l need=$need, got=", length $raw );
    }} while ($stopon >= time) ;     # do it until time's up

    $remote->{connected}=0;
    confess "Timed out waiting for response ", specifier_name( $rsvp );
#    die "Timed out waiting for response ", specifier_name( $rsvp ), "\n",
#        "start=$start stopon=$stopon now=", time;
    return;
}









#------------------------------------------------------------------------------
# Try to require one of the default freeze/thaw packages.
sub _default_freezer
{
  local $SIG{'__DIE__'} = 'DEFAULT';
  my $ret;

  foreach my $p (qw(Storable FreezeThaw POE::Component::IKC::Freezer)) {
    my $q=$p;
    $q=~s(::)(/)g;
    eval { require "$q.pm"; import $p ();};
    DEBUG and warn $@ if $@;
    return $p if $@ eq '';
  }
  die __PACKAGE__." requires Storable or FreezeThaw or POE::Component::IKC::Freezer\n";
}

sub _get_freezer
{
    my($freezer)=@_;
    unless(ref $freezer) {
    my $symtable=$::{"main::"};
    my $loaded=1;                       # find out of the package was loaded
    foreach my $p (split /::/, $freezer) {
        unless(exists $symtable->{"$p\::"}) {
            $loaded=0;
            last;
        }
        $symtable=$symtable->{"$p\::"};
    }

    unless($loaded) {        my $q=$freezer;
        $q=~s(::)(/)g;
        eval {require "$q.pm"; import $freezer ();};
        croak $@ if $@;
      }
    }

    # Now get the methodes we want
    my $freeze=$freezer->can('nfreeze') || $freezer->can('freeze');
    carp "$freezer doesn't have a freeze method" unless $freeze;
    my $thaw=$freezer->can('thaw');
    carp "$freezer doesn't have a thaw method" unless $thaw;

    # If it's an object, we use closures to create a $self->method()
    my $tf=$freeze;
    my $tt=$thaw;
    if(ref $freezer) {
        $tf=sub {  return $freeze->($freezer, @_) };
        $tt=sub {  return ($thaw->($freezer, @_))[0] };
    }
    else {
        # FreezeThaw::thaw returns an array now!  We only want the first
        # element.
        $tt=sub {  return ($thaw->( @_ ))[0] };
    }
    return($tf, $tt);
}

1;

__END__