File Coverage

blib/lib/POE/Component/Client/opentick.pm

Criterion	Covered	Total	%
statement	136	203	67.0
branch	23	56	41.0
condition	2	9	22.2
subroutine	30	50	60.0
pod	15	15	100.0
total	206	333	61.8

package POE::Component::Client::opentick;
#
#   opentick.com POE client
#
#   infi/2008
#
#   $Id: opentick.pm 56 2009-01-08 16:51:14Z infidel $
#
#   Full POD documentation after __END__
#

use strict;
use warnings;
use Socket;
use Carp qw( croak );
use Data::Dumper;
use POE qw( Wheel::SocketFactory Wheel::ReadWrite
            Driver::SysRW        Filter::Stream   );

# Ours
use POE::Component::Client::opentick::Constants;
use POE::Component::Client::opentick::Util;
use POE::Component::Client::opentick::Output;
use POE::Component::Client::opentick::Error;
use POE::Component::Client::opentick::Protocol;
use POE::Component::Client::opentick::Socket;

###
### Variables
###

use vars qw( $VERSION $TRUE $FALSE $KEEP $DELETE $poe_kernel );

$VERSION = '0.21';
*TRUE    = \1;
*FALSE   = \0;
*KEEP    = \0;
*DELETE  = \1;

# These arguments are for this object; pass the rest on.
my %our_args = (
    autologin       => $KEEP,
    alias           => $KEEP,
    events          => $DELETE,
    notifyee        => $DELETE,
    realtime        => $KEEP,
    debug           => $KEEP,
    quiet           => $KEEP,
);

########################################################################
###   Public methods                                                 ###
########################################################################

# Create the object and POE session instance
sub spawn
{
    my( $class, @args ) = @_;
    croak( "$class requires an even number of parameters" ) if( @args & 1 );

    # Set our default variables
    my $self = {
            READY       => OTConstant( 'OT_STATUS_INACTIVE' ),
                                        # ready to accept requests?
            debug       => $FALSE,                      # in debug mode
            quiet       => $FALSE,                      # in silent mode
            alias       => OTDefault( 'alias' ),        # our POE alias
            realtime    => OTDefault( 'realtime' ),     # RealTime quote mode
            autologin   => OTDefault( 'autologin' ),    # Auto login?
            # event callbacks
            # events => { event_id => { $sess_id => 1, ... }, ... }
            events      => {},          # event notification map
            notifyee    => undef,
            # object containers
            sock        => undef,       # Socket object
            protocol    => undef,       # Protocol object
            # Statistical information
            start_time  => time,
    };

    # Set up event hashref with all event names
    $self->{events}->{$_} = {} for( OTEventList() );

    # Create and init the object
    bless( $self, $class );
    my @leftovers = $self->initialize( @args );

    # Create the protocol handling object
    $self->{protocol} =
        POE::Component::Client::opentick::Protocol->new( @leftovers );

    # Create the socket handling object
    $self->{sock} =
        POE::Component::Client::opentick::Socket->new( @leftovers );

    # GO!
    $self->_POE_startup();

    return( $self );
}

# Initialize this object instance
sub initialize
{
    my( $self, %args ) = @_;

    # Make sure we have mandatory arguments.  I don't really like this.
    my ($key) = grep { /^notifyee$/i } keys( %args )
        or croak( 'Notifyee is a mandatory argument' );
    my $notifyee = delete( $args{$key} );
    grep { /^events$/i } keys( %args )
        or croak( 'Events is a mandatory argument' );

    # Stash our args...
    for( keys( %args ) )
    {
        if( $_ =~ /^events$/i )
        {
            croak( "Events must be an arrayref" )
                unless( ref( $args{$_} ) eq 'ARRAY' );
            $self->_reg_event( $notifyee, $args{$_} );
            delete( $args{ $_ } );
        }
        else
        {
            # grab our args
            if( exists( $our_args{lc $_} ) )
            {
                $self->{lc $_} = $args{$_};
                # delete them if appropriate
                delete( $args{ $_ } ) if( $our_args{lc $_} == $DELETE );
            }
        }
    }

    # Set the debug output flag.
    POE::Component::Client::opentick::Output->set_debug( $TRUE )
        if( $self->{debug} );
    POE::Component::Client::opentick::Output->set_quiet( $TRUE )
        if( $self->{quiet} );

    # ... and return the rest.
    return( %args );
}

sub new
{
    croak( 'Please use spawn() to create a session.' );
}

# Shut down the OT connection and POE session
sub shutdown
{
    my( $self ) = @_;

    $poe_kernel->post( $self->{alias}, 'shutdown' );

    return;
}

sub login
{
    my( $self ) = @_;

    $poe_kernel->call( $self->{alias}, '_ot_proto_issue_command',
                       OTConstant( 'OT_LOGIN' ) );

    return;
}

sub logout
{
    my( $self ) = @_;

    $poe_kernel->call( $self->{alias}, '_ot_proto_issue_command',
                       OTConstant( 'OT_LOGOUT' ) );

    return;
}

# Send an event to OT via object method
sub yield
{
    my $self = shift;

    $poe_kernel->post( $self->{alias} => @_ );

    return;
}

# Call a synchronous event in OT via object method
sub call
{
    my $self = shift;

    return( $poe_kernel->call( $self->{alias} => @_ ) );
}

# Are we ready for action?
sub ready
{
    my $self = shift;

    return( $self->{READY} == OTConstant( 'OT_STATUS_LOGGED_IN' )
            ? $TRUE
            : $FALSE );
}

# The next 3 functions are lame, added to be compatible with the otFeed API
sub set_hosts
{
    my( $self, @hosts ) = @_;
    @hosts = @{ $hosts[0] } if( ref( $hosts[0] ) eq 'ARRAY' );

    $self->{'socket'}->_set_servers( \@hosts );

    return;
}

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

    $self->{'socket'}->_set_port( $port );

    return;
}

sub set_platform_id
{
    my( $self, $platform_id, $platform_pass ) = @_;

    $self->{protocol}->{state_obj}->_set_platform_id( $platform_id );
    $self->{protocol}->{state_obj}->_set_platform_pass( $platform_pass );

    return;
}

# return our actual state
sub get_status
{
    my( $self ) = @_;

    return( $self->{READY} );
}

# Return some statistics
# HYBRID METHOD/POE EVENT HANDLER
sub statistics
{
    my( $self ) = shift;

    my @fields = (
        $self->{sock}->get_packets_sent(),
        $self->{sock}->get_packets_recv(),
        $self->{sock}->get_bytes_sent(),
        $self->{sock}->get_bytes_recv(),
        $self->{protocol}->get_messages_sent(),
        $self->{protocol}->get_messages_recv(),
        $self->{protocol}->get_records_recv(),
        $self->{protocol}->get_errors_recv(),
        $self->get_uptime(),
        $self->{sock}->get_connect_time(),
    );

    return( @fields );
}

# Admit our age
sub get_uptime
{
    my( $self ) = shift;

    return( time - $self->{start_time} );
}

########################################################################
###   POE State handlers                                             ###
########################################################################

# Called at session start
# First callback upon POE start of this session.
sub _ot_start
{
    my( $self, $kernel ) = @_[OBJECT, KERNEL];

    O_DEBUG( sprintf 'Starting POE session (ID=%s)',
             $kernel->get_active_session()->ID() );

    $kernel->alias_set( $self->{alias} );
    $kernel->yield( 'connect' ) if( $self->_auto_login() );

    return;
}

# Called at session shutdown
# Final callback before shutdown
sub _ot_stop
{
    my( $self, $kernel ) = @_[OBJECT, KERNEL];

    O_DEBUG( "Final shutdown called.  Bye!" );

    return;
}

# Called on receipt of 'register' event
# Register a client for particular events
sub _register
{
    my( $self, $kernel, $sender, $events ) = @_[OBJECT, KERNEL, SENDER, ARG0];

    return( $self->_reg_event( $sender->ID(), $events ) );
}

# Called on receipt of 'unregister' event
# Unregister a client for particular events
sub _unregister
{
    my( $self, $kernel, $sender, $events ) = @_[OBJECT, KERNEL, SENDER, ARG0];

    return( $self->_unreg_event( $sender->ID(), $events ) );
}

# Maximum Reconnect Attempts reached.  Complain to someone.
sub _reconn_giveup
{
    my( $self, $kernel ) = @_[OBJECT, KERNEL];

    O_WARN( "Connection retry limit reached." );
    $kernel->yield( '_notify_of_event', OTEvent( 'OT_CONNECT_FAILED' ) );

    return;
}

# Server sent us a redirect request in OT_LOGIN response packet
sub _server_redirect
{
    my( $self, $host, $port ) = @_[OBJECT, ARG0, ARG1];

    O_NOTICE( "Server redirected us to $host:$port." );
    $self->{sock}->redirect( $host, $port );

    return;
}

# Logged in; set up heartbeat and say we're ready to go!
sub _ot_on_login
{
    my( $self, $kernel ) = @_[OBJECT, KERNEL];

    # Remove the connection timeout alarm.
    $kernel->alarm_remove( delete( $self->{sock}->{timeout_id} ) )
        if( $self->{sock}->{timeout_id} );

    $self->{READY} = OTConstant( 'OT_STATUS_LOGGED_IN' );
    $self->{sock}->_set_state( OTConstant( 'OT_STATUS_LOGGED_IN' ) );
    $kernel->yield( '_ot_proto_heartbeat_send' );

    return;
}

# Logged out; stop heartbeat and disable ready flag
sub _ot_on_logout
{
    my( $self, $kernel ) = @_[OBJECT, KERNEL];

    O_DEBUG( "We are logged out" );

    $kernel->yield( '_ot_proto_heartbeat_stop' );
    $self->{READY} = OTConstant( 'OT_STATUS_INACTIVE' );
    $self->{sock}->_set_state( OTConstant( 'OT_STATUS_INACTIVE' ) );

    return;
}

sub _status_changed
{
    my( $self, $status ) = @_[OBJECT, ARG0];

    return( $self->{READY} = $status );
}

# Pass on event notifications to their recipients
sub _notify_of_event
{
    my( $self, $kernel, $event_type, $extra_recips, @args )
                                            = @_[OBJECT,KERNEL,ARG0..$#_];
    # Resolve event properly.
    my $event = ( $event_type =~ /^\d+$/ )
                ? OTEvent( $event_type )
                : $event_type;
    my( $notify_count, %seen );

    # Prepend our extra recipients to the list, but don't send two events.
    no warnings 'uninitialized';
    my @recipients = grep { not $seen{$_}++ }
        ( @$extra_recips, keys( %{$self->{events}->{$event}} ) );

    # Send!
    for my $recipient ( @recipients )
    {
        $poe_kernel->post( $recipient, $event, @args );
        $notify_count++;
    }

    return( $notify_count );
}

### API event receiver/dispatcher
sub _api_dispatch
{
    my( $self, $kernel, $event, $sender, @args )
                        = @_[OBJECT, KERNEL, STATE, SENDER, ARG0..$#_];

    O_DEBUG( "_api_dispatch( $event ) from sender: " . $sender->ID() );

    # Find the command number, and report on irregularities.
    my ($cmd_number, $deprecated) = OTAPItoCommand( $event );
    O_WARN( "$event is deprecated by opentick; please use " .
                        OTCommandtoAPI( $deprecated ) . " instead." )
        if( $deprecated );
    O_ERROR( "No known command mapping for $event." )
        unless( $cmd_number );

    # Dispatch the command
    my $retval = $kernel->call( $self->{alias}, '_ot_proto_issue_command',
                                $cmd_number,    @args )
        if( $cmd_number );

    return( $retval );
}

# Logout event trap.
sub _logged_out
{
    my( $self, $kernel ) = @_[OBJECT, KERNEL];

    $self->{sock}->_reset_object() if( $self and $self->{sock} );

    $self->yield( _notify_of_event => OTEvent( 'OT_ON_LOGOUT' ) );

    $self->_final_cleanup() if( $self->_is_disconnecting() );

    return;
}

# We got some unknown event.
# XXX: Perhaps we should send this back as an ot_on_error event.
sub _unknown_event
{
    my( $self, $event ) = @_[OBJECT, ARG0];

    O_DEBUG( "Unhandled event '$event'" );

    return;
}

# Do nothing, for useless events
sub _do_nothing {}

########################################################################
###   Private methods                                                ###
########################################################################

# Start me up.
sub _POE_startup
{
    my( $self ) = @_;

    POE::Session->create(
            object_states => [
                # General events for the entire interface
                $self => {
                    _start                  => '_ot_start',
                    _stop                   => '_ot_stop',
                    _default                => '_unknown_event',
                    _server_redirect        => '_server_redirect',
                    _reconn_giveup          => '_reconn_giveup',
                    _notify_of_event        => '_notify_of_event',
                    _logged_out             => '_logged_out',
                    # public sendable events
                    shutdown                => '_POE_shutdown',
                    register                => '_register',
                    unregister              => '_unregister',
                    statistics              => 'statistics',
                    # public receivable events that we need to handle, too
                    OTEvent('OT_ON_LOGIN')  => '_ot_on_login',
                    OTEvent('OT_ON_DATA')   => '_do_nothing',
                    OTEvent('OT_ON_LOGOUT') => '_do_nothing',
                    OTEvent('OT_ON_ERROR')  => '_do_nothing',
                    OTEvent('OT_REQUEST_COMPLETE')  => '_do_nothing',
                    OTEvent('OT_REQUEST_CANCELLED') => '_do_nothing',
                    OTEvent('OT_STATUS_CHANGED') => '_status_changed',
                    # API commands
                    requestSplits               => '_api_dispatch',
                    requestDividends            => '_api_dispatch',
                    requestOptionInit           => '_api_dispatch',
                    requestHistData             => '_api_dispatch',
                    requestHistTicks            => '_api_dispatch',
                    requestTickStream           => '_api_dispatch',
                    requestTickStreamEx         => '_api_dispatch',
                    requestTickSnapshot         => '_api_dispatch',
                    requestOptionChain          => '_api_dispatch',
                    requestOptionChainEx        => '_api_dispatch',
                    requestOptionChainU         => '_api_dispatch',
                    requestOptionChainSnapshot  => '_api_dispatch',
                    requestEqInit               => '_api_dispatch',
                    requestEquityInit           => '_api_dispatch', # alias
                    requestBookStream           => '_api_dispatch',
                    requestBookStreamEx         => '_api_dispatch',
                    requestHistBooks            => '_api_dispatch',
                    requestListSymbols          => '_api_dispatch',
                    requestListSymbolsEx        => '_api_dispatch',
                    requestListExchanges        => '_api_dispatch',
                    cancelTickStream            => '_api_dispatch',
                    cancelBookStream            => '_api_dispatch',
                    cancelHistData              => '_api_dispatch',
                    cancelOptionChain           => '_api_dispatch',
                },
                $self->{sock} => [
                    # Socket events
                    qw(
                        connect
                        disconnect
                        reconnect
                        _redirect
                        _ot_sock_connected
                        _ot_sock_connfail
                        _ot_sock_conntimeout
                        _ot_sock_error
                        _ot_sock_receive_packet
                        _ot_sock_send_packet
                    ),
                ],
                $self->{protocol} => [
                    # Protocol events
                    qw(
                        logout
                        login
                        _ot_proto_issue_command
                        _ot_proto_process_response
                        _ot_proto_end_of_data
                        _ot_proto_heartbeat_send
                        _ot_proto_heartbeat_stop
                    ),
                ],
                $self->{protocol}->{state_obj} => [
                    # Individual protocol message type events
                    qw(
                        _ot_msg_login_o
                        _ot_msg_generic_o
                        _ot_msg_nobody_o
                        _ot_msg_login_i
                        _ot_msg_logout_i
                        _ot_msg_single_i
                        _ot_msg_singledt_i
                        _ot_msg_multi_i
                        _ot_msg_multidt_i
                        _ot_msg_listex_i
                        _ot_msg_cancel_i
                        _ot_msg_nobody_i
                    ),
                ],
            ],
            heap => $self,
    );

    return;
}

# Shut the client down gracefully.
sub _POE_shutdown
{
    my( $self, $kernel ) = @_[OBJECT, KERNEL];

    $self->_is_disconnecting( $TRUE );
    $self->{sock}->disconnect();

    return;
}

# Clean up everything and die already, already.
sub _final_cleanup
{
    my( $self ) = @_;

    delete( $self->{sock} );
    delete( $self->{protocol} );
    $poe_kernel->alarm_remove_all();
    $poe_kernel->alias_remove( $self->{alias} );
    undef( $self );

    return;
}

# Register an event handler message to be sent to a session from opentick
# $events = \@aryref
sub _reg_event
{
    my( $self, $sender_id, $events ) = @_;

    my $regged = 0;
    if( $sender_id && ref( $events ) eq 'ARRAY' )
    {
        $events = [ OTEventList() ] if( grep { /^all$/i } @$events );
        for( @$events )
        {
            if( OTEventByEvent( $_ ) )
            {
                $self->{events}->{$_}->{$sender_id} = $TRUE;
                $regged++;
            }
        }
    }

    return( $regged );
}

# Register an event handler message to be sent to a session from opentick
# $events = \@aryref
sub _unreg_event
{
    my( $self, $sender_id, $events ) = @_;

    my $unregged = 0;
    if( $sender_id && ref( $events ) eq 'ARRAY' )
    {
        $events = [ OTEventList() ] if( grep { /^all$/i } @$events );
        for( @$events )
        {
            next unless( OTEventByEvent( $_ ) );
            $unregged += delete( $self->{events}->{$_}->{$sender_id} );
        }
    }

    return( $unregged );
}

#######################################################################
###  Accessor methods                                               ###
#######################################################################

sub _auto_login
{
    my( $self ) = @_;

    return( $self->{autologin} ? $TRUE : $FALSE );
}

sub _is_disconnecting
{
    my( $self, $value ) = @_;

    $self->{is_disconnecting} = $value ? $TRUE : $FALSE
        if( defined( $value ) );

    return( $self->{is_disconnecting} );
}

1;

__END__