File Coverage

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

Criterion	Covered	Total	%
statement	39	162	24.0
branch	0	44	0.0
condition	0	12	0.0
subroutine	13	31	41.9
pod	3	3	100.0
total	55	252	21.8

# -*- encoding: utf-8; mode: cperl -*-

package POE::Component::Server::BigBrother;

use strict;
use warnings;
use Carp;

#sub POE::Kernel::TRACE_REFCNT () { 1 }
#sub POE::Kernel::ASSERT_DEFAULT () { 1 }
#sub POE::Kernel::TRACE_DEFAULT  () { 1 }
#sub POE::Kernel::TRACE_SIGNALS ()  { 1 }
#sub POE::Kernel::ASSERT_EVENTS ()  { 1 }

use base qw(POE::Component::Pluggable);
use POE;
use POE::Component::Pluggable::Constants qw(:ALL);
use POE::Component::Server::TCP;
use POE::Filter::BigBrother;
use POE::Filter::Stream;

use Log::Report syntax => 'SHORT';

# use Smart::Comments;

use vars qw($VERSION);

$VERSION='0.08';

#
# constants
#
use constant DATA_TRUNCATED_MESSAGE        => "... DATA TRUNCATED ...";
use constant DATA_TRUNCATED_MESSAGE_LENGTH => length(DATA_TRUNCATED_MESSAGE);

sub spawn {
    my $package = shift;
    my %opts    = @_;
    $opts{ lc $_ } = delete $opts{$_}
      for keys %opts;  # convert opts to lower case
    my $options = delete $opts{options};
    my $self = bless \%opts, $package;

	$self->_pluggable_init(prefix => 'bb_', types => [ 'MESSAGE', 'EVENT' ]);

    # default values
    $self->{time_out}  ||= 30;        # default time_out
    $self->{bind_port} ||= 1984;      # default bind port
    $self->{max_msg_size} ||= 16384;  # default max message size

    $self->{session_id} = POE::Session->create(
        object_states => [
            $self => {
                       shutdown   => '_on_shutdown',
                       register   => '_on_register',
                       unregister => '_on_unregister'
                     },
			$self => [
                qw (
                  _start
                  _on_dispatch
                  )
            ],
        ],
        heap => $self,
        ( ref($options) eq 'HASH' ? ( options => $options ) : () ),
                                              )->ID();
    return $self;
}

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

sub _pluggable_event {
	my ($self) = @_;
	### _pluggable_event: @_
	$poe_kernel->post($self->{session_id}, '_dispatch');
}

sub _on_dispatch {
	my ($kernel,$self,$event,@args) = @_[KERNEL,OBJECT,ARG0,ARG1..$#_];
	$self->_dispatch( $event, @args );
	return;
}

sub _dispatch {
    my ( $self, $event, @args ) = @_;
    ## _dispatch event: $event
	return 1 if $self->_pluggable_process('MESSAGE', $event, \(@args)) == PLUGIN_EAT_ALL;

    my %sessions;

    # concatenate all session wich correspond to the event
    foreach (
            values %{ $self->{events}->{ $self->{_pluggable_prefix} . 'all' } },
            values %{ $self->{events}->{$event} } ) {
        $sessions{$_} = $_;
    }
    $poe_kernel->post( $_, $event, @args ) for values %sessions;
}

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__ );
    }

    # Adapt to POE::Component::Server::TCP v1.020 new args
    my $poco_tcp_args
      = POE::Component::Server::TCP->VERSION > 1.007 ? "ClientArgs" : "Args";

    ## Create a tcp server that receives BigBrother messages.  It
    ## will be referred to by the name "server_tcp" when necessary.
    ## It listen on port 1984.  It uses POE::Filter::Block to parse
    ## input and format output.
    $self->{listener} =
      POE::Component::Server::TCP->new(
        ( $self->{alias} ? ( Alias => $self->{alias} . '_tcp_listener' ) : () ),
        $poco_tcp_args     => [ self => $self ],
        Address            => $self->{bind_addr},
        Port               => $self->{bind_port},
        Concurrency        => -1,
        Error              => \&_on_tcp_server_error,
        ClientConnected    => \&_on_client_connect,
        ClientDisconnected => \&_on_client_disconnect,
        ClientFilter       => POE::Filter::Stream->new(),
        ClientInput        => \&_on_client_input,
        InlineStates       => { '_conn_alarm' => \&_conn_alarm, },
      );

    return;
}

sub _conn_alarm {
	### _conn_alarm
	$_[KERNEL]->yield('shutdown');
}

# Register some event(s)
sub _on_register {
	my ($kernel, $session, $sender, $self, @events) = @_[KERNEL, SESSION, SENDER, OBJECT, ARG0 .. $#_ ];
	croak 'Not enough arguments' unless @events;

	my $sender_id = $sender->ID();

	foreach my $event (@events) {
		$event = $self->{_pluggable_prefix} . $event unless $event =~ /^_/;
		$self->{events}->{$event}->{$sender_id} = $sender_id;
		$self->{sessions}->{$sender_id}->{'ref'} = $sender_id;
		unless ($self->{sessions}->{$sender_id}->{refcnt}++ or $session == $sender) {
			# One count for every event
			$kernel->refcount_increment($sender_id, __PACKAGE__);
		}
		$kernel->yield(_dispatch => $self->{_pluggable_prefix} . 'registered', $sender_id);
	}
	return;
}

sub _on_unregister {
    my ( $kernel, $self, $session, $sender, @events ) =
      @_[ KERNEL, OBJECT, SESSION, SENDER, ARG0 .. $#_ ];

    die "Not enough arguments for unregister event" unless @events;

    my $sender_id = $sender->ID();
    foreach (@events) {
        delete $self->{events}->{$_}->{$sender_id};
        if ( --$self->{sessions}->{$sender_id}->{refcnt} <= 0 ) {
            delete $self->{sessions}->{$sender_id};
            unless ( $session == $sender ) {
                $kernel->refcount_decrement( $sender_id, __PACKAGE__ );
            }
        }
    }
    return;
}

sub _unregister_sessions {
  my $self = shift;
  foreach my $session_id ( keys %{ $self->{sessions} } ) {
     my $refcnt = $self->{sessions}->{$session_id}->{refcnt};
     while ( $refcnt --> 0 ) {
		 $poe_kernel->refcount_decrement($session_id, __PACKAGE__);
     }
     delete $self->{sessions}->{$session_id};
  }
}

sub _on_shutdown {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  ### Shutting down BigBrother Gateway
  $self->_unregister_sessions();
  $kernel->alarm_remove_all();
  $kernel->alias_remove( $_ ) for $kernel->alias_list();
  $kernel->post( $self->{listener}, 'shutdown' ) if $self->{listener};
  $self->_pluggable_destroy();
  ### Waiting all clients to disconnect
  return;
}

sub _on_tcp_server_error {
	my ($syscall_name, $error_number, $error_string) = @_[ARG0, ARG1, ARG2];
	croak "BigBrother Gateway: $syscall_name error because $error_string\n";
}

sub _on_client_connect {
	my ($kernel, $sender, $heap) = @_ [ KERNEL, SESSION, HEAP ];
	my %args;
	if ( ref $_[ARG0] eq 'HASH' ) {
		%args = %{ $_[ARG0] };
	} elsif ( ref $_[ARG0] eq 'ARRAY' ) {
		%args = @{ $_[ARG0] };
	} else {
		%args = @_[ARG0..$#_];
	}
	my $self = delete $args{self};
	$heap->{bb_server} = $self; # store self object on the client session as server
    $heap->{buffer}  = '';
	_delay_timeout($kernel, $self);
}

sub _on_client_input {
    # Accumulate datas
    $_[HEAP]->{buffer} .= $_[ARG0];
}

sub _decode_bb_message {
    my ($self, $input) = @_;
    my $message = undef;
    my $input_length = length($input);

    if ( $input_length > $self->{max_msg_size} ) {
        print STDERR "Truncated too long BigBrother message ($input_length > ",
          $self->{max_msg_size} . "):\n", substr( $input, 0, 80 ), "\n";
        my $pos = $self->{max_msg_size} - DATA_TRUNCATED_MESSAGE_LENGTH;
        substr( $input, $pos, $input_length - $pos, DATA_TRUNCATED_MESSAGE );
    }

    if (
        $input =~ m/^
                    ((?:(?:(?:dis|en)abl|pag)e|status)) # the command ($1)
                    (\+\d+)? 		# the offset ($2)
                    \s+				# some spaces
                    (\S+?)\.(\S+)	# server.probe ($3, $4)
                    \s+ 			# some spaces
                    (.*)$ 			# last args ($5)
                   /sx
      ) {
        my $command = lc($1);
        $message->{command}   = $command;
        $message->{offset}    = $2 if defined $2;
        $message->{host_name} = $3;
        $message->{probe}     = $4;
        my $args 			  = $5;

        $message->{host_name} =~ tr/,/./;    # Translate server fqdn

        if ( $command eq 'enable' ) {    # Enable command
            $message->{data} = $args;
        } else {
            my ( $arg1, $arg2 ) = split( /\s+/, $args, 2 );
            if ( $command eq 'status' or $command eq 'page' ) {
                $message->{color} = $arg1;
            }
			else {   # Disable command
                $message->{period} = $arg1;
            }
            $message->{data} = $arg2;
        }
    } elsif ($input =~ m/^(?:event\s+)(.+)$/si) {
		$message = { command => 'event', params => $1 };
    } else {
        warning "Unknown BB message: ".substr($input,0,80);
    }

    return $message;
}

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

    # Remove all alarms
    $kernel->alarm_remove_all();

    # Check if we have receive some datas
    if (length $heap->{buffer}) {
        # Create a new filter to parse raw BB messages
        my $filter = POE::Filter::BigBrother->new();
        my $bb_server = $heap->{bb_server};
        # Decode any BB messages
        foreach my $cooked_input ( @{ $filter->get( [ $heap->{buffer} ] ) } ) {
            if ( my $message = $bb_server->_decode_bb_message($cooked_input) ) {
                $bb_server->_dispatch(
                          $bb_server->{_pluggable_prefix} . $message->{command},
                          $message, $bb_server );
            }
        }
    }
}

sub _delay_timeout {
	my ($kernel,$self) = @_;
	$kernel->delay( '_conn_alarm', $self->{time_out} );
}

sub session_id {
	my ($self) = @_;
	return $self->{session_id};
}

1; # End of POE::Component::Server::BigBrother
__END__