File Coverage

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

Criterion	Covered	Total	%
statement	206	263	78.3
branch	63	116	54.3
condition	6	16	37.5
subroutine	37	46	80.4
pod	18	18	100.0
total	330	459	71.9

package Test::POE::Server::TCP;
$Test::POE::Server::TCP::VERSION = '1.20';
# ABSTRACT: A POE Component providing TCP server services for test cases

use strict;
use warnings;
use POE qw(Wheel::SocketFactory Wheel::ReadWrite Filter::Line);
use Socket;
use Carp qw(carp croak);

our $GOT_SOCKET6;

BEGIN {
    eval {
        Socket->import(qw(AF_INET6 IN6ADDR_ANY NI_NUMERICHOST NI_NUMERICSERV getnameinfo));
        $GOT_SOCKET6 = 1;
    };
    if (!$GOT_SOCKET6) {
       # provide a dummy subs so code compiles
       *AF_INET6 = sub { ~0 };
       *IN6ADDR_ANY = sub { ~0 };
    }
}

sub spawn {
  my $package = shift;
  my %opts = @_;
  $opts{lc $_} = delete $opts{$_} for keys %opts;
  my $options = delete $opts{options};
  my $self = bless \%opts, $package;
  $self->{_prefix} = delete $self->{prefix};
  $self->{_prefix} = 'testd_' 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_client      => '_send_to_client',
		      send_to_all_clients => '_send_to_all_clients',
		      disconnect          => '_disconnect',
		      terminate           => '_terminate',
          start_listener      => '_start_listener',
	            },
	   $self => [ qw(_start register unregister _accept_client _conn_input _conn_error _conn_flushed _conn_alarm
                _send_to_client __send_event _disconnect _send_to_all_clients _accept_failed4 _accept_failed6) ],
	],
	heap => $self,
	( ref($options) eq 'HASH' ? ( options => $options ) : () ),
  )->ID();
  return $self;
}

sub session_id {
  return $_[0]->{session_id};
}

sub pause_listening {
  $_[0]->{listener}->pause_accept() if $_[0]->{listener};
  $_[0]->{listener6}->pause_accept() if $_[0]->{listener6};
}

sub resume_listening {
  $_[0]->{listener}->resume_accept() if $_[0]->{listener};
  $_[0]->{listener6}->resume_accept() if $_[0]->{listener6};
}

sub getsockname {
  return unless $_[0]->{listener};
  return $_[0]->{listener}->getsockname();
}

sub port {
  my $self = shift;
  return ( sockaddr_in( $self->getsockname() ) )[0];
}

sub getsockname6 {
  return unless $_[0]->{listener6};
  return $_[0]->{listener6}->getsockname();
}

sub port6 {
  my $self = shift;
  return ( sockaddr_in6( $self->getsockname6() ) )[0];
}

sub _conn_exists {
  my ($self,$wheel_id) = @_;
  return 0 unless $wheel_id and defined $self->{clients}->{ $wheel_id };
  return 1;
}

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

sub _start {
  my ($kernel,$self,$sender) = @_[KERNEL,OBJECT,SENDER];
  $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->call( $self->{session_id}, 'start_listener' );
  return;
}

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

sub _start_listener {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  return if $self->{listener};

  $self->{listener} = POE::Wheel::SocketFactory->new(
      ( defined $self->{address} ? ( BindAddress => $self->{address} ) : () ),
      ( defined $self->{port} ? ( BindPort => $self->{port} ) : ( BindPort => 0 ) ),
      SuccessEvent   => '_accept_client',
      FailureEvent   => '_accept_failed4',
      SocketDomain   => AF_INET,             # Sets the socket() domain
      SocketType     => SOCK_STREAM,         # Sets the socket() type
      SocketProtocol => 'tcp',               # Sets the socket() protocol
      Reuse          => 'on',                # Lets the port be reused
  );

  return unless $GOT_SOCKET6;

  $self->{listener6} = POE::Wheel::SocketFactory->new(
      #BindAddress => IN6ADDR_ANY,
      ( defined $self->{port} ? ( BindPort => $self->{port} ) : ( BindPort => 0 ) ),
      SuccessEvent   => '_accept_client',
      FailureEvent   => '_accept_failed6',
      SocketDomain   => AF_INET6,            # Sets the socket() domain
      SocketType     => SOCK_STREAM,         # Sets the socket() type
      SocketProtocol => 'tcp',               # Sets the socket() protocol
      Reuse          => 'on',                # Lets the port be reused
  );

  return;
}

sub _accept_client {
  my ($kernel,$self,$socket,$listener_id) = @_[KERNEL,OBJECT,ARG0,ARG3];

  my (undef,$peeraddr,$peerport) = getnameinfo( CORE::getpeername( $socket ), NI_NUMERICHOST | NI_NUMERICSERV );
  my (undef,$sockaddr,$sockport) = getnameinfo( CORE::getsockname( $socket ), NI_NUMERICHOST | NI_NUMERICSERV );

  s!^::ffff:!! for ( $sockaddr, $peeraddr );

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

  return unless $wheel;

  my $id = $wheel->ID();
  $self->{clients}->{ $id } =
  {
				wheel    => $wheel,
				peeraddr => $peeraddr,
				peerport => $peerport,
				sockaddr => $sockaddr,
				sockport => $sockport,
  };
  $self->_send_event( $self->{_prefix} . 'connected', $id, $peeraddr, $peerport, $sockaddr, $sockport );

  #$self->{clients}->{ $id }->{alarm} = $kernel->delay_set( '_conn_alarm', $self->{time_out} || 300, $id );
  return;
}

sub client_info {
  my $self = shift;
  my $id = shift || return;
  return unless $self->_conn_exists( $id );
  my %hash = %{ $self->{clients}->{ $id } };
  delete $hash{wheel};
  return map { $hash{$_} } qw(peeraddr peerport sockaddr sockport) if wantarray;
  return \%hash;
}

sub client_wheel {
  my $self = shift;
  my $id = shift || return;
  return unless $self->_conn_exists( $id );
  return $self->{clients}->{ $id }->{wheel};
}

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 _accept_failed4 {
  my ($kernel,$self,$operation,$errnum,$errstr,$wheel_id) = @_[KERNEL,OBJECT,ARG0..ARG3];
  warn "Wheel $wheel_id generated $operation error $errnum: $errstr\n";
  delete $self->{listener} if $operation eq 'listen';
  $self->_send_event( $self->{_prefix} . 'listener_failed', $operation, $errnum, $errstr );
  return;
}

sub _accept_failed6 {
  my ($kernel,$self,$operation,$errnum,$errstr,$wheel_id) = @_[KERNEL,OBJECT,ARG0..ARG3];
  warn "Wheel $wheel_id generated $operation error $errnum: $errstr\n";
  delete $self->{listener6} if $operation eq 'listen';
  $self->_send_event( $self->{_prefix} . 'listener_failed', $operation, $errnum, $errstr );
  return;
}

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

sub _disconnect {
  my ($kernel,$self,$id) = @_[KERNEL,OBJECT,ARG0];
  return unless $self->_conn_exists( $id );
  $self->{clients}->{ $id }->{quit} = 1;
  return 1;
}

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

sub _terminate {
  my ($kernel,$self,$id) = @_[KERNEL,OBJECT,ARG0];
  return unless $self->_conn_exists( $id );
  delete $self->{clients}->{ $id };
  $self->_send_event( $self->{_prefix} . 'disconnected', $id );
  return 1;
}

sub _conn_input {
  my ($kernel,$self,$input,$id) = @_[KERNEL,OBJECT,ARG0,ARG1];
  return unless $self->_conn_exists( $id );
  #$kernel->delay_adjust( $self->{clients}->{ $id }->{alarm}, $self->{time_out} || 300 );
  $self->_send_event( $self->{_prefix} . 'client_input', $id, $input );
  return;
}

sub _conn_error {
  my ($self,$errstr,$id) = @_[OBJECT,ARG2,ARG3];
  return unless $self->_conn_exists( $id );
  my $href = delete $self->{clients}->{ $id };
  delete $href->{wheel};
  $self->_send_event( $self->{_prefix} . 'disconnected', $id,  map { $href->{$_} } qw(peeraddr peerport sockaddr sockport) );
  return;
}

sub _conn_flushed {
  my ($self,$id) = @_[OBJECT,ARG0];
  return unless $self->_conn_exists( $id );
  if ( $self->{clients}->{ $id }->{BUFFER} ) {
    my $item = shift @{ $self->{clients}->{ $id }->{BUFFER} };
    unless ( $item ) {
      delete $self->{clients}->{ $id }->{BUFFER};
      $self->_send_event( $self->{_prefix} . 'client_flushed', $id );
      return;
    }
    $self->{clients}->{ $id }->{wheel}->put($item);
    return;
  }
  unless ( $self->{clients}->{ $id }->{quit} ) {
    $self->_send_event( $self->{_prefix} . 'client_flushed', $id );
    return;
  }
  delete $self->{clients}->{ $id };
  $self->_send_event( $self->{_prefix} . 'disconnected', $id );
  return;
}

sub _conn_alarm {
  my ($kernel,$self,$id) = @_[KERNEL,OBJECT,ARG0];
  return unless $self->_conn_exists( $id );
  delete $self->{clients}->{ $id };
  $self->_send_event( $self->{_prefix} . 'disconnected', $id );
  return;
}

sub _shutdown {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  delete $self->{listener};
  delete $self->{listener6};
  delete $self->{clients};
  $kernel->alarm_remove_all();
  $kernel->alias_remove( $_ ) for $kernel->alias_list();
  $kernel->refcount_decrement( $self->{session_id} => __PACKAGE__ ) unless $self->{alias};
#  $self->_pluggable_destroy();
  $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;
  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_client {
  my $self = shift;
  $poe_kernel->call( $self->{session_id}, '_send_to_client', @_ );
}

sub _send_to_client {
  my ($kernel,$self,$id,$output) = @_[KERNEL,OBJECT,ARG0..ARG1];
  return unless $self->_conn_exists( $id );
  return unless defined $output;

  if ( ref $output eq 'ARRAY' ) {
    my $temp = [ @{ $output } ];
    my $first = shift @{ $temp };
    $self->{clients}->{ $id }->{BUFFER} = $temp if scalar @{ $temp };
    $self->{clients}->{ $id }->{wheel}->put($first) if defined $first;
    return 1;
  }

  $self->{clients}->{ $id }->{wheel}->put($output);
  return 1;
}

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

sub _send_to_all_clients {
  my ($kernel,$self,$output) = @_[KERNEL,OBJECT,ARG0];
  return unless defined $output;
  $self->send_to_client( $_, $output ) for
    keys %{ $self->{clients} };
  return 1;
}

q{Putting the test into POE};

__END__