File Coverage

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

Criterion	Covered	Total	%
statement	73	99	73.7
branch	12	28	42.8
condition	3	6	50.0
subroutine	13	16	81.2
pod	4	6	66.6
total	105	155	67.7

# $Id: UDP.pm 449 2004-12-27 01:37:33Z sungo $
package POE::Component::Server::Syslog::UDP;

use warnings;
use strict;

our $VERSION = '1.20';

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 POE;
use POE::Filter::Syslog;

use Socket;
use IO::Socket::INET;

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} = 'udp';
	$args{filter} = POE::Filter::Syslog->new();

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

			select_read    => \&select_read,
			register	   => \®ister,
			unregister	   => \&unregister,
			shutdown       => \&shutdown,

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

		},
		heap => \%args,
	);

	return $sess;
}


# This is a really good spot to discuss why this is using IO::Socket
# instead of a POE wheel of some variety for this. The answer, for once
# in my life, is pretty simple. POE::Wheel::SocketFactory doesn't support
# connectionless sockets as of the time of writing. In this scenario,
# there is no chance of IO::Socket blocking, unless IO::Socket decides
# to lose its mind. If it does THAT, there's not a whole hell of a lot
# left that's right in the world. :) except maybe pizza. well, good
# pizza like you find at Generous George's in Alexandria, VA. and rum.
# pretty much any rum. Um, but anyway...

sub socket_start {
	$_[HEAP]->{handle} = IO::Socket::INET->new(
		Blocking   => 0,
		LocalAddr  => $_[HEAP]->{BindAddress},
		LocalPort  => $_[HEAP]->{BindPort},
		Proto      => 'udp',
		ReuseAddr  => 1,
		SocketType => SOCK_DGRAM,
	);

	if (defined $_[HEAP]->{handle}) {
		$_[KERNEL]->select_read( $_[HEAP]->{handle}, 'select_read' );
	} else {
		croak "Unable to create UDP Listener: $!";
	}
	$_[KERNEL]->alias_set( $_[HEAP]->{Alias} ) if $_[HEAP]->{Alias};
	return;
}

sub select_read {
	my $message;
	my $remote_socket = $_[HEAP]->{handle}->recv($message, $_[HEAP]->{MaxLen}, 0 );
	if (defined $message) {
		$_[HEAP]->{filter}->get_one_start([ $message ]);
		my $records = [];
		while( ($records = $_[HEAP]->{filter}->get_one()) and (@$records > 0)) {
			if(defined $records and ref $records eq 'ARRAY') {
				foreach my $record (@$records) {
					if (my $addr = (sockaddr_in($remote_socket))[1]) {
						$record->{addr} = inet_ntoa($addr);
						if (my $host = gethostbyaddr($addr, AF_INET)) {
							$record->{host} = $host;
						}
					}

					$_[KERNEL]->yield( 'client_input', $record );
					$_[KERNEL]->post( $_, $_[HEAP]->{sessions}->{$_}->{inputevent}, $record )
						for keys %{ $_[HEAP]->{sessions} };
				}
			} else {
				$_[KERNEL]->yield( 'client_error', $message );
				$_[KERNEL]->post( $_, $_[HEAP]->{sessions}->{$_}->{errorevent}, $message )
					for grep { defined $_[HEAP]->{sessions}->{errorevent} }
						keys %{ $_[HEAP]->{sessions} };
			}
		}
	}
	return;
}

sub shutdown {
	my ($kernel,$heap) = @_[KERNEL,HEAP];
	if($heap->{handle}) {
		$kernel->select_read($heap->{handle});
		$heap->{handle}->close();
	}
	delete $heap->{handle};
	$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;
__END__