File Coverage

blib/lib/POE/Event/Message.pm

Criterion	Covered	Total	%
statement	13	15	86.6
branch			n/a
condition			n/a
subroutine	5	5	100.0
pod			n/a
total	18	20	90.0

# -*- Perl -*-
#
# File:  POE/Event/Message.pm
# Desc:  A generic messaging protocol - to use as a starting point
# Date:  Mon Oct 10 10:35:59 2005
# Stat:  Prototype, Experimental
# Note:  Fix to remove message length limitation contributed
#        by Martin Holste, Sun 21 Feb 2010
#
package POE::Event::Message;
use 5.006;
use strict;
use warnings;

our $PACK    = __PACKAGE__;
our $VERSION = '0.11';
### @ISA     = qw( );

# WARN:  Don't use POE or the POE::Kernel classes here. This is a
#        generic messaging class that can also be used outside of
#        the POE environment. Support is included for sending and
#        receiving messages across sockets (such as between a POE-
#        based server process and a non-POE client process), and
#        when sending messages across file handles (as in the case
#        of a POE-based parent process communicating with a non-POE
#        child process).
#

### POE::Kernel;                      # Don't use POE here!
use POE::Event::Message::Header;      # Header class wraps attributes
### POE::Event::Message::Body;        # ((not yet created))
use POE::Filter::Reference;           # Perl object filtering
use POE::Driver::SysRW;               # sysread/syswrite driver

my $HeaderClass = "POE::Event::Message::Header";
## $BodyClass   = "POE::Event::Message::Body";  # ((not yet created))
our $Driver     = new POE::Driver::SysRW;
our $Filter     = new POE::Filter::Reference;

#-------------------------------------------------------------------------
# Message Creation (new message and reply message)
# Envelope method to create new message with body data

sub new
{   my($class,$header,$body) = @_;

    # This method allows for the following usage:
    # . Creating a new message
    #     $message = $msgClass->new();
    #     $message = $msgClass->new( undef, $msgBody );  # see 'package' method
    #
    # . Creating a reply to an existing message
    #     $response = $msgClass->new( $message->header() );    # header obj
    #     $response = $msgClass->new( $message           );    # message obj
    #
    if (ref($header) and $header =~ /=HASH/ and $header->can("new") ) {
	my $msg = $header;
	$header = $HeaderClass->new( $msg->header() ),

    } else {
	$header = $HeaderClass->new( $header ),
    }

    return( bless my $self = {
	     header => $header,
	       body => $body,
	    ## body => $BodyClass->new( $body ),    # ((no BodyClass yet))
	}, ref($class)||$class
    );
}

*envelope = \&package;

sub package
{   my($class,$msgBody) = @_;
    return $class->new("", $msgBody);        # create new message "envelope"
}

#-----------------------------------------------------------------------
# Pass through methods for message header object
# and a method to get/set the message body

sub set    { $_[0]->{header} and $_[0]->{header}->set   ( $_[1], $_[2] ) }
sub get    { $_[0]->{header} and $_[0]->{header}->get   ( $_[1]        ) }
sub del    { $_[0]->{header} and $_[0]->{header}->del   ( $_[1]        ) }
sub param  { $_[0]->{header} and $_[0]->{header}->param ( $_[1], $_[2] ) }
sub setErr { $_[0]->{header} and $_[0]->{header}->setErr( $_[1], $_[2] ) }
sub status { $_[0]->{header} and $_[0]->{header}->status()               }
sub stat   { $_[0]->{header} and $_[0]->{header}->stat()                 }
sub err    { $_[0]->{header} and $_[0]->{header}->err()                  }

sub getMode      { $_[0]->{header} and $_[0]->{header}->mode()           }
sub setMode      { $_[0]->{header} and $_[0]->{header}->mode( $_[1] )    }
sub delRouteBack { $_[0]->{header} and $_[0]->{header}->delRouteBack()   }
sub delRouteTo   { $_[0]->{header} and $_[0]->{header}->delRouteTo()     }

*delete = \&del;
*reset  = \&del;
*inResponseToId = \&r2id;

sub hasRouting   { $_[0]->{header} and $_[0]->{header}->hasRouting()     }
sub hasRouteTo   { $_[0]->{header} and $_[0]->{header}->hasRouteTo()     }
sub hasRouteBack { $_[0]->{header} and $_[0]->{header}->hasRouteBack()   }

*getRouting   = \&hasRouting;
*getRouteTo   = \&hasRouteTo;
*getRouteBack = \&hasRouteBack;

sub nextRouteType     { $_[0]->{header}->nextRouteType()     if $_[0]->{header}}
sub nextRouteIsRemote { $_[0]->{header}->nextRouteIsRemote() if $_[0]->{header}}
sub nextRouteIsLocal  { $_[0]->{header}->nextRouteIsLocal()  if $_[0]->{header}}
sub nextRouteIsPost   { $_[0]->{header}->nextRouteIsPost()   if $_[0]->{header}}
sub nextRouteIsCall   { $_[0]->{header}->nextRouteIsCall()   if $_[0]->{header}}

sub id     { $_[0]->{header}->get('id')  }    # unique message ID
sub r2id   { $_[0]->{header}->get('r2id')}    # orig. unique message ID
sub header { $_[0]->{header} }                                 # get only
sub body   { $_[1] ? $_[0]->{body} = $_[1] : $_[0]->{body} }   # get or set

sub addRouteTo
{   my($self, @args) = @_;
    return undef unless $self->{header};
    return $self->{header}->addRouteTo( @args );
}

sub addRouteBack
{   my($self, @args) = @_;
    return undef unless $self->{header};
    return $self->{header}->addRouteBack( @args );
}

sub addRemoteRouteTo
{   my($self, @args) = @_;
    return undef unless $self->{header};
    return $self->{header}->addRemoteRouteTo( @args );
}

sub addRemoteRouteBack
{   my($self, @args) = @_;
    return undef unless $self->{header};
    return $self->{header}->addRemoteRouteBack( @args );
}

#-----------------------------------------------------------------------
# Message Routing and Auto-Replies
# This is an experimental interface to provide a flexible,
# semi-automatic message routing mechanism. It provides
# the following features as an alternative to "postbacks"
# and extends the concept somewhat.
#
# . immediate routing via "post"  (queued, asynchronous) - post method
# . immediate routing via "call"  (direct,  synchronous) - call method
#
# . forward routing, delayed      (via "post" or "call)  - addRouteTo
# . return routing,  delayed      (via "post" or "call)  - addRouteBack
# . combinations of these two
#
# . forward routing, delayed remote (via socket "write") -addRemoteRouteTo
# . return  routing, delayed remote (via socket "write") -addRemoteRouteBack
# . combinations of these two
#
# In addition, this allows the "stacking" of both forward and
# return destinations. This is the basis of the "semi-automatic
# routing" feature. The "route" method DOES need to be called
# on a message, but the caller DOESN'T need to know anything
# about the next destination(s).
#
# Some experimentation and experience will determine whether
# these are useful as designed, or if changes are needed.
# The INTENT is that intermediate destinations can be added
# to interrupt the original routing, TEMPORARIALY redirecting
# a message, while STILL allowing it to return as intended.

#---------------------------------------------
# Direct message routing, via "post" or "call"

sub post { shift->_postOrCall( "post", @_ ) }
sub call { shift->_postOrCall( "call", @_ ) }

sub _postOrCall                                      # For immediate routing
{   my($self, $mode, $session, $event, @args) = @_;

    if (! defined $INC{'POE/Kernel.pm'}) {
	return $self->setErr(-1, "'POE::Kernel' module is not loaded in '_postOrCall' method of '$PACK'");
    }

    $session ||= POE::Kernel->get_active_session()->ID();

    return unless ($session and $event);
    $mode = lc $mode;                            # a mode arg?
    $mode ||= lc $self->getMode();               # use prior setting?
    $mode = "post" unless ($mode eq "call");     # default value!

    # NOTE that this changes the default behavior of POE's post/call.
    # Here, arguments are bundled into list references so the result
    # is "more compatible" with POE's postback/callback mechanism.
    #
    # Here, we ensure that the arguments passed here are sent to the
    # target event handler as a list reference to be received in ARG0.
    # The "$self" var (current message) is sent as the only element in
    # a list reference to be received in ARG1. This allows "$self" to
    # be "received" as if it was sent by the method where post/call was
    # invoked. In addition, this maintains a compatible result with
    # the "addRouteTo()/addRouteBack()" method, below. Clear as mud?
    # This is either a good idea or really cludgy. Time will tell.
    #
    # All message events sent via "post()", "call()" or "route()" will
    # result in ONLY two parameters received by the target event handler.

    # issue one of post/call

    my $argRef;

    if ( ($#args == 0) and (ref($args[0]) eq "ARRAY") ) {
	$argRef = $args[0];
    } else {
	$argRef = [ @args ];
    }

    # FIX: handle 'call' return here as done in sub '_autoRouting'

    if (! POE::Kernel->$mode( $session, $event, $argRef, [ $self ] ) ) {

	my $msg = "Error(1): POE_Kernel->$mode( $session, $event, [ $argRef ], [ $self ] )";

	$self->setErr(-1, "$msg: $! in '$PACK'");
	## warn $msg;
    }

  # warn "DEBUG: Route: POE_Kernel->$mode( $session, $event, @args )\n";

    return;
}

#---------------------------------------------
# Automatic message routing, via "post" or "call"

*autoroute = \&route;
*autoRoute = \&route;

sub route                              # forward, if possible, or reply
{   my($self,@newArgs) = @_;

    # routeTo() until empty, then routeBack()
    # 'post' returns:   1=routed,  0=nowhere to route
    # 'call' returns:   whatever 'called' method returns

    my(@list) = ();

    if ( $self->hasRouteTo() ) {
	(@list) = $self->routeTo( @newArgs );
    } elsif ( $self->hasRouteBack() ) {
	(@list) = $self->routeBack( @newArgs );
    }

    return( @list );
}

#---------------------------------------------
# Forward message routing, via "post" or "call"

*forward = \&routeTo;
*routeto = \&routeTo;

sub routeTo                            # activate "auto-forward", if any
{   my($self,@newArgs) = @_;

    my(@list) = $self->_autoRouting( "delRouteTo", @newArgs );
    return( @list );
}

#---------------------------------------------
# Return message routing, via "post" or "call"

*reply     = \&routeBack;
*routeback = \&routeBack;

sub routeBack                          # activate "auto-reply", if any
{   my($self,@newArgs) = @_;

    my(@list) = $self->_autoRouting( "delRouteBack", @newArgs );
    return( @list );
}

#---------------------------------------------
# Generic message routing method

sub _autoRouting                       # activate "auto-routing," if any
{   my($self,$type,@newArgs) = @_;

    # This "generic" method is intended to be called by
    # either the "routeTo" or "routeBack" method, which
    # in turn might be invoked by the "route" method.

    return undef unless $self->{header};

    #---------------------------------------------------------------
    # Depending on the "$type" collect a "RouteTo" or a "RouteBack"
    # Note: "@origArgs" is now included for 'delRouteTo' type.

    $type = "delRouteBack" unless $type eq "delRouteTo";

    my($host, $port, $mode, $session, $event, @origArgs) = $self->$type();

    ##warn "($host, $port, $mode, $session, $event, @origArgs)\n";

    # Must have something to post/call and somewhere to send it!
    # FIX: differentiate between call when "no routing"
    # and call when "incomplete routing" attributes.
    #
    if (! ($session and $event) ) {
	## warn "DEBUG: session='$session'  event='$event'";
	return;
    }

    #---------------------------------------------------------------
    # FIX: ToDo: complete remote routing via '$host', '$port'
    # using the "write" method, below. First, fix up message such
    # that, upon being received, a "route()" will forward it on.
    #
 ###if ($host or $port) {

    if ($port) {
	$host ||= 'localhost';

	my $socket = $self->createSocket( $host, $port );
	if (! $socket) {
	    my $err = "Error: connection to '${host}:$port' refused in '$PACK'";

	    # FIX: Do we overwrite the original error message here? or not??
	    # YES. This is a better (clearer) message here for the user.
	    #
	  # $self->setErr(-1, $err)  unless $self->status();
	    $self->setErr(-1, $err);

	    return $self;
	}

	my($stat,$err) = $self->write( $socket );
	if ($stat) {
	    $self->setErr( $stat, $err );
	    return $self;
	}

	my $response;

	# With a "call" or "sync" mode we expect an immediate
	# response message from the server. With any other
	# mode, such as "async", only a "successful send"
	# response is generated here--this assumes that if
	# an actual response message will be generated, an
	# appropriate "addRemoteRouteBack" was added earlier.

	if (($mode eq "call") or ($mode eq "sync")) {
	    $response = $self->read( $socket );
	} else {
            # FIX: Should this be handled differently??
	    $response = $self->new( $self );
	    $response->setErr( 0, "message sent successfully" );
        }

	$self->shutdownSocket();

	return $response;
    }

    #---------------------------------------------------------------
    # Determine whether we "post" (asych) or "call" (synch)

    $mode ||= lc $self->mode();                  # use prior setting?
    $mode = "post" unless ($mode eq "call");     # default value!

    #---------------------------------------------------------------
    # If we're going to rely on POE, make sure it's available.

    if (! defined $INC{'POE/Kernel.pm'}) {
	$self->setErr(-1, "'POE::Kernel' module is not loaded in '_autoRouting' method of '$PACK'");
	return undef;
    }

    #---------------------------------------------------------------
    # This now works as does 'post/call', when using 'routeTo'.
    # STRANGE SYNTAX here, but it allows for reasonably compatible
    # syntax (and a plug-compatible result) to POE's "postback"
    # mechanism.
    #
    # NOTE that "$self" is prepended to the "@args" list here! This
    # allows "$self" (current message) to be "received" as if it was
    # sent by the method where the 'post/call/postback' was invoked.
    # --- This obviously needs some well documented examples. ---

    my (@args);
    if ($type eq "delRouteTo") {                  # issue a post/call
	# This is kind of funky usage but, at the moment, it
	# seems to be "the right thing." Feedback is welcome.
	# Currently we PREPEND "$self" (current message) to
	# any "new args" when "route()" method is called.
	#
	# Adding "$self" AFTER the orig args and BEFORE added args
	# is either really good or really cludgy. Time will tell.
	#
	(@args) = ( [ @origArgs ], [ $self, @newArgs ] );

    } else {                                      # emulate postback
	# Currently, these are the same. Should the above change?
	#
	(@args) = ( [ @origArgs ], [ $self, @newArgs ] );
    }

    my(@list) = POE::Kernel->$mode( $session, $event, @args );

    if (! @list ) {

	if ($!) {
	    my $err = "Error(2): POE_Kernel->$mode( $session, $event, @args )";

	    $self->setErr(-1, "$err: $! in '$PACK'");
	    ## warn $err;
	}
    }

    ## warn "DEBUG: Route: POE_Kernel->$mode( $session, $event, @args )\n";

    # Here we return the same result as the POE::Kernel
    # .  for 'post' return a boolean indicating success
    # .  for 'call' return what the called method returns
    #
    return( @list )  if ($mode eq 'call');
    return 1;
}

#-----------------------------------------------------------------------
# Reading and Writing messages, via POE::Filter::Reference,
# for use with file/socket handles. Note that the "$message"
# argument defaults to "$self" -- this allows for nice clean
# syntax in most situations:     $message->write( $fh );

*send     = \&write;
*recv     = \&read;

sub write
{   my($self,$fh,$message) = @_;

    (defined($fh) and length($fh)) or return (1,"nowhere to write");

    # Support class OR object method
    $message = $self   if ((! $message) and ref($self));
    return (1,"nothing to write") unless $message;

    ## warn "DEBUG: write message='$message'\n";

    my $tmp = delete $self->{_socket_}   if (ref $self);
    ## warn "DEBUG fh='$fh' in 'write()'";

    $Driver->put( $Filter->put( [ $message ] ) );    # queue the message
    $Driver->flush( $fh );                           # send the message

    $! and warn "OUCH: Driver/flush failed: $!";
 ## $!  or warn "OKAY: Driver/flush succeeded";

    $self->{_socket_} = $tmp   if ($tmp and ref $self);
    return(0,"");                                    # return success
}

sub read
{   my($self,$fh) = @_;

    ## warn "DEBUG: fh='$fh' in 'read()'";

    (defined($fh) and length($fh)) or return (1,"nowhere to read");

    my $response = $Driver->get( $fh );              # fetch a reply

    if (! $response) {
	$response = $self->new();
	$response->setErr(-1, "no response from server");
	return $response;
    }

    #-------------------------------------------------------------------
    # Fix to remove message length limitation contributed
    # by Martin Holste, Sun 21 Feb 2010:
    # Parse expected length from the beginning of the first payload
    unless (ref($response) eq 'ARRAY' and $response->[0] =~ /^(\d+)\0/){
        $response = $self->new();
        $response->setErr(-1, 'Error: Did not receive expected payload length at beginning of payload');
        return $response;
    }
    my $expected_length = $1 + length($1) + 1;
    my $received = length($response->[0]);
    READ_LOOP: while ($received < $expected_length){
        my $buf = $Driver->get($fh);
        foreach (@$buf){
	    push @$response, $_;
	    $received += length($_);
        }
    }
    #-------------------------------------------------------------------
    # filter the reply, or report error at this line:
    ($response) = @{ $Filter->get( $response ) };    my $line = __LINE__;

    if (($response) and ($response =~ /=HASH/) and ($response->can( "body" ))) {
	# response is okay
    } else {
	my $body = $response ||"";
	$response = $self->new();
	$response->body( $body );
	my $err  = "Error: unknown message type in 'read' in '$PACK' at line $line";
	   $err .= "\nmessage = '$body'";
	   $err .= "  (was message too big to send?)"  if (! $body);
	$response->setErr(-1, $err );
    }

    return $response;                                # return the reply
}

sub createSocket
{   my($self,$host,$port) = @_;

    $host ||= 'localhost';
    if (! $port) {
	$self->setErr(-1, "Error: port is undefined in 'socket' in '$PACK'" );
	return undef;

    } elsif (! defined $INC{'IO/Socket.pm'} ) {
	$self->setErr(-1, "'IO::Socket' module is not loaded in 'socket' method of '$PACK'");
	return undef;
    }

    my $socket = IO::Socket::INET->new( 'PeerAddr' => $host,
                                        'PeerPort' => $port,
                                        'Proto'    => 'tcp', );
    if (! $socket) {
	$self->setErr(-1, "Error: IO_Socket_INET failed in '$PACK': $!" );
	return undef;
    }

    $self->{_socket_} = $socket;
    return $socket;
}

sub getOpenSocket
{   my($self) = @_;
    return $self->{_socket_} || undef;
}

sub shutdownSocket
{   my($self, $socket) = @_;

    $socket ||= $self->getOpenSocket();

    $socket->shutdown(2);         # proper socket etiquette:
    $socket->close();             #  a shutdown then a close.

    return;
}

#-----------------------------------------------------------------------
# Somewhat useful for debugging ... replace with a better dumper.

sub dump {
    my($self)= @_;
    my($pack,$file,$line)=caller();
    my $text  = "DEBUG: ($PACK\:\:dump)\n  self='$self'\n";
       $text .= "CALLER $pack at line $line\n  ($file)\n";
    my $value;
    foreach my $param (sort keys %$self) {
	$value = $self->{$param};
	$value = $self->zeroStr( $value, "" );  # handles value of "0"
	$text .= " $param = $value\n";
    }
    $text .= "_" x 25 ."\n";

    if (ref($self->{header})) {
	$text .= "header:\n";
	$text .= $self->{header}->dump("nohead");
    }
    return($text);
}

sub zeroStr
{   my($self,$value,$undef) = @_;
    return $undef unless defined $value;
    return "0"    if (length($value) and ! $value);
    return $value;
}
#_________________________
1; # Required by require()

__END__