File Coverage

blib/lib/AnyEvent/WebSocket/Client.pm

Criterion	Covered	Total	%
statement	78	98	79.5
branch	19	32	59.3
condition	6	9	66.6
subroutine	15	15	100.0
pod	1	1	100.0
total	119	155	76.7

package AnyEvent::WebSocket::Client;

use strict;
use warnings;
use Moo;
use AE;
use AnyEvent;
use AnyEvent::Handle;
use AnyEvent::Socket ();
use Protocol::WebSocket::Request;
use Protocol::WebSocket::Handshake::Client;
use AnyEvent::WebSocket::Connection;
use PerlX::Maybe qw( maybe provided );

# ABSTRACT: WebSocket client for AnyEvent
our $VERSION = '0.54'; # VERSION


has timeout => (
  is      => 'ro',
  default => sub { 30 },
);


has ssl_no_verify => (
  is => 'ro',
);


has ssl_ca_file => (
  is => 'ro',
);


has protocol_version => (
  is => 'ro',
);


has subprotocol => (
  is     => 'ro',
  coerce => sub { ref $_[0] ? $_[0] : [$_[0]] },
);


has http_headers => (
  is => 'ro',
  coerce => sub {
    ref $_[0] eq 'ARRAY' ? $_[0] : do {
      my $h = shift;
      [
        map {
          my($k,$v) = ($_, $h->{$_});
          $v = [$v] unless ref $v;
          map { $k => $_ } @$v;
          # sorted to make testing easier.
          # may be removed in the future
          # so do not depend on it.
        } sort keys %$h
      ],
    };
  },
);


has max_payload_size => (
  is => 'ro',
);


has max_fragments => (
  is => 'ro',
);


has env_proxy => (
  is => 'ro',
  default => sub { 0 },
);



sub connect
{
  my($self, $uri, $host, $port) = @_;
  unless(ref $uri)
  {
    require URI;
    $uri = URI->new($uri);
  }

  my $done = AE::cv;

  # TODO: should we also accept http and https URLs?
  # probably.
  if($uri->scheme ne 'ws' && $uri->scheme ne 'wss')
  {
    $done->croak("URI is not a websocket");
    return $done;
  }

  $host = $uri->host unless defined $host;
  $port = $uri->port unless defined $port;

  $self->_make_tcp_connection($uri->scheme, $host, $port, sub {
    my $fh = shift;
    unless($fh)
    {
      $done->croak("unable to connect");
      return;
    }
    my $req = Protocol::WebSocket::Request->new( maybe headers => $self->http_headers );
    my $handshake = Protocol::WebSocket::Handshake::Client->new(
            url     => $uri->as_string,
      maybe version => $self->protocol_version,
            req     => $req,
    );

    my %subprotocol;
    if($self->subprotocol)
    {
      %subprotocol = map { $_ => 1 } @{ $self->subprotocol };
      $handshake->req->subprotocol(join(',', @{ $self->subprotocol }));
    }

    my $hdl = AnyEvent::Handle->new(
                                                      fh       => $fh,
      provided $uri->secure,                          tls      => 'connect',
      provided $uri->secure && !$self->ssl_no_verify, peername => $uri->host,
      provided $uri->secure && !$self->ssl_no_verify, tls_ctx  => {
                                                              verify => 1,
                                                              verify_peername => "https",
                                                        maybe ca_file => $self->ssl_ca_file,
                                                      },
                                                      on_error => sub {
                                                        my ($hdl, $fatal, $msg) = @_;
                                                        if($fatal)
                                                        { $done->croak("connect error: " . $msg) }
                                                        else
                                                        { warn $msg }
                                                      },
    );

    $hdl->push_write($handshake->to_string);
    $hdl->on_read(sub {
      $handshake->parse($_[0]{rbuf});
      if($handshake->error)
      {
        $done->croak("handshake error: " . $handshake->error);
        undef $hdl;
        undef $handshake;
        undef $done;
      }
      elsif($handshake->is_done)
      {
        my $sb;
        if($self->subprotocol)
        {
          $sb = $handshake->res->subprotocol;
          if(defined $sb)
          {
            unless($subprotocol{$sb})
            {
              $done->croak("subprotocol mismatch, requested: @{[ join ', ', @{ $self->subprotocol } ]}, got: $sb");
            }
          }
          else
          {
            $done->croak("no subprotocol in response");
          }
        }
        undef $handshake;
        $done->send(
          AnyEvent::WebSocket::Connection->new(
                  handle               => $hdl,
                  masked               => 1,
            maybe subprotocol          => $sb,
            maybe max_payload_size     => $self->max_payload_size,
            maybe max_fragments        => $self->max_fragments,
          )
        );
        undef $hdl;
        undef $done;
      }
    });
  }, sub { $self->timeout });
  $done;
}

sub _make_tcp_connection
{
  my $self = shift;
  my $scheme = shift;
  my ($host, $port) = @_;
  if(!$self->env_proxy)
  {
    return &AnyEvent::Socket::tcp_connect(@_);
  }
  require AnyEvent::Connector;
  my @connectors =
      $scheme eq "ws"
      ? (map { AnyEvent::Connector->new(env_proxy => $_) } qw(ws http))
      : $scheme eq "wss"
      ? (map { AnyEvent::Connector->new(env_proxy => $_) } qw(wss https))
      : ();
  foreach my $connector (@connectors)
  {
    if(defined($connector->proxy_for($host, $port)))
    {
      return $connector->tcp_connect(@_);
    }
  }
  return &AnyEvent::Socket::tcp_connect(@_);
}

1;

__END__