NAME

POE::Component::Server::AsyncEndpoint::ChannelAdapter

SYNOPSIS

shell$ endpointcreate

This will create the Endpoint subdirectory and files. Depending on the endpoint type, it should create a module similar to the one below. For simplicity, we show the single phase Outbound endpoint skeleton. For further information, generate one and read the fully commented skeleton code.

Typical single-phase Outbound Endpoint

package OB_SampleSP;

use POE;
use base POE::Component::Server::AsyncEndpoint::ChannelAdapter;
use POE::Component::Server::AsyncEndpoint::ChannelAdapter::Stomp;
use POE::Component::Server::AsyncEndpoint::ChannelAdapter::SOAP;
use POE::Component::Server::AsyncEndpoint::ChannelAdapter::Config;
use JSON; #optional

sub new {

    my $class = shift;

    # Init the new class with added Object States
    my $Alias = OB_SampleSP;
    my $self = $class->SUPER::new({
        Alias => $Alias,
        OStates =>
            [ 'pop_fifo', 'RCPT_OB_SampleSP' ],
    });

    # Init the Stomp Client
    my $stc = POE::Component::Server::AsyncEndpoint::ChannelAdapter::Stomp->spawn({
        Alias         => $Alias . '_STC',
        mq_address    => $self->{config}->mq_address,
        mq_port       => $self->{config}->mq_port,
        queue         => '/queue/ob_sample_sp',
        direction     => 'OB',
        user          => $self->{config}->stcuser,
        pass          => $self->{config}->stcpass,
        # For OB Endpoints you MUST specicfy the callback for STOMP RECEIPT
        rcpt_callback => {
            to_session => $Alias,
            to_handler => 'RCPT_OB_SampleSP',
        },

    });

    # Init the SOAP Client
    my $soc = POE::Component::Server::AsyncEndpoint::ChannelAdapter::SOAP->spawn({
        proxy   => $self->{config}->soap_proxy,
        service => $self->{config}->soap_service,
    });

    # Save the STOMP Client
    $self->{STC} = $stc;

    # Save the SOAP Client
    $self->{SOC} = $soc;

    # Init the popper
    POE::Kernel->post( 'OB_SampleSP', 'pop_fifo' );

    return $self;

}

# Sample overload of run
sub run {
    my $self = shift;
    $self->SUPER::run();
}

# Sample FIFO popper routine
sub pop_fifo {
    my ( $self, $kernel, $session ) = @_[ OBJECT, KERNEL, SESSION ];

    # reset the popper
    $kernel->delay_set('pop_fifo', $self->{config}->popper_freq);

    # not yet completely initialized
    return unless ($self->{ikc_stat} && $self->{STC}->{stc_stat});

    # return if a fifo record is being processed 
    # (see POE::Component::Server::AsyncEndpoint)
    return if defined($self->{fifo_id});

    my $soc = $self->{SOC};

    # pop the FIFO via SOAP (sample)
    my $call = $soc->popOBSampleOutboundFIFO(
        $self->{config}->socuser,
        $self->{config}->socpass
    );

    my $stc = $self->{STC};

    unless ( $call->fault ) {
        # Sample use of JSON Serialization
        # (why JSON? see POE::Component::Server::AsyncEndpoint)
        my $retval = from_jason( $call->result );

        # Sample data retrieval
        if ( my $fifo_id = $retval->{fifo_id} ) {

            $self->{fifo_id} = $fifo_id;
            $self->{data}  = from_json( $retval->{params} );

            # publish the data on the MQ, and ask for RECEIPT
            my $nframe = $stc->stomp->send({
                destination => $stc->config('Queue'),
                data        => to_json($self->{data}),
                receipt     => $fifo_id,
            });
            $kernel->call( $stc->config('Alias'), 'send_data', $nframe );

        }
        else {
            $self->{fifo_id} = undef;
            $self->{data}  = undef;
        }
    }
    else {
        my $logmsg = "FAILED pop_fifo (".
            $call->faultstring.")\n";
        $kernel->yield('logit', 'alert', $logmsg);
    }

}

# RECEIPT callback
sub RCPT_OB_SampleSP {

    my ($self, $kernel, $frame) = @_[OBJECT, KERNEL, ARG0];

    my $receipt_id = $frame->headers->{receipt};
    my $fifo_id    = $self->{fifo_id};

    # sanity
    if($receipt_id != $fifo_id){
        my $logmsg = "ENDPOINT PANIC! THE RECEIPT ID DOES NOT MATCH THE FIFO ID! BAILING OUT...!\n";
        $kernel->yield('logit', 'emergency', $logmsg);
        croak $logmsg;
    }

    my $soc = $self->{SOC};

    # Mark the FIFO record as 'popped' via SOAP
    my $call = $soc->poppedOBSampleSPFIFO(
        $self->{config}->socuser,
        $self->{config}->socpass,
        $fifo_id
    );

    my $popped_fail = 0;

    unless ( $call->fault ) {
        if ( $call->result eq 'OK' ) {
            $self->{fifo_id} = undef;
            $self->{params}  = undef;
        }
        else {
            my $logmsg = "The popped WS executed fine, ".
                "but it did NOT return OK (".$call->result.")\n";
            $kernel->yield('logit', 'alert', $logmsg);
            $popped_fail = 1;
        }

    }
    else {
        my $logmsg = "poppedOB_SampleSPFIFO FAILED for FIFO ID $fifo_id (".
            $call->faultstring.")\n";
        $kernel->yield('logit', 'alert', $logmsg);
        $popped_fail = 1;
    }

    # retry popped service if failure
    if($popped_fail){
        $kernel->delay_set('RCPT_OB_SampleSP', int($self->{config}->soc_retry), $frame);
    }


}

1;

DESCRIPTION

The ChannelAdapter class is the base of all Endpoints, so you should start your endpoint by using this class as base (as the example above suggests). If you use the helper scripts this will all be done for you automatically.

Methods

new

You should override this method and then call SUPER::new to initialize the Channel Adapter. The parameter OStates allows you to add object states to the main Channel Adapter session so you can post events and handle them in your object methods. Like any object methods these will call back to subroutines in your object of the same name.

my $self = $class->SUPER::new({
    Alias => $Alias,
    OStates =>
        [ 'pop_fifo', 'RCPT_OB_SampleSP' ],
});

sub pop_fifo {

}

sub RCPT_OB_SampleSP {


}

In this example we are declaring two object methods that will be called when an event is posted to the session identified by $Alias. The new method is also a good place to initialize the clients for the different protocols your Endpoint is going to support. For example to initialize a STOMP Client for an Outbound Endpoint you would do something like:

my $stc = POE::Component::Server::AsyncEndpoint::ChannelAdapter::Stomp->spawn({
    Alias         => $Alias . '_STC',
    mq_address    => $self->{config}->mq_address,
    mq_port       => $self->{config}->mq_port,
    queue         => '/queue/ob_sample_sp',
    direction     => 'OB',
    user          => $self->{config}->stcuser,
    pass          => $self->{config}->stcpass,
    rcpt_callback => {
        to_session => $Alias,
        to_handler => 'RCPT_OB_SampleSP',
    },

});

Here we initialize a STOMP client and since it's an Outbound Endpoint we are obligated to specify the session and handler that the STOMP client will call when it has had a RECEIPT from the MQ. As you can see, we use one of the object states declared in SUPER::new above. This means that when the STOMP RECEIPT arrives, the client will post this to the session in $Alias (which is the main Channel Adapter session) and it, in turn, will call our local object method (local subroutine). For more information on how this is actually done see POE and POE::Component::Server::AsyncEndpoint::ChannelAdapter::Stomp

run

If you don't override this method, it will just call the Channel Adapter's CA run method which at the moment just runs the POE Kernel.

EXAMPLES

Just use the endpointcreate helper script, follow the prompts and look inside the generated code which is fully documented.

SEE ALSO

POE::Component::Server::AsyncEndpoint::ChannelAdapter::SOAP POE::Component::Server::AsyncEndpoint::ChannelAdapter::Stomp POE::Component::Server::AsyncEndpoint::ChannelAdapter::Config

POE::Component::Server::AsyncEndpoint POE

AUTHOR

Alejandro Imass <ait@p2ee.org> Alejandro Imass <aimass@corcaribe.com>

COPYRIGHT AND LICENSE

Copyright (C) 2008 by Alejandro Imass / Corcaribe Tecnología C.A. for the P2EE Project

This library is free software; you can redistribute it and/or modify it under the same terms as Perl itself, either Perl version 5.8.8 or, at your option, any later version of Perl 5 you may have available.