File Coverage

lib/Finance/Alpaca/DataStream.pm

Criterion	Covered	Total	%
statement	30	72	41.6
branch	0	18	0.0
condition			n/a
subroutine	10	15	66.6
pod	2	3	66.6
total	42	108	38.8

package Finance::Alpaca::DataStream 0.9902 {
    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::Bar qw[to_Bar Bar];
    use Finance::Alpaca::Struct::Trade qw[to_Trade Trade];
    use Finance::Alpaca::Struct::Quote qw[to_Quote Quote];
    #
    has source => ( is => 'ro', isa => Enum [ 'iex', 'sip' ], required => 1, default => 'iex' );
    has tx => ( is => 'rwp', isa => InstanceOf ['Mojo::Transaction::WebSocket'], predicate => 1 );
    has cb => ( is => 'ro',  isa => CodeRef, required => 1 );
    has subscriptions => (
        is  => 'rwp',
        isa => Dict [ bars => ArrayRef [Str], quotes => ArrayRef [Str], trades => ArrayRef [Str] ],
        default => sub { { bars => [], quotes => [], trades => [] } },
        lazy    => 1
    );

    sub authorize ( $s, $ua, $keys ) {
        $ua->websocket_p( 'wss://stream.data.alpaca.markets/v2/'
                . $s->source => { 'Sec-WebSocket-Extensions' => 'permessage-deflate' } )->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(
                    json => sub ( $tx, $msgs ) {
                        for my $msg (@$msgs) {

                            if ( $msg->{T} eq 'success' ) {
                                if ( $msg->{msg} eq 'connected' ) {    # Send auth
                                    $tx->send(
                                        {
                                            json => {
                                                action => 'auth',
                                                key    => $keys->[0],
                                                secret => $keys->[1]
                                            }
                                        }
                                    );
                                }
                                elsif ( $msg->{msg} eq 'authenticated' ) {
                                    $promise->resolve;
                                }
                            }
                            elsif ( $msg->{T} eq 'error' ) {
                                $s->cb->($msg);
                                if ( $msg->{code} eq 406 ) {    # Already connected; ignore

                                    # Send auth
                                    $tx->send(
                                        {
                                            json => {
                                                action => 'auth',
                                                key    => $s->keys->[0],
                                                secret => $s->keys->[1]
                                            }
                                        }
                                    );
                                }
                            }
                            elsif ( $msg->{T} eq 't' ) {
                                $s->cb->( to_Trade($msg) );
                            }
                            elsif ( $msg->{T} eq 'q' ) {
                                $s->cb->( to_Quote($msg) );
                            }
                            elsif ( $msg->{T} eq 'b' ) {
                                $s->cb->( to_Bar($msg) );
                            }
                            elsif ( $msg->{T} eq 'subscription' ) {
                                delete $msg->{T};
                                $s->_set_subscriptions($msg);
                            }
                            else {
                                #warn 'unknown data';
                                #...;
                                $s->cb->($msg);
                            }
                        }

                        #$tx->finish;
                    }
                );
                return $promise;
            }
        )->catch(
            sub ($err) {
                warn "WebSocket error: $err";
            }
        );
    }

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

    sub unsubscribe ( $s, %params ) {
        $s->tx->send( { json => { action => 'unsubscribe', %params } } );
    }
}
1;
__END__