view release on metacpan or  search on metacpan

lib/Finance/Bitcoin/Feed/Site/BtcChina/Socket.pm  view on Meta::CPAN

package Finance::Bitcoin::Feed::Site::BtcChina::Socket;

use JSON;
use Mojo::Base 'Mojo::Transaction::WebSocket';
use Scalar::Util qw(weaken);
has 'owner';
has 'ping_interval';
has 'ping_timeout';
has 'last_ping_at';
has 'last_pong_at';
has 'timer';

sub configure {
    my $self  = shift;
    my $owner = shift;
    $self->owner($owner);
    weaken($self->{owner});

    # call parse when receive text event
    $self->on(
        text => sub {
            my ($self, $message) = @_;
            $self->parse($message);
        });

    ################################################
    # setup events
    $self->on(
        subscribe => sub {
            my ($self, $channel) = @_;
            $self->on(
                'setup',
                sub {
                    $self->send({text => qq(42["subscribe","$channel"])});
                });
        });
    $self->emit('subscribe', 'marketdata_cnybtc');
    $self->emit('subscribe', 'marketdata_cnyltc');
    $self->emit('subscribe', 'marketdata_btcltc');

    #receive trade vent
    $self->on(
        trade => sub {
            my ($self, $data) = @_;
            $self->owner->emit(
                'data_out',
                $data->{date} * 1000,    # the unit of timestamp is ms
                uc($data->{market}),
                $data->{price});

        });

    $self->on(
        'ping',
        sub {
            $self->send({text => '2'});
        });

    # ping ping!
    my $timer = AnyEvent->timer(
        after    => 10,
        interval => 1,
        cb       => sub {
            if (time() - $self->last_ping_at > $self->ping_interval / 1000) {
                $self->emit('ping');
                $self->last_ping_at(time());
            }
        });
    $self->timer($timer);
    return;
}

#socket.io v2.2.2
sub parse {
    my ($self, $data) = @_;
    $self->owner->last_activity_at(time());
    return unless $data =~ /^\d+/;