File Coverage

blib/lib/AnyEvent/MtGox/Stream.pm

Criterion	Covered	Total	%
statement	33	90	36.6
branch	0	30	0.0
condition	0	17	0.0
subroutine	11	17	64.7
pod	1	1	100.0
total	45	155	29.0

package AnyEvent::MtGox::Stream;

use strict;
use warnings;

use AnyEvent;
use AnyEvent::HTTP qw(http_post);
use AnyEvent::Util qw(guard);
use Carp qw(croak);
use Errno qw(EPIPE);
use JSON ();
use Protocol::WebSocket::Frame;
use Protocol::WebSocket::Handshake::Client;
use URI;

our $VERSION = '0.02';
$VERSION = eval $VERSION;

sub new {
    my ($class, %params) = @_;

    my $secure        = $params{secure}        || 0;
    my $on_disconnect = $params{on_disconnect} || sub { croak 'Disconnected' };
    my $on_error      = $params{on_error}      || sub { croak @_ };
    my $on_message    = $params{on_message}    || sub { };

    my $server = 'socketio.mtgox.com';
    my $handle;

    # Socket.IO handshake.
    my $uri = URI->new("http://$server/socket.io/1");
    $uri->scheme('https') if $secure;
    AE::log debug => "Making Socket.IO handshake to $uri";
    http_post $uri, undef, sub {
        my ($body, $headers) = @_;
        return $on_error->('Socket.IO handshake failed')
            unless '200' eq $headers->{Status} and defined $body;
        my ($sid, $heartbeat) = split ':', $body, 3;
        AE::log debug => "Socket.IO handshake succeeded: $sid";

        my $timer;
        $handle = AnyEvent::Handle->new(
            connect  => [ $uri->host, $uri->port ],
            tls      => $secure ? 'connect' : undef,
            on_error => sub {
                my ($handle, $fatal, $msg) = @_;
                $handle->destroy;
                undef $timer;
                $!{EPIPE} == $! ? $on_disconnect->() : $on_error->($msg);
            },
        );

        # WebSocket handshake.
        $uri->path("/socket.io/1/websocket/$sid");
        $uri->scheme($secure ? 'wss' : 'ws');
        AE::log debug => "Making WebSocket handshake to $uri";
        my $wsh = Protocol::WebSocket::Handshake::Client->new(url => "$uri");
        $handle->push_write($wsh->to_string);

        $handle->push_read(sub {
            $wsh->parse($handle->rbuf)
                or return 1, $on_error->($wsh->error);
            if ($wsh->is_done) {
                AE::log debug => 'WebSocket handshake succeeded';
                return 1;
            }
        });

        my $frame = Protocol::WebSocket::Frame->new;
        my $send_message = sub {
            $handle->push_write($frame->new(@_)->to_bytes);
        };
        my $send_heartbeat = sub {
            AE::log debug => 'Sending heartbeat';
            $send_message->('2::');
        };

        $timer = AE::timer $heartbeat - 2, $heartbeat - 2, $send_heartbeat
            if $heartbeat;

        my $json = JSON->new;

        $handle->push_read(sub {
            $frame->append($handle->rbuf);
            while (defined(my $msg = $frame->next)) {
                my ($type, $id, $endpoint, $data) = split ':', $msg, 4;
                if ('4' eq $type and '/mtgox' eq $endpoint) {
                    if (my $scalar = eval { $json->decode($data) }) {
                        $on_message->($scalar);
                    }
                }
                # Respond to heartbeats only if a heartbeat timeout wasn't
                # given in the handshake.
                elsif ('2' eq $type and not $heartbeat) {
                    $send_heartbeat->();
                }
                elsif ('1::' eq $msg) { $send_message->('1::/mtgox') }
                elsif ('0' eq $type)  { return 1, $on_disconnect->() }
                elsif ('7' eq $type)  { return 1, $on_error->($data) }

                # Unhandled message types:
                #   3: message
                #   5: event
                #   6: ack
                #   8: noop
            }
        });
    };

    return unless defined wantarray;
    return guard { $handle->destroy };
}

1;

__END__