File Coverage

blib/lib/Net/WebSocket/Server/Connection.pm

Criterion	Covered	Total	%
statement	97	129	75.1
branch	30	68	44.1
condition	6	15	40.0
subroutine	17	31	54.8
pod	15	15	100.0
total	165	258	63.9

package Net::WebSocket::Server::Connection;

use 5.006;
use strict;
use warnings FATAL => 'all';

use Carp;
use Protocol::WebSocket::Handshake::Server;
use Protocol::WebSocket::Frame;
use Socket qw(IPPROTO_TCP TCP_NODELAY);
use Encode;

sub new {
  my $class = shift;

  my %params = @_;

  my $self = {
    socket        => undef,
    server        => undef,
    nodelay       => 1,
    max_send_size => eval { Protocol::WebSocket::Frame->new->{max_payload_size} } || 65536,
    max_recv_size => eval { Protocol::WebSocket::Frame->new->{max_payload_size} } || 65536,
    on_handshake  => sub{},
    on_ready      => sub{},
    on_disconnect => sub{},
    on_utf8       => sub{},
    on_pong       => sub{},
    on_binary     => sub{},
  };

  while (my ($key, $value) = each %params ) {
    croak "Invalid $class parameter '$key'" unless exists $self->{$key};
    croak "$class parameter '$key' expects a coderef" if ref $self->{$key} eq 'CODE' && ref $value ne 'CODE';
    $self->{$key} = $value;
  }

  croak "$class construction requires '$_'" for grep { !defined $self->{$_} } qw(socket server);

  $self->{handshake} = new Protocol::WebSocket::Handshake::Server();
  $self->{disconnecting} = 0;
  $self->{ip} = $self->{socket}->peerhost;
  $self->{port} = $self->{socket}->peerport;

  # only attempt to start SSL if this is an IO::Socket::SSL-like socket that also has not completed its SSL handshake (SSL_startHandshake => 0)
  $self->{needs_ssl} = 1 if $self->{socket}->can("accept_SSL") && !$self->{socket}->opened;

  bless $self, $class;
}

sub on {
  my $self = shift;
  my %params = @_;

  while (my ($key, $value) = each %params ) {
    croak "Invalid event '$key'" unless exists $self->{"on_$key"};
    croak "Expected a coderef for event '$key'" unless ref $value eq 'CODE';
    $self->{"on_$key"} = $value;
  }
}


### accessors

sub server { $_[0]->{server} }

sub socket { $_[0]->{socket} }

sub is_ready { !$_[0]->{handshake} }

sub ip { $_[0]{ip} }

sub port { $_[0]{port} }

sub nodelay {
  my $self = shift;
  if (@_) {
    $self->{nodelay} = $_[0];
    setsockopt($self->{socket}, IPPROTO_TCP, TCP_NODELAY, $self->{nodelay} ? 1 : 0) unless $self->{handshake};
  }
  return $self->{nodelay};
}

sub max_send_size {
  my $self = shift;
  $self->{max_send_size} = $_[0] if @_;
  return $self->{max_send_size};
}

sub max_recv_size {
  my $self = shift;
  if (@_) {
    croak "Cannot change max_recv_size; handshake is already complete" if $self->{parser};
    $self->{max_recv_size} = $_[0];
  }
  return $self->{max_recv_size};
}


### methods

sub disconnect {
  my ($self, $code, $reason) = @_;
  return if $self->{disconnecting};
  $self->{disconnecting} = 1;

  $self->_event('on_disconnect', $code, $reason);

  my $data = '';
  if (defined $code || defined $reason) {
    $code ||= 1000;
    $reason = '' unless defined $reason;
    $data = pack("na*", $code, $reason);
  }
  $self->send(close => $data) unless $self->{handshake};

  $self->{server}->disconnect($self->{socket});
}

sub send_binary {
  $_[0]->send(binary => $_[1]);
}

sub send_utf8 {
  $_[0]->send(text => Encode::encode('UTF-8', $_[1]));
}

sub send {
  my ($self, $type, $data) = @_;

  if ($self->{handshake}) {
    carp "tried to send data before finishing handshake";
    return 0;
  }

  my $frame = new Protocol::WebSocket::Frame(type => $type, max_payload_size => $self->{max_send_size});
  $frame->append($data) if defined $data;

  my $bytes = eval { $frame->to_bytes };
  if (!defined $bytes) {
    carp "error while building message: $@" if $@;
    return;
  }

  syswrite($self->{socket}, $bytes);
}

sub recv {
  my ($self) = @_;

  if ($self->{needs_ssl}) {
    my $ssl_done = $self->{socket}->accept_SSL;
    if ($self->{socket}->errstr) {
      $self->disconnect;
      return;
    }
    return unless $ssl_done;
    $self->{needs_ssl} = 0;
  }

  my ($len, $data) = (0, "");
  if (!($len = sysread($self->{socket}, $data, 8192))) {
    $self->disconnect();
    return;
  }

  # read remaining data
  $len = sysread($self->{socket}, $data, 8192, length($data)) while $len >= 8192;

  if ($self->{handshake}) {
    $self->{handshake}->parse($data);
    if ($self->{handshake}->error) {
      $self->disconnect(1002);
    } elsif ($self->{handshake}->is_done) {
      $self->_event(on_handshake => $self->{handshake});
      return unless do { local $SIG{__WARN__} = sub{}; $self->{socket}->connected };

      syswrite($self->{socket}, $self->{handshake}->to_string);
      delete $self->{handshake};

      $self->{parser} = new Protocol::WebSocket::Frame(max_payload_size => $self->{max_recv_size});
      setsockopt($self->{socket}, IPPROTO_TCP, TCP_NODELAY, 1) if $self->{nodelay};
      $self->_event('on_ready');
    }
    return;
  }

  $self->{parser}->append($data);

  my $bytes;
  while (defined ($bytes = eval { $self->{parser}->next_bytes })) {
    if ($self->{parser}->is_binary) {
      $self->_event(on_binary => $bytes);
    } elsif ($self->{parser}->is_text) {
      $self->_event(on_utf8 => Encode::decode('UTF-8', $bytes));
    } elsif ($self->{parser}->is_pong) {
      $self->_event(on_pong => $bytes);
    } elsif ($self->{parser}->is_close) {
      $self->disconnect(length $bytes ? unpack("na*",$bytes) : ());
      return;
    }
  }

  if ($@) {
    $self->disconnect(1002);
    return;
  }
}

### internal methods

sub _event {
  my ($self, $event, @args) = @_;
  $self->{$event}($self, @args);
}

1; # End of Net::WebSocket::Server

__END__