File Coverage

lib/Finance/Alpaca/TradeStream.pm

Criterion	Covered	Total	%
statement	54	69	78.2
branch	7	10	70.0
condition			n/a
subroutine	11	13	84.6
pod	0	3	0.0
total	72	95	75.7

package Finance::Alpaca::TradeStream 0.9904 {
    use strictures 2;
    use Moo;
    use feature 'signatures';
    no warnings 'experimental::signatures';
    use Types::Standard qw[ArrayRef CodeRef Dict Enum InstanceOf Str];
    use Mojo::Promise;
    #
    use lib './lib/';
    use Finance::Alpaca::Struct::TradeUpdate qw[to_TradeUpdate];
    #
    has tx => ( is => 'rwp', isa => InstanceOf ['Mojo::Transaction::WebSocket'], predicate => 1 );
    has cb => ( is => 'ro',  isa => CodeRef, required => 1 );
    has subscriptions => ( is => 'rwp', isa => ArrayRef [Str], default => sub { [] }, lazy => 1 );

    sub authorize ( $s, $ua, $keys, $paper ) {

        #use Data::Dump;
        #ddx $keys;
        $ua->inactivity_timeout(120);    # XXX - Testing!

        # warn(
        #     $paper ? 'wss://paper-api.alpaca.markets/stream' : 'wss://api.alpaca.markets/stream' );
        $ua->websocket_p(
            (
                $paper
                ? 'wss://paper-api.alpaca.markets/stream'
                : 'wss://api.alpaca.markets/stream'
            )
        )->then(
            sub ($tx) {
                my $promise = Mojo::Promise->new;
                $s->_set_tx($tx);

                #$tx->on( finish => sub { $promise->resolve } );
                # my $promise = Mojo::Promise->new;
                #$tx->on( finish => sub { $promise->resolve } );
                $tx->on(
                    finish => sub ( $tx, $code, $reason = '' ) {
                        warn "WebSocket closed with status $code. $reason";
                        $promise->resolve;
                    }
                );
                $tx->on( error => sub ( $e, $err ) { warn "This looks bad: $err" } );
                $tx->on(
                    json => sub ( $tx, $msg = () ) {
                        if ( $msg->{stream} eq 'authorization' ) {
                            if ( $msg->{data}{status} eq 'authorized' ) {
                                $s->subscribe( streams => ['trade_updates'] );
                                $promise->resolve();
                            }
                        }
                        elsif ( $msg->{stream} eq 'listening' ) {
                            $s->_set_subscriptions( $msg->{data}{streams} );
                        }
                        elsif ( $msg->{stream} eq 'trade_updates' ) {
                            $s->cb->( to_TradeUpdate( $msg->{data} ) );
                        }
                        else {
                            #warn 'unknown data';
                            #...;
                            $s->cb->($msg);
                        }

                        #$tx->finish;
                    }
                );
                $tx->send(
                    {
                        json => {
                            action => 'authenticate',
                            data   => { key_id => $keys->[0], secret_key => $keys->[1] }
                        }
                    }
                );
                return $promise;
            }
        )->catch(
            sub ($err) {
                warn "WebSocket error: $err";
            }
        );
    }

    sub subscribe ( $s, %params ) {
        $s->tx->send( { json => { action => 'listen', data => {%params} } } );
    }

    sub unsubscribe ( $s, %params ) {    # XXX - Grep current list
        $s->tx->send( { json => { action => 'listen', data => {%params} } } );
    }
}
1;
__END__