File Coverage

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

Criterion	Covered	Total	%
statement	65	99	65.6
branch	11	28	39.2
condition	2	9	22.2
subroutine	12	16	75.0
pod	4	8	50.0
total	94	160	58.7

# $Id: TCP.pm 446 2004-12-27 00:57:57Z sungo $
package POE::Component::Server::Syslog::TCP;
$POE::Component::Server::Syslog::TCP::VERSION = '1.22';
#ABSTRACT: syslog tcp server

use warnings;
use strict;

sub BINDADDR        () { '0.0.0.0' }
sub BINDPORT        () { 514 }
sub DATAGRAM_MAXLEN () { 1024 }  # syslogd defaults to this. as do most
                                 # libc implementations of syslog

use Params::Validate qw(validate_with);
use Carp qw(carp croak);
use Socket;

use POE qw(
	Driver::SysRW
	Wheel::SocketFactory
	Wheel::ReadWrite
	Filter::Syslog
);


sub spawn {
	my $class = shift;

	my %args = validate_with(
		params => \@_,
		spec => {
			InputState   => {
				type     => &Params::Validate::CODEREF,
				optional => 1,
				default  => sub {},
			},
			ErrorState   => {
				type     => &Params::Validate::CODEREF,
				optional => 1,
				default  => sub {},
			},
			BindAddress  => {
				type     => &Params::Validate::SCALAR,
				optional => 1,
				default  => BINDADDR,
			},
			BindPort     => {
				type     => &Params::Validate::SCALAR,
				optional => 1,
				default  => BINDPORT,
			},
			MaxLen       => {
				type     => &Params::Validate::SCALAR,
				optional => 1,
				default  => DATAGRAM_MAXLEN,
			},
			Alias  		 => {
				type     => &Params::Validate::SCALAR,
				optional => 1,
			},
		},
	);

	$args{type} = 'tcp';
	$args{filter} = POE::Filter::Syslog->new();

	my $sess = POE::Session->create(
		inline_states => {
			_start         => \&start,
			_stop          => \&shutdown,

			socket_connect => \&socket_connect,
			socket_error   => \&socket_error,
			socket_input   => \&socket_input,
			register	   => \®ister,
			unregister	   => \&unregister,
			shutdown       => \&shutdown,

			client_input => $args{InputState},
			client_error => $args{ErrorState},

		},
		heap => \%args,
	);


	return $sess;
}

sub start {
	$_[HEAP]->{socketfactory} = POE::Wheel::SocketFactory->new(
		BindAddress  => $_[HEAP]->{BindAddress},
		BindPort     => $_[HEAP]->{BindPort},
		SuccessEvent => 'socket_connect',
		FailureEvent => 'client_error',
		ListenQueue  => $_[HEAP]->{MaxLen},
		Reuse        => 'yes',
	);

	unless($_[HEAP]->{socketfactory}) {
		croak("Unable to setup socketfactory");
	}
	$_[KERNEL]->alias_set( $_[HEAP]->{Alias} ) if $_[HEAP]->{Alias};
    return;
}

sub socket_connect {
	my $handle = $_[ARG0];
	my $host;

	if( ( sockaddr_in( getpeername($handle) ) )[1]) {
		$host = gethostbyaddr( ( sockaddr_in( getpeername($handle) ) )[1], AF_INET );
	}
    else {
		$host = '[unknown]';
	}

	my $wheel = POE::Wheel::ReadWrite->new(
		Handle     => $handle,
		Driver     => POE::Driver::SysRW->new(),
		Filter     => POE::Filter::Syslog->new(),
		InputEvent => 'socket_input',
		ErrorEvent => 'socket_error',
	);

	$_[HEAP]->{wheels}->{ $wheel->ID } = {
		wheel => $wheel,
		host  => $host,
	};
    return;
}

sub socket_error {
	my ($errop, $errnum, $errstr, $wid) = @_[ARG0 .. ARG3];
	unless( ($errnum == 0) && ($errop eq 'read') ) {
		$_[KERNEL]->yield( 'client_error', $errop, $errnum, $errstr );
	}
	delete $_[HEAP]->{wheels}->{ $wid };
    return;
}

sub socket_input {
	my ($input, $wid) = @_[ARG0, ARG1];
	my $info = $_[HEAP]->{wheels}->{ $wid };

	if(ref $input && ref $input eq 'ARRAY') {
		foreach my $record (@{ $input }) {
			$input->{host} = $info->{host};
			$_[KERNEL]->yield( 'client_input', $record );
		}
	}
    elsif(ref $input && ref $input eq 'HASH') {
		$input->{host} = $info->{host};
		$_[KERNEL]->yield( 'client_input', $input );
		$_[KERNEL]->post( $_, $_[HEAP]->{sessions}->{$_}->{inputevent}, $input )
			for keys %{ $_[HEAP]->{sessions} };
	}
    else {
		$_[KERNEL]->yield( 'client_error', $input );
		$_[KERNEL]->post( $_, $_[HEAP]->{sessions}->{$_}->{errorevent}, $input )
			for grep { defined $_[HEAP]->{sessions}->{errorevent} }
			    keys %{ $_[HEAP]->{sessions} };
	}
    return;
}

sub shutdown {
    my ($kernel,$heap) = @_[KERNEL,HEAP];
	if($heap->{socketfactory}) {
		$heap->{socketfactory}->pause_accept();
		delete $heap->{socketfactory};
	}
	delete $heap->{wheels};
	$kernel->alarm_remove_all();
	$kernel->alias_remove( $_ ) for $kernel->alias_list();
	$kernel->refcount_decrement( $_, __PACKAGE__ )
		for keys %{ $heap->{sessions} };
    return;
}

sub register {
  my ($kernel,$self,$sender) = @_[KERNEL,HEAP,SENDER];
  my $sender_id = $sender->ID();
  my %args;
  if ( ref $_[ARG0] eq 'HASH' ) {
    %args = %{ $_[ARG0] };
  }
  elsif ( ref $_[ARG0] eq 'ARRAY' ) {
    %args = @{ $_[ARG0] };
  }
  else {
    %args = @_[ARG0..$#_];
  }
  $args{lc $_} = delete $args{$_} for keys %args;
  unless ( $args{inputevent} ) {
    warn "No 'inputevent' argument supplied\n";
    return;
  }
  if ( defined $self->{sessions}->{ $sender_id } ) {
    $self->{sessions}->{ $sender_id } = \%args;
  }
  else {
    $self->{sessions}->{ $sender_id } = \%args;
    $kernel->refcount_increment( $sender_id, __PACKAGE__ );
  }
  return;
}

sub unregister {
  my ($kernel,$self,$sender) = @_[KERNEL,HEAP,SENDER];
  my $sender_id = $sender->ID();
  my %args;
  if ( ref $_[ARG0] eq 'HASH' ) {
    %args = %{ $_[ARG0] };
  }
  elsif ( ref $_[ARG0] eq 'ARRAY' ) {
    %args = @{ $_[ARG0] };
  }
  else {
    %args = @_[ARG0..$#_];
  }
  $args{lc $_} = delete $args{$_} for keys %args;
  my $data = delete $self->{sessions}->{ $sender_id };
  $kernel->refcount_decrement( $sender_id, __PACKAGE__ ) if $data;
  return;
}

1;


# sungo // vim: ts=4 sw=4 noexpandtab

__END__