File Coverage

blib/lib/Test/POE/Client/TCP.pm

Criterion	Covered	Total	%
statement	203	278	73.0
branch	65	128	50.7
condition	16	42	38.1
subroutine	30	38	78.9
pod	13	13	100.0
total	327	499	65.5

package Test::POE::Client::TCP;
$Test::POE::Client::TCP::VERSION = '1.26';
#ABSTRACT: A POE Component providing TCP client services for test cases

use strict;
use warnings;
use POE qw(Wheel::SocketFactory Wheel::ReadWrite Filter::Line);
use POSIX qw[ETIMEDOUT];
use Socket qw(AF_INET AF_INET6 SOCK_STREAM inet_ntoa inet_ntop inet_pton
              unpack_sockaddr_in unpack_sockaddr_in6);
use Carp qw(carp croak);

our $GOT_SSL;

BEGIN {
    eval {
        require POE::Component::SSLify;
        import POE::Component::SSLify qw( Client_SSLify SSLify_ContextCreate );
        $GOT_SSL = 1;
    };
}

sub spawn {
  my $package = shift;
  my %opts = @_;
  $opts{lc $_} = delete $opts{$_} for keys %opts;
  my $options = delete $opts{options};
  my $autoconnect = delete $opts{autoconnect};
  if ( $autoconnect and !( $opts{address} and $opts{port} ) ) {
     carp "You must provide both 'address' and 'port' parameters when specifying 'autoconnect'\n";
     return;
  }
  my $usessl = delete $opts{usessl};
  if ( $usessl && !$GOT_SSL ) {
     carp "'usessl' specified but POE::Component::SSLify is not available\n";
     return;
  }
  delete $opts{timeout} unless $opts{timeout} and $opts{timeout} =~ m!^\d+$!;
  my $self = bless \%opts, $package;
  $self->{usessl}  = $usessl;
  $self->{_prefix} = delete $self->{prefix};
  $self->{_prefix} = 'testc_' unless defined $self->{_prefix};
  $self->{_prefix} .= '_' unless $self->{_prefix} =~ /\_$/;
  $self->{session_id} = POE::Session->create(
	object_states => [
	   $self => { shutdown       => '_shutdown',
		      send_event     => '__send_event',
		      send_to_server => '_send_to_server',
		      disconnect     => '_disconnect',
		      terminate      => '_terminate',
		      connect	     => '_connect',
          _timeout     => '_socket_fail',
	            },
	   $self => [ qw(_start register unregister _socket_up _socket_fail _conn_input _conn_error _conn_flushed _send_to_server __send_event _disconnect) ],
	],
	heap => $self,
	( ref($options) eq 'HASH' ? ( options => $options ) : () ),
	args => [ $autoconnect ],
  )->ID();
  return $self;
}

sub session_id {
  shift->{session_id};
}

sub context {
  my $self = shift;
  return $self->{context} if defined $self->{context};
  return;
}

sub shutdown {
  $poe_kernel->call( shift->{session_id}, 'shutdown' );
}

sub server_info {
  my $self = shift;
  return unless $self->{_server_info};
  my @vals = @{ $self->{_server_info} };
  return @vals if wantarray;
  return { map { $_ => shift @vals } qw(peeraddr peerport sockaddr sockport) };
}

sub connect {
  $poe_kernel->call( shift->{session_id}, 'connect' );
}

sub wheel {
  my $self = shift;
  return unless $self->{socket};
  return $self->{socket};
}

sub alias {
  shift->{alias};
}

sub _start {
  my ($kernel,$self,$sender,$autoconnect) = @_[KERNEL,OBJECT,SENDER,ARG0];
  $self->{session_id} = $_[SESSION]->ID();

  if ( $self->{alias} ) {
	$kernel->alias_set( $self->{alias} );
  }
  else {
	$kernel->refcount_increment( $self->{session_id} => __PACKAGE__ );
  }

  if ( $kernel != $sender ) {
    my $sender_id = $sender->ID;
    $self->{events}->{$self->{_prefix} . 'all'}->{$sender_id} = $sender_id;
    $self->{sessions}->{$sender_id}->{'ref'} = $sender_id;
    $self->{sessions}->{$sender_id}->{'refcnt'}++;
    $kernel->refcount_increment($sender_id, __PACKAGE__);
    $kernel->post( $sender, $self->{_prefix} . 'registered', $self );
    $kernel->detach_myself();
  }

  $kernel->yield( 'connect' ) if $autoconnect and $self->{address} and $self->{port};
  return;
}

sub _connect {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  my $args;
  if ( ref( $_[ARG0] ) eq 'HASH' ) {
    $args = { %{ $_[ARG0] } };
  }
  else {
    $args = { @_[ARG0..$#_] };
  }
  $args->{lc $_} = delete $args->{$_} for keys %{ $args };
  unless ( $self->{address} and $self->{port} ) {
    unless ( $args->{address} and $args->{port} ) {
      carp "You must provide both 'address' and 'port' parameters\n";
      return;
    }
    $self->{address} = $args->{address};
    $self->{port} = $args->{port};
  }
  my $usessl = delete $args->{usessl};
  if ( $usessl && !$GOT_SSL ) {
     carp "'usessl' specified but POE::Component::SSLify is not available\n";
     return;
  }

  $self->{usessl} = $usessl if defined $usessl;

  $self->{localaddr} = $args->{localaddr} if $args->{localaddr};
  $self->{localport} = $args->{localaddr} if $args->{localport};

  if ( $self->{socket} ) {
    carp "Already connected. Disconnect and call 'connect' again\n";
    return;
  }

  if ( $self->{factory} ) {
    carp "Connection already in progress\n";
    return;
  }

  $self->{domain} =
    defined( inet_pton( AF_INET6, $self->{address} ) ) ? AF_INET6 : AF_INET;
  $self->{factory} = POE::Wheel::SocketFactory->new(
    RemoteAddress  => $self->{address},
    RemotePort     => $self->{port},
    ( defined $self->{address} ? ( BindAddress => $self->{localaddr} ) : () ),
    ( defined $self->{port} ? ( BindPort => $self->{localport} ) : () ),
    SuccessEvent   => '_socket_up',
    FailureEvent   => '_socket_fail',
    SocketDomain   => $self->{domain},     # Sets the socket() domain
    SocketType     => SOCK_STREAM,         # Sets the socket() type
    SocketProtocol => 'tcp',               # Sets the socket() protocol
    Reuse          => 'yes',               # Lets the port be reused
  );

  $kernel->delay( '_timeout', $self->{timeout}, 'connect', ETIMEDOUT, POSIX::strerror( ETIMEDOUT ) ) if $self->{timeout};
  return;
}

sub _socket_up {
  my ($kernel,$self,$socket,$peeraddr,$peerport) = @_[KERNEL,OBJECT,ARG0..ARG2];
  my @sockaddr = ( $self->{domain} == AF_INET6 )
               ? unpack_sockaddr_in6 ( CORE::getsockname $socket )
               : unpack_sockaddr_in  ( CORE::getsockname $socket );
  my $sockaddr = inet_ntop( $self->{domain}, $sockaddr[1] );
  my $sockport = $sockaddr[0];
  $peeraddr = inet_ntop( $self->{domain}, $peeraddr );
  $kernel->delay( '_timeout' );

  delete $self->{factory};

  if ( $self->{usessl} && $GOT_SSL ) {
    eval {
        my $ctx;
        if ($self->{sslctx}) {
            $ctx = $self->{sslctx};
        }
        elsif ($self->{sslkey} && $self->{sslcert}) {
            $ctx = SSLify_ContextCreate($self->{sslkey}, $self->{sslcert});
        }
        else {
            $ctx = undef;
        }
        $socket = Client_SSLify( $socket, undef, undef, $ctx );
        if ( $@ ) {
           warn "Couldn't use an SSL socket: $@\n";
           $self->{usessl} = 0;
        }
    };
  }

  $self->{socket} = POE::Wheel::ReadWrite->new(
    Handle => $socket,
    _get_filters(
        $self->{filter},
        $self->{inputfilter},
        $self->{outputfilter}
    ),
    InputEvent => '_conn_input',
    ErrorEvent => '_conn_error',
    FlushedEvent => '_conn_flushed',
  );

  $self->{_server_info} = [ $peeraddr, $peerport, $sockaddr, $sockport ];

  $self->_send_event( $self->{_prefix} . 'connected', $peeraddr, $peerport, $sockaddr, $sockport );
  return;
}

sub _get_filters {
    my ($client_filter, $client_infilter, $client_outfilter) = @_;
    if (defined $client_infilter or defined $client_outfilter) {
      return (
        "InputFilter"  => _load_filter($client_infilter),
        "OutputFilter" => _load_filter($client_outfilter)
      );
      if (defined $client_filter) {
        carp(
          "Filter ignored with InputFilter or OutputFilter"
        );
      }
    }
    elsif (defined $client_filter) {
     return ( "Filter" => _load_filter($client_filter) );
    }
    else {
      return ( Filter => POE::Filter::Line->new(), );
    }

}

# Get something: either arrayref, ref, or string
# Return filter
sub _load_filter {
    my $filter = shift;
    if (ref ($filter) eq 'ARRAY') {
        my @args = @$filter;
        $filter = shift @args;
        if ( _test_filter($filter) ){
            return $filter->new(@args);
        } else {
            return POE::Filter::Line->new(@args);
        }
    }
    elsif (ref $filter) {
        return $filter->clone();
    }
    else {
        if ( _test_filter($filter) ) {
            return $filter->new();
        } else {
            return POE::Filter::Line->new();
        }
    }
}

# Test if a Filter can be loaded, return sucess or failure
sub _test_filter {
    my $filter = shift;
    my $eval = eval {
        (my $mod = $filter) =~ s!::!/!g;
        require "$mod.pm";
        1;
    };
    if (!$eval and $@) {
        carp(
          "Failed to load [$filter]\n" .
          "Reason $@\nUsing defualt POE::Filter::Line "
        );
        return 0;
    }
    return 1;
}

sub _socket_fail {
  my ($kernel,$self,$operation,$errnum,$errstr,$wheel_id) = @_[KERNEL,OBJECT,ARG0..ARG3];
  carp "Wheel $wheel_id generated $operation error $errnum: $errstr\n" if $self->{debug};
  $kernel->delay( '_timeout' );
  delete $self->{factory};
  $self->_send_event( $self->{_prefix} . 'socket_failed', $operation, $errnum, $errstr );
  return;
}

sub disconnect {
  my $self = shift;
  $poe_kernel->call( $self->{session_id}, 'disconnect', @_ );
}

sub _disconnect {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  return unless $self->{socket};
  $self->{_quit} = 1;
  return 1;
}

sub terminate {
  my $self = shift;
  $poe_kernel->call( $self->{session_id}, 'terminate', @_ );
}

sub _terminate {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  return unless $self->{socket};
  if ( $^O =~ /(cygwin|MSWin)/ ) {
    $self->{socket}->shutdown_input();
    $self->{socket}->shutdown_output();
  }
  delete $self->{socket};
  delete $self->{_server_info};
  $self->_send_event( $self->{_prefix} . 'disconnected' );
  return 1;
}

sub _conn_input {
  my ($kernel,$self,$input,$id) = @_[KERNEL,OBJECT,ARG0,ARG1];
  $self->_send_event( $self->{_prefix} . 'input', $input );
  return;
}

sub _conn_error {
  my ($self,$errstr,$id) = @_[OBJECT,ARG2,ARG3];
  return unless $self->{socket};
  delete $self->{socket};
  delete $self->{_server_info};
  $self->_send_event( $self->{_prefix} . 'disconnected' );
  return;
}

sub _conn_flushed {
  my ($self,$id) = @_[OBJECT,ARG0];
  return unless $self->{socket};
  if ( $self->{BUFFER} ) {
    my $item = shift @{ $self->{BUFFER} };
    unless ( $item ) {
      delete $self->{BUFFER};
      $self->_send_event( $self->{_prefix} . 'flushed' );
      return;
    }
    $self->{socket}->put($item);
    return;
  }
  unless ( $self->{_quit} ) {
    $self->_send_event( $self->{_prefix} . 'flushed' );
    return;
  }
  delete $self->{socket};
  delete $self->{_server_info};
  $self->_send_event( $self->{_prefix} . 'disconnected' );
  return;
}

sub _shutdown {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  delete $self->{factory};
  delete $self->{socket};
  delete $self->{_server_info};
  $kernel->alarm_remove_all();
  $kernel->alias_remove( $_ ) for $kernel->alias_list();
  $kernel->refcount_decrement( $self->{session_id} => __PACKAGE__ ) unless $self->{alias};
  $self->_unregister_sessions();
  return;
}

sub register {
  my ($kernel, $self, $session, $sender, @events) =
    @_[KERNEL, OBJECT, SESSION, SENDER, ARG0 .. $#_];

  unless (@events) {
    warn "register: Not enough arguments";
    return;
  }

  my $sender_id = $sender->ID();

  foreach (@events) {
    $_ = $self->{_prefix} . $_ unless /^_/;
    $self->{events}->{$_}->{$sender_id} = $sender_id;
    $self->{sessions}->{$sender_id}->{'ref'} = $sender_id;
    unless ($self->{sessions}->{$sender_id}->{refcnt}++ or $session == $sender) {
      $kernel->refcount_increment($sender_id, __PACKAGE__);
    }
  }

  $kernel->post( $sender, $self->{_prefix} . 'registered', $self );
  return;
}

sub unregister {
  my ($kernel, $self, $session, $sender, @events) =
    @_[KERNEL,  OBJECT, SESSION,  SENDER,  ARG0 .. $#_];

  unless (@events) {
    warn "unregister: Not enough arguments";
    return;
  }

  $self->_unregister($session,$sender,@events);
  undef;
}

sub _unregister {
  my ($self,$session,$sender) = splice @_,0,3;
  my $sender_id = $sender->ID();

  foreach (@_) {
    $_ = $self->{_prefix} . $_ unless /^_/;
    my $blah = delete $self->{events}->{$_}->{$sender_id};
    unless ( $blah ) {
	warn "$sender_id hasn't registered for '$_' events\n";
	next;
    }
    if (--$self->{sessions}->{$sender_id}->{refcnt} <= 0) {
      delete $self->{sessions}->{$sender_id};
      unless ($session == $sender) {
        $poe_kernel->refcount_decrement($sender_id, __PACKAGE__);
      }
    }
  }
  undef;
}

sub _unregister_sessions {
  my $self = shift;
  my $testd_id = $self->session_id();
  foreach my $session_id ( keys %{ $self->{sessions} } ) {
     if (--$self->{sessions}->{$session_id}->{refcnt} <= 0) {
        delete $self->{sessions}->{$session_id};
	$poe_kernel->refcount_decrement($session_id, __PACKAGE__)
		unless ( $session_id eq $testd_id );
     }
  }
}

sub __send_event {
  my( $self, $event, @args ) = @_[ OBJECT, ARG0, ARG1 .. $#_ ];
  $self->_send_event( $event, @args );
  return;
}

#sub send_event {
#  my $self = shift;
#  $poe_kernel->post( $self->{session_id}, '__send_event', @_ );
#}

sub _send_event  {
  my $self = shift;
  my ($event, @args) = @_;
  my $kernel = $POE::Kernel::poe_kernel;
  my %sessions;

  $sessions{$_} = $_ for (values %{$self->{events}->{$self->{_prefix} . 'all'}}, values %{$self->{events}->{$event}});

  $kernel->post( $_ => $event => @args ) for values %sessions;
  undef;
}

sub send_to_server {
  my $self = shift;
  $poe_kernel->call( $self->{session_id}, '_send_to_server', @_ );
}

sub _send_to_server {
  my ($kernel,$self,$output) = @_[KERNEL,OBJECT,ARG0];
  return unless $self->{socket};
  return unless $output;

  if ( ref $output eq 'ARRAY' ) {
    my $first = shift @{ $output };
    $self->{BUFFER} = $output if scalar @{ $output };
    $self->{socket}->put($first) if defined $first;
    return 1;
  }

  $self->{socket}->put($output);
  return 1;
}

q{Putting the test into POE};

__END__