File Coverage

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

Criterion	Covered	Total	%
statement	140	166	84.3
branch	75	104	72.1
condition	9	18	50.0
subroutine	20	27	74.0
pod	1	1	100.0
total	245	316	77.5

package POE::Component::Client::TCP;

use strict;

use vars qw($VERSION);
$VERSION = '1.365'; # NOTE - Should be #.### (three decimal places)

use Carp qw(carp croak);
use Errno qw(ETIMEDOUT ECONNRESET);

# Explicit use to import the parameter constants;
use POE::Session;
use POE::Driver::SysRW;
use POE::Filter::Line;
use POE::Wheel::ReadWrite;
use POE::Wheel::SocketFactory;

# Create the client.  This is just a handy way to encapsulate
# POE::Session->create().  Because the states are so small, it uses
# real inline coderefs.

sub new {
  my $type = shift;

  # Helper so we don't have to type it all day.  $mi is a name I call
  # myself.
  my $mi = $type . '->new()';

  # If they give us lemons, tell them to make their own damn
  # lemonade.
  croak "$mi requires an even number of parameters" if (@_ & 1);
  my %param = @_;

  # Validate what we're given.
  croak "$mi needs a RemoteAddress parameter"
    unless exists $param{RemoteAddress};
  croak "$mi needs a RemotePort parameter"
    unless exists $param{RemotePort};

  # Extract parameters.
  my $alias           = delete $param{Alias};
  my $address         = delete $param{RemoteAddress};
  my $port            = delete $param{RemotePort};
  my $domain          = delete $param{Domain};
  my $bind_address    = delete $param{BindAddress};
  my $bind_port       = delete $param{BindPort};
  my $ctimeout        = delete $param{ConnectTimeout};
  my $args            = delete $param{Args};
  my $session_type    = delete $param{SessionType};
  my $session_params  = delete $param{SessionParams};

  $args = [] unless defined $args;
  croak "Args must be an array reference" unless ref($args) eq "ARRAY";

  foreach (
    qw(
      PreConnect Connected ConnectError Disconnected ServerInput
      ServerError ServerFlushed Started
      ServerHigh ServerLow
    )
  ) {
    croak "$_ must be a coderef" if(
      defined($param{$_}) and ref($param{$_}) ne 'CODE'
    );
  }

  my $high_mark_level = delete $param{HighMark};
  my $low_mark_level  = delete $param{LowMark};
  my $high_event      = delete $param{ServerHigh};
  my $low_event       = delete $param{ServerLow};

  # this is ugly, but now its elegant :)  grep++
  my $using_watermarks = grep { defined $_ }
    ($high_mark_level, $low_mark_level, $high_event, $low_event);
  if ($using_watermarks > 0 and $using_watermarks != 4) {
    croak "If you use the Mark settings, you must define all four";
  }

  $high_event = sub { } unless defined $high_event;
  $low_event  = sub { } unless defined $low_event;

  my $pre_conn_callback   = delete $param{PreConnect};
  my $conn_callback       = delete $param{Connected};
  my $conn_error_callback = delete $param{ConnectError};
  my $disc_callback       = delete $param{Disconnected};
  my $input_callback      = delete $param{ServerInput};
  my $error_callback      = delete $param{ServerError};
  my $flush_callback      = delete $param{ServerFlushed};
  my $start_callback      = delete $param{Started};
  my $filter              = delete $param{Filter};
  # TODO should we have ServerInputFilter/ServerOutputFilter like Server-TCP does?

  # Extra states.

  my $inline_states = delete $param{InlineStates};
  $inline_states = {} unless defined $inline_states;

  my $package_states = delete $param{PackageStates};
  $package_states = [] unless defined $package_states;

  my $object_states = delete $param{ObjectStates};
  $object_states = [] unless defined $object_states;

  croak "InlineStates must be a hash reference"
    unless ref($inline_states) eq 'HASH';

  croak "PackageStates must be a list or array reference"
    unless ref($package_states) eq 'ARRAY';

  croak "ObjectStates must be a list or array reference"
    unless ref($object_states) eq 'ARRAY';

  # Errors.

  croak "$mi requires a ServerInput parameter" unless defined $input_callback;

  foreach (sort keys %param) {
    carp "$mi doesn't recognize \"$_\" as a parameter";
  }

  # Defaults.

  $session_type = 'POE::Session' unless defined $session_type;
  if (defined($session_params) && ref($session_params)) {
    if (ref($session_params) ne 'ARRAY') {
      croak "SessionParams must be an array reference";
    }
  } else {
    $session_params = [ ];
  }

  $address = '127.0.0.1' unless defined $address;

  $conn_error_callback = \&_default_error unless defined $conn_error_callback;
  $error_callback      = \&_default_io_error unless defined $error_callback;

  # Spawn the session that makes the connection and then interacts
  # with what was connected to.

  return $session_type->create
    ( @$session_params,
      inline_states =>
      { _start => sub {
          my ($kernel, $heap) = @_[KERNEL, HEAP];
          $heap->{shutdown_on_error} = 1;
          $kernel->alias_set( $alias ) if defined $alias;
          $kernel->yield( 'reconnect' );
          $start_callback and $start_callback->(@_);
        },

        # To quiet ASSERT_STATES.
        _stop   => sub { },
        _child  => sub { },

        reconnect => sub {
          my ($kernel, $heap) = @_[KERNEL, HEAP];

          $heap->{shutdown} = 0;
          $heap->{connected} = 0;

          # Tentative patch to re-establish the alias upon reconnect.
          # Necessary because otherwise the alias goes away for good.
          # Unfortunately, there is a gap where the alias may not be
          # set, and any events dispatched then will be dropped.
          $kernel->alias_set( $alias ) if defined $alias;

          $heap->{server} = POE::Wheel::SocketFactory->new
            ( RemoteAddress => $address,
              RemotePort    => $port,
              SocketDomain  => $domain,
              BindAddress   => $bind_address,
              BindPort      => $bind_port,
              SuccessEvent  => 'got_connect_success',
              FailureEvent  => 'got_connect_error',
            );
          $_[KERNEL]->alarm_remove( delete $heap->{ctimeout_id} )
            if exists $heap->{ctimeout_id};
          $heap->{ctimeout_id} = $_[KERNEL]->alarm_set
            ( got_connect_timeout => time + $ctimeout
            ) if defined $ctimeout;
        },

        connect => sub {
          my ($new_address, $new_port) = @_[ARG0, ARG1];
          $address = $new_address if defined $new_address;
          $port    = $new_port    if defined $new_port;
          $_[KERNEL]->yield("reconnect");
        },

        got_connect_success => sub {
          my ($kernel, $heap, $socket) = @_[KERNEL, HEAP, ARG0];

          $kernel->alarm_remove( delete $heap->{ctimeout_id} )
            if exists $heap->{ctimeout_id};

          # Pre-connected callback.
          if ($pre_conn_callback) {
            unless ($socket = $pre_conn_callback->(@_)) {
              $heap->{connected} = 0;
              # TODO - Error callback?  Disconnected callback?
              return;
            }
          }

          # Ok to overwrite like this as of 0.13.
          $_[HEAP]->{server} = POE::Wheel::ReadWrite->new
            ( Handle       => $socket,
              Driver       => POE::Driver::SysRW->new(),
              Filter       => _get_filter($filter),
              InputEvent   => 'got_server_input',
              ErrorEvent   => 'got_server_error',
              FlushedEvent => 'got_server_flush',
              (
                $using_watermarks
                ? (
                  HighMark  => $high_mark_level,
                  HighEvent => 'got_high',
                  LowMark   => $low_mark_level,
                  LowEvent  => 'got_low',
                )
                : ()
              )
            );

          $heap->{connected} = 1;
          $conn_callback and $conn_callback->(@_);
        },
        got_high => $high_event,
        got_low => $low_event,

        got_connect_error => sub {
          my $heap = $_[HEAP];
          $_[KERNEL]->alarm_remove( delete $heap->{ctimeout_id} )
            if exists $heap->{ctimeout_id};
          $heap->{connected} = 0;
          $conn_error_callback->(@_);
          delete $heap->{server};
        },

        got_connect_timeout => sub {
          my $heap = $_[HEAP];
          $heap->{connected} = 0;
          $_[KERNEL]->alarm_remove( delete $heap->{ctimeout_id} )
            if exists $heap->{ctimeout_id};
          $! = ETIMEDOUT;
          @_[ARG0,ARG1,ARG2] = ('connect', $!+0, $!);
          $conn_error_callback->(@_);
          delete $heap->{server};
        },

        got_server_error => sub {
          $error_callback->(@_);
          if ($_[HEAP]->{shutdown_on_error}) {
            $_[KERNEL]->yield("shutdown");
            $_[HEAP]->{got_an_error} = 1;
          }
        },

        got_server_input => sub {
          my $heap = $_[HEAP];
          return if $heap->{shutdown};
          $input_callback->(@_);
        },

        got_server_flush => sub {
          my $heap = $_[HEAP];
          $flush_callback and $flush_callback->(@_);
          if ($heap->{shutdown}) {
            delete $heap->{server};
            $disc_callback and $disc_callback->(@_);
          }
        },

        shutdown => sub {
          my ($kernel, $heap) = @_[KERNEL, HEAP];
          $heap->{shutdown} = 1;

          $kernel->alarm_remove( delete $heap->{ctimeout_id} )
            if exists $heap->{ctimeout_id};

          if ($heap->{connected}) {
            $heap->{connected} = 0;
            if (defined $heap->{server}) {
              if (
                $heap->{got_an_error} or
                not $heap->{server}->get_driver_out_octets()
              ) {
                delete $heap->{server};
                $disc_callback and $disc_callback->(@_);
              }
            }
          }
          else {
            delete $heap->{server};
          }

          $kernel->alias_remove($alias) if defined $alias;
        },

        # User supplied states.
        %$inline_states,
      },

      # User arguments.
      args => $args,

      # User supplied states.
      package_states => $package_states,
      object_states  => $object_states,
    )->ID;
}

sub _get_filter {
  my $filter = shift;
  if (ref $filter eq 'ARRAY') {
    my @filter_args = @$filter;
    $filter = shift @filter_args;
    return $filter->new(@filter_args);
  } elsif (ref $filter) {
    return $filter->clone();
  } elsif (!defined($filter)) {
    return POE::Filter::Line->new();
  } else {
    return $filter->new();
  }
}

# The default error handler logs to STDERR and shuts down the socket.

sub _default_error {
  unless ($_[ARG0] eq "read" and ($_[ARG1] == 0 or $_[ARG1] == ECONNRESET)) {
    warn(
      'Client ', $_[SESSION]->ID, " got $_[ARG0] error $_[ARG1] ($_[ARG2])\n"
    );
  }
  delete $_[HEAP]->{server};
}

sub _default_io_error {
  my ($syscall, $errno, $error) = @_[ARG0..ARG2];
  $error = "Normal disconnection" unless $errno;
  warn('Client ', $_[SESSION]->ID, " got $syscall error $errno ($error)\n");
  $_[KERNEL]->yield("shutdown");
}

1;

__END__