File Coverage

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

Criterion	Covered	Total	%
statement	230	273	84.2
branch	96	154	62.3
condition	32	72	44.4
subroutine	24	28	85.7
pod	1	1	100.0
total	383	528	72.5

package POE::Component::Server::TCP;

use strict;

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

use Carp qw(carp croak);
use Socket qw(INADDR_ANY inet_ntoa inet_aton AF_INET AF_UNIX PF_UNIX);
use Errno qw(ECONNABORTED ECONNRESET);

BEGIN {
  # under perl-5.6.2 the warning "leaks" from the eval, while newer versions don't...
  # it's due to Exporter.pm behaving differently, so we have to shut it up
  no warnings 'redefine';
  local *Carp::carp = sub { die @_ };

  # Socket::GetAddrInfo provides getaddrinfo where earlier Perls' Socket don't.
  eval { Socket->import('getaddrinfo') };
  if ($@) {
    # :newapi is legacy, but we include it to be sure in case the user has an old version of GAI
    eval { require Socket::GetAddrInfo; Socket::GetAddrInfo->import( qw(:newapi getaddrinfo) ) };
    if ($@) {
      *getaddrinfo = sub { Carp::confess("Unable to use IPv6: Socket::GetAddrInfo not available") };
    }
  }
}

# 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;

sub DEBUG () { 0 }

# Create the server.  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 = @_;

  # Extract parameters.
  my $alias   = delete $param{Alias};
  my $address = delete $param{Address};
  my $hname   = delete $param{Hostname};
  my $port    = delete $param{Port};
  my $domain  = delete($param{Domain}) || AF_INET;
  my $concurrency = delete $param{Concurrency};

  $port = 0 unless defined $port;

  foreach (
    qw(
      Acceptor Error ClientInput
      ClientPreConnect ClientConnected ClientDisconnected
      ClientError ClientFlushed
      ClientLow ClientHigh
    )
  ) {
    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{ClientHigh};
  my $low_event       = delete $param{ClientLow};

  my $mark_param_count = (
    grep { defined $_ }
    ($high_mark_level, $low_mark_level, $high_event, $low_event)
  );
  if ($mark_param_count and $mark_param_count < 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 $accept_callback = delete $param{Acceptor};
  my $error_callback  = delete $param{Error};

  my $client_input    = delete $param{ClientInput};

  # Acceptor and ClientInput are mutually exclusive.
  croak "$mi needs either an Acceptor or a ClientInput but not both"
    unless defined($accept_callback) xor defined($client_input);

  # Make sure ClientXyz are accompanied by ClientInput.
  unless (defined($client_input)) {
    foreach (grep /^Client/, keys %param) {
      croak "$_ not permitted without ClientInput";
    }
  }

  my $client_pre_connect  = delete $param{ClientPreConnect};
  my $client_connected    = delete $param{ClientConnected};
  my $client_disconnected = delete $param{ClientDisconnected};
  my $client_error        = delete $param{ClientError};
  my $client_filter       = delete $param{ClientFilter};
  my $client_infilter     = delete $param{ClientInputFilter};
  my $client_outfilter    = delete $param{ClientOutputFilter};
  my $client_flushed      = delete $param{ClientFlushed};
  my $session_type        = delete $param{SessionType};
  my $session_params      = delete $param{SessionParams};
  my $server_started      = delete $param{Started};
  my $server_stopped      = delete $param{Stopped};
  my $listener_args       = delete $param{ListenerArgs};

  $listener_args = [] unless defined $listener_args;
  croak "ListenerArgs must be an array reference"
    unless ref($listener_args) eq 'ARRAY';

  if (exists $param{Args}) {
    if (exists $param{ClientArgs}) {
      carp "Args is deprecated, and ignored since ClientArgs is present";
      delete $param{Args};
    }
    else {
      carp "Args is deprecated but allowed for now.  Please use ClientArgs";
    }
  }

  my $client_args = delete($param{ClientArgs}) || delete($param{Args});

  if ( (defined $client_infilter and ! defined $client_outfilter) or
    (defined $client_outfilter and ! defined $client_infilter) ) {
    croak "ClientInputFilter must be used with ClientOutputFilter";
  }

  if (defined $client_filter and defined $client_infilter) {
    carp "ClientFilter ignored with ClientInputFilter and ClientOutputFilter";
    undef $client_filter;
  }

  # Defaults.

  $concurrency = -1 unless defined $concurrency;
  my $accept_session_id;

  if (!defined $address && defined $hname) {
    $address = inet_aton($hname);
  }
  $address = INADDR_ANY unless defined $address;

  $error_callback = \&_default_server_error unless defined $error_callback;

  $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 = [ ];
  }

  if (defined $client_input) {
    $client_error  = \&_default_client_error unless defined $client_error;
    $client_args         = []     unless defined $client_args;

    # 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;

    my $shutdown_on_error = 1;
    if (exists $param{ClientShutdownOnError}) {
      $shutdown_on_error = delete $param{ClientShutdownOnError};
    }

    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 "ObjectsStates must be a list or array reference"
      unless ref($object_states) eq 'ARRAY';

    croak "ClientArgs must be an array reference"
      unless ref($client_args) eq 'ARRAY';

    # Sanity check, thanks to crab@irc for making this mistake, ha!
    # TODO we could move this to POE::Session and make it a
    # "sanity checking" sub somehow...
    if (POE::Kernel::ASSERT_USAGE) {
      my %forbidden_handlers = (
        _child => 1,
        _start => 1,
        _stop => 1,
        shutdown => 1,
        tcp_server_got_error => 1,
        tcp_server_got_flush => 1,
        tcp_server_got_high => 1,
        tcp_server_got_input => 1,
        tcp_server_got_low => 1,
      );

      if (
        my @forbidden_inline_handlers = (
          grep { exists $inline_states->{$_} }
          keys %forbidden_handlers
        )
      ) {
        croak "These InlineStates aren't allowed: @forbidden_inline_handlers";
      }

      my %handlers = (
        PackageStates => $package_states,
        ObjectStates => $object_states,
      );

      while (my ($name, $states) = each(%handlers)) {
        my %states_hash = @$states;
        my @forbidden_handlers;
        while (my ($package, $handlers) = each %states_hash) {
          croak "Undefined $name member for $package" unless (
            defined $handlers
          );

          if (ref($handlers) eq 'HASH') {
            push(
              @forbidden_handlers,
              grep { exists $handlers->{$_} }
              keys %forbidden_handlers
            );
          }
          elsif (ref($handlers) eq 'ARRAY') {
            push(
              @forbidden_handlers,
              grep { exists $forbidden_handlers{$_} }
              @$handlers
            );
          }
          else {
            croak "Unknown $name member type for $package";
          }
        }

        croak "These $name aren't allowed: @forbidden_handlers" if (
          @forbidden_handlers
        );
      }
    }

    # Revise the acceptor callback so it spawns a session.

    unless (defined $accept_callback) {
      $accept_callback = sub {
        my ($socket, $remote_addr, $remote_port) = @_[ARG0, ARG1, ARG2];

        $session_type->create(
          @$session_params,
          inline_states => {
            _start => sub {
              my ( $kernel, $session, $heap ) = @_[KERNEL, SESSION, HEAP];

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

              # Unofficial UNIX support, suggested by Damir Dzeko.
              # Real UNIX socket support should go into a separate
              # module, but if that module only differs by four
              # lines of code it would be bad to maintain two
              # modules for the price of one.  One solution would be
              # to pull most of this into a base class and derive
              # TCP and UNIX versions from that.
              if (
                $domain == AF_UNIX or $domain == PF_UNIX
              ) {
                $heap->{remote_ip} = "LOCAL";
              }
              elsif (length($remote_addr) == 4) {
                $heap->{remote_ip} = inet_ntoa($remote_addr);
              }
              else {
                $heap->{remote_ip} = ( getaddrinfo($remote_addr) )[1];
              }

              $heap->{remote_port} = $remote_port;

              my $socket = $_[ARG0];
              if ($client_pre_connect) {
                $socket = $client_pre_connect->(@_);
                unless (defined($socket) and ref($socket) and fileno($socket)) {
                  # TODO - The user ought to know what's going on
                  # here, since it's triggered by something their
                  # callback has done.  Should we expose a callback
                  # anyway to avoid potential confusion?
                  return;
                }
              }

              $heap->{client} = POE::Wheel::ReadWrite->new(
                Handle       => $socket,
                Driver       => POE::Driver::SysRW->new(),
                _get_filters(
                  $client_filter,
                  $client_infilter,
                  $client_outfilter
                ),
                InputEvent   => 'tcp_server_got_input',
                ErrorEvent   => 'tcp_server_got_error',
                FlushedEvent => 'tcp_server_got_flush',

                (
                  $mark_param_count
                  ? (
                    HighMark  => $high_mark_level,
                    HighEvent => 'tcp_server_got_high',
                    LowMark   => $low_mark_level,
                    LowEvent  => 'tcp_server_got_low',
                  )
                  : ()
                ),
              );

              # Expand the Args constructor array, and place a copy
              # into @_[ARG0..].  There are only 2 parameters.
              splice(@_, ARG0, 2, @{$_[ARG1]});

              $client_connected and $client_connected->(@_);
            },
            tcp_server_got_high => $high_event,
            tcp_server_got_low => $low_event,

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

            tcp_server_got_input => sub {
              return if $_[HEAP]->{shutdown};
              $client_input->(@_);
              undef;
            },
            tcp_server_got_error => sub {
              DEBUG and warn(
                "$$: $alias child Error ARG0=$_[ARG0] ARG1=$_[ARG1]"
              );
              unless ($_[ARG0] eq 'accept' and $_[ARG1] == ECONNABORTED) {
                $client_error->(@_);
                if ($_[HEAP]->{shutdown_on_error}) {
                  $_[HEAP]->{got_an_error} = 1;
                  $_[KERNEL]->yield("shutdown");
                }
              }
            },
            tcp_server_got_flush => sub {
              my $heap = $_[HEAP];
              DEBUG and warn "$$: $alias child Flush";
              $client_flushed and $client_flushed->(@_);
              if ($heap->{shutdown}) {
                DEBUG and warn "$$: $alias child Flush, callback";
                $client_disconnected and $client_disconnected->(@_);
                delete $heap->{client};
              }
            },
            shutdown => sub {
              DEBUG and warn "$$: $alias child Shutdown";
              my $heap = $_[HEAP];
              $heap->{shutdown} = 1;
              if (defined $heap->{client}) {
                if (
                  $heap->{got_an_error} or
                  not $heap->{client}->get_driver_out_octets()
                ) {
                  DEBUG and warn "$$: $alias child Shutdown, callback";
                  $client_disconnected and $client_disconnected->(@_);
                  delete $heap->{client};
                }
              }
            },
            _stop => sub {
              ## concurrency on close
              DEBUG and warn(
                "$$: $alias _stop accept_session = $accept_session_id"
              );
              if( defined $accept_session_id ) {
                $_[KERNEL]->call( $accept_session_id, 'disconnected' );
              }
              else {
                # This means that the Server::TCP was shutdown before
                # this connection closed.  So it doesn't really matter that
                # we can't decrement the connection counter.
                DEBUG and warn(
                  "$$: $_[HEAP]->{alias} Disconnected from a connection ",
                  "without POE::Component::Server::TCP parent"
                );
              }
              return;
            },

            # User supplied states.
            %$inline_states
          },

          # More user supplied states.
          package_states => $package_states,
          object_states  => $object_states,

          # XXX - If you change the number of args here, also change
          # the splice elsewhere.
          args => [ $socket, $client_args ],
        );
      };
    }
  };

  # Complain about strange things we're given.
  foreach (sort keys %param) {
    carp "$mi doesn't recognize \"$_\" as a parameter";
  }

  ## verify concurrency on accept
  my $orig_accept_callback = $accept_callback;
  $accept_callback = sub {
    $_[HEAP]->{connections}++;
    DEBUG and warn(
      "$$: $_[HEAP]->{alias} Connection opened ",
      "($_[HEAP]->{connections} open)"
    );
    if( $_[HEAP]->{concurrency} != -1 and $_[HEAP]->{listener} ) {
      if( $_[HEAP]->{connections} >= $_[HEAP]->{concurrency} ) {
        DEBUG and warn(
          "$$: $_[HEAP]->{alias} Concurrent connection limit reached, ",
          "pausing accept"
        );
        $_[HEAP]->{listener}->pause_accept()
      }
    }
    $orig_accept_callback->(@_);
  };

  # Create the session, at long last.
  # This is done inline so that closures can customize it.
  # We save the accept session's ID to avoid self reference.

  $accept_session_id = $session_type->create(
    @$session_params,
    inline_states => {
      _start => sub {
        if (defined $alias) {
          $_[HEAP]->{alias} = $alias;
          $_[KERNEL]->alias_set( $alias );
        }

        $_[HEAP]->{concurrency} = $concurrency;
        $_[HEAP]->{connections} = 0;

        $_[HEAP]->{listener} = POE::Wheel::SocketFactory->new(
          ( ($domain == AF_UNIX or $domain == PF_UNIX)
            ? ()
            : ( BindPort => $port )
          ),
          BindAddress  => $address,
          SocketDomain => $domain,
          Reuse        => 'yes',
          SuccessEvent => 'tcp_server_got_connection',
          FailureEvent => 'tcp_server_got_error',
        );
        $server_started and $server_started->(@_);
      },
      # Catch an error.
      tcp_server_got_error => $error_callback,

      # We accepted a connection.  Do something with it.
      tcp_server_got_connection => $accept_callback,

      # concurrency on close.
      disconnected => sub {
        $_[HEAP]->{connections}--;
        DEBUG and warn(
          "$$: $_[HEAP]->{alias} Connection closed ",
          "($_[HEAP]->{connections} open)"
        );
        if ($_[HEAP]->{connections} < 0) {
          warn(
            "Excessive 'disconnected' event ",
            "from $_[CALLER_FILE] at line $_[CALLER_LINE]\n"
          );
          $_[HEAP]->{connections} = 0;
        }
        if( $_[HEAP]->{concurrency} != -1 and $_[HEAP]->{listener} ) {
          if( $_[HEAP]->{connections} == ($_[HEAP]->{concurrency}-1) ) {
            DEBUG and warn(
              "$$: $_[HEAP]->{alias} Concurrent connection limit ",
              "reestablished, resuming accept"
            );
            $_[HEAP]->{listener}->resume_accept();
          }
        }
      },

      set_concurrency => sub {
        $_[HEAP]->{concurrency} = $_[ARG0];
        DEBUG and warn(
          "$$: $_[HEAP]->{alias} Concurrent connection ",
          "limit = $_[HEAP]->{concurrency}"
        );
        if( $_[HEAP]->{concurrency} != -1 and $_[HEAP]->{listener} ) {
          if( $_[HEAP]->{connections} >= $_[HEAP]->{concurrency} ) {
            DEBUG and warn(
              "$$: $_[HEAP]->{alias} Concurrent connection limit ",
              "reached, pausing accept"
            );
            $_[HEAP]->{listener}->pause_accept()
          }
          else {
            DEBUG and warn(
              "$$: $_[HEAP]->{alias} Concurrent connection limit ",
              "reestablished, resuming accept"
            );
            $_[HEAP]->{listener}->resume_accept();
          }
        }
      },

      # Shut down.
      shutdown => sub {
        delete $_[HEAP]->{listener};
        $_[KERNEL]->alias_remove( $_[HEAP]->{alias} )
          if defined $_[HEAP]->{alias};
      },

      # Dummy states to prevent warnings.
      _stop   => sub {
        DEBUG and warn "$$: $_[HEAP]->{alias} _stop";
        $server_stopped and $server_stopped->(@_);
        undef($accept_session_id);
        return 0;
      },
      _child  => sub { },
    },

    args => $listener_args,
  )->ID;

  # Return the session ID.
  return $accept_session_id;
}

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)
      );
    }
    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 success 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 default POE::Filter::Line "
        );
        return 0;
    }
    return 1;
}

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

sub _default_server_error {
  warn("$$: ".
    'Server ', $_[SESSION]->ID,
    " got $_[ARG0] error $_[ARG1] ($_[ARG2])\n"
  );
  delete $_[HEAP]->{listener};
}

# The default client error handler logs to STDERR

sub _default_client_error {
  my ($syscall, $errno, $error) = @_[ARG0..ARG2];
  unless ($syscall eq "read" and ($errno == 0 or $errno == ECONNRESET)) {
    $error = "(no error)" unless $errno;
    warn("$$: ".
      'Client session ', $_[SESSION]->ID,
      " got $syscall error $errno ($error)\n"
    );
  }
}

1;

__END__