File Coverage

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

Criterion	Covered	Total	%
statement	107	158	67.7
branch	25	58	43.1
condition	5	20	25.0
subroutine	20	27	74.0
pod	5	5	100.0
total	162	268	60.4

package POE::Component::Server::Inet;
$POE::Component::Server::Inet::VERSION = '0.06';
#ABSTRACT: a super-server daemon implementation in POE

use strict;
use warnings;
use POE qw(Wheel::SocketFactory Wheel::Run Wheel::ReadWrite Filter::Stream);
use Net::Netmask;
use Socket;
use Carp;

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->{session_id} = POE::Session->create(
	object_states => [
	    $self => { shutdown       => '_shutdown',
		       add_tcp	      => '_add_tcp',
		       del_tcp	      => '_del_tcp',
#		       add_udp	      => '_add_udp',
#		       del_udp	      => '_del_udp',
	    },
	    $self => [ qw(_start _accept_new_client _accept_failed _get_datagram _sig_child _client_input _client_flushed _client_error _wheel_out _wheel_close _wheel_error _wheel_alarm) ],
	],
	heap => $self,
	( ref($options) eq 'HASH' ? ( options => $options ) : () ),
  )->ID();
  return $self;
}

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

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

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__ );
  }
  $self->{stream_filter} = POE::Filter::Stream->new();
  return;
}

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

sub _shutdown {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  $kernel->alarm_remove_all();
  $kernel->alias_remove( $_ ) for $kernel->alias_list();
  $kernel->refcount_decrement( $self->{session_id} => __PACKAGE__ ) unless $self->{alias};
  # Shutdown TCP listeners
  delete $self->{tcp_ports};
  # Shutdown UDP listeners
  $kernel->select_read( $_->{socket} ) for values %{ $self->{udp_ports} };
  # Shutdown wheels.
  delete $self->{clients};
  delete $self->{wheels};
  return;
}

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

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


sub _add_tcp {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  my $args;
  if ( ref( $_[ARG0] ) eq 'HASH' ) {
    $args = { %{ $_[ARG0] } };
  }
  else {
    $args = { @_[ARG0..$#_] };
  }
  unless ( defined $args->{port} ) {
    warn "You must specify a 'port' parameter\n";
    return;
  }
  if ( grep { $_->{port} eq $args->{port} } values %{ $self->{tcp_ports} } ) {
    warn "There already exists a TCP port definition for '$args->{port}'\n";
    return;
  }
  unless ( $args->{program} ) {
    warn "You must specify a 'program' parameter\n";
    return;
  }
  delete $args->{programargs} unless $args->{programargs} and ref $args->{programargs} eq 'ARRAY';
  if ( $args->{allow} and !$args->{allow}->isa('Net::Netmask') ) {
    warn "'allow' parameter must be a Net::Netmask object, ignoring.\n";
    delete $args->{allow};
  }
  if ( $args->{deny} and !$args->{deny}->isa('Net::Netmask') ) {
    warn "'deny' parameter must be a Net::Netmask object, ignoring.\n";
    delete $args->{deny};
  }
  my $sockfactory = POE::Wheel::SocketFactory->new(
    ( defined $args->{bindaddress} ? ( BindAddress => $args->{bindaddress} ) : () ),
    BindPort => $args->{port},
    SuccessEvent   => '_accept_new_client',
    FailureEvent   => '_accept_failed',
    SocketDomain   => AF_INET,
    SocketType     => SOCK_STREAM,
    SocketProtocol => 'tcp',
    Reuse          => 'on',
  );
  $args->{sockfactory} = $sockfactory;
  $self->{tcp_ports}->{ $sockfactory->ID() } = $args;
  my $port = ( sockaddr_in( $sockfactory->getsockname() ) )[0];
  $args->{port} = $port;
  return $port;
}

sub _del_tcp {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  my $args;
  if ( ref( $_[ARG0] ) eq 'HASH' ) {
    $args = { %{ $_[ARG0] } };
  }
  else {
    $args = { @_[ARG0..$#_] };
  }
  unless ( defined $args->{port} ) {
    warn "You must specify a 'port' parameter\n";
    return;
  }
  foreach my $sockfactory_id ( keys %{ $self->{tcp_ports} } ) {
    next unless $self->{tcp_ports}->{ $sockfactory_id }->{port} eq $args->{port};
    delete $self->{tcp_ports}->{ $sockfactory_id };
    return;
  }
  return;
}


sub _accept_failed {
  my ($kernel,$self,$wheel_id) = @_[KERNEL,OBJECT,ARG3];
  warn join(' ', @_[ARG0..ARG2] ), "\n";
  delete $self->{tcp_ports}->{ $wheel_id }->{sockfactory};
  return;
}

sub _accept_new_client {
  my ($kernel,$self,$socket,$peeraddr,$peerport,$factory_id) = @_[KERNEL,OBJECT,ARG0 .. ARG3];
  $peeraddr = inet_ntoa($peeraddr);
  # Check if 'accept' or 'deny'
  my $client = POE::Wheel::ReadWrite->new (
        Handle => $socket,
        Filter => $self->{stream_filter},
        InputEvent => '_client_input',
        ErrorEvent => '_client_error',
	FlushedEvent => '_client_flushed',
  );
  my $args = $self->{tcp_ports}->{ $factory_id };
  my $wheel = POE::Wheel::Run->new(
     Program => $args->{program},
     ProgramArgs => $args->{programargs},
     StdioFilter => $self->{stream_filter},
     StderrFilter => $self->{stream_filter},
     StdoutEvent => '_wheel_out',    # Received data from the child's STDOUT.
     StderrEvent => '_wheel_out',    # Received data from the child's STDERR.
     ErrorEvent  => '_wheel_error',          # An I/O error occurred.
     CloseEvent  => '_wheel_close',  # Child closed all output handles.
     ( defined $args->{user} ? ( User => $args->{user} ) : () ),
     ( defined $args->{group} ? ( Group => $args->{group} ) : () ),
  );
  my $client_id = $client->ID();
  my $wheel_id = $wheel->ID();
  $self->{wheels}->{ $wheel_id } = { wheel => $wheel, client => $client_id, tcp => 1 };
  $self->{clients}->{ $client_id } = { wheel => $wheel_id, client => $client };
  $kernel->sig_child( $wheel->PID(), '_sig_child' );
  return;
}

sub _sig_child {
  $poe_kernel->sig_handled();
}

sub _client_input {
  my ($kernel,$self,$data,$client_id) = @_[KERNEL,OBJECT,ARG0,ARG1];
  my $wheel_id = $self->{clients}->{ $client_id }->{wheel};
  return unless $self->{wheels}->{ $wheel_id };
  $self->{wheels}->{ $wheel_id }->{wheel}->put( $data );
  return;
}

sub _client_error {
  my ($kernel,$self,$client_id) = @_[KERNEL,OBJECT,ARG3];
  my $wheel_id = $self->{clients}->{ $client_id }->{wheel};
  delete $self->{clients}->{ $client_id };
  return unless $self->{wheels}->{ $wheel_id };
  $self->{wheels}->{ $wheel_id }->{wheel}->shutdown_stdin();
  $self->{wheels}->{ $wheel_id }->{alarm} =
	$kernel->delay_set( '_wheel_alarm', $self->{timeout} || 30, $wheel_id );
  return;
}

sub _client_flushed {
  my ($kernel,$self,$client_id) = @_[KERNEL,OBJECT,ARG0];
  $self->{clients}->{ $client_id }->{pending} = 0;
  return unless $self->{clients}->{ $client_id }->{shutdown};
  delete $self->{clients}->{ $client_id };
  return;
}

sub _wheel_out {
  my ($kernel,$self,$data,$wheel_id) = @_[KERNEL,OBJECT,ARG0,ARG1];
  if ( defined $self->{wheels}->{ $wheel_id }->{tcp} ) {
    my $client_id = $self->{wheels}->{ $wheel_id }->{client};
    return unless $self->{clients}->{ $client_id };
    $self->{clients}->{ $client_id }->{client}->put( $data );
    $self->{clients}->{ $client_id }->{pending} = 1;
  }
  return;
}

sub _wheel_alarm {
  my ($kernel,$self,$wheel_id) = @_[KERNEL,OBJECT,ARG0];
  return unless $self->{wheels}->{ $wheel_id };
  $self->{wheels}->{ $wheel_id }->{wheel}->kill(9);
  return;
}

sub _wheel_close {
  my ($kernel,$self,$wheel_id) = @_[KERNEL,OBJECT,ARG0];
  my $wdata = delete $self->{wheels}->{ $wheel_id };
  $kernel->alarm_remove( $wdata->{alarm} ) if $wdata->{alarm};
  if ( defined $wdata->{tcp} ) {
    my $client_id = $wdata->{client};
    return unless $self->{clients}->{ $client_id };
    if ( $self->{clients}->{ $client_id }->{pending} ) {
	$self->{clients}->{ $client_id }->{shutdown} = 1;
	return;
    }
    delete $self->{clients}->{ $client_id };
  }
  return;
}

sub _wheel_error {
  my ($operation, $errnum, $errstr, $wheel_id) = @_[ARG0..ARG3];
  return if $operation eq "read" and !$errnum;
  $errstr = "remote end closed" if $operation eq "read" and !$errnum;
  warn "Wheel $wheel_id generated $operation error $errnum: $errstr\n";
  return;
}

sub _get_datagram {
}

qq[Inet in'it];

__END__