File Coverage

blib/lib/Net/WebSocket/Server.pm

Criterion	Covered	Total	%
statement	111	149	74.5
branch	24	56	42.8
condition	6	12	50.0
subroutine	19	26	73.0
pod	12	12	100.0
total	172	255	67.4

package Net::WebSocket::Server;

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

use Carp;
use IO::Socket::INET;
use IO::Select;
use Net::WebSocket::Server::Connection;
use Time::HiRes qw(time);
use List::Util qw(min);

our $VERSION = '0.003002';
$VERSION = eval $VERSION;

$SIG{PIPE} = 'IGNORE';

sub new {
  my $class = shift;

  my %params = @_;

  my $self = {
    listen         => 80,
    silence_max    => 20,
    tick_period    => 0,
    watch_readable => [],
    watch_writable => [],
    on_connect     => sub{},
    on_tick        => sub{},
    on_shutdown    => sub{},
  };

  while (my ($key, $value) = each %params ) {
    croak "Invalid $class parameter '$key'" unless exists $self->{$key};
    croak "$class parameter '$key' expected type is ".ref($self->{$key}) if ref $self->{$key} && ref $value ne ref $self->{$key};
    $self->{$key} = $value;
  }

  bless $self, $class;

  # send a ping every silence_max by checking whether data was received in the last silence_max/2
  $self->{silence_checkinterval} = $self->{silence_max} / 2;

  foreach my $watchtype (qw(readable writable)) {
    $self->{"select_$watchtype"} = IO::Select->new();
    my $key = "watch_$watchtype";
    croak "$class parameter '$key' expects an arrayref containing an even number of elements" unless @{$self->{$key}} % 2 == 0;
    my @watch = @{$self->{$key}};
    $self->{$key} = {};
    $self->_watch($watchtype, @watch);
  }

  return $self;
}

sub watch_readable {
  my $self = shift;
  croak "watch_readable expects an even number of arguments" unless @_ % 2 == 0;
  $self->_watch(readable => @_);
}

sub watched_readable {
  my $self = shift;
  return $self->{watch_readable}{$_[0]}{cb} if @_;
  return map {$_->{fh}, $_->{cb}} values %{$self->{watch_readable}};
}

sub watch_writable {
  my $self = shift;
  croak "watch_writable expects an even number of arguments" unless @_ % 2 == 0;
  $self->_watch(writable => @_);
}

sub watched_writable {
  my $self = shift;
  return $self->{watch_writable}{$_[0]}{cb} if @_;
  return map {$_->{fh}, $_->{cb}} values %{$self->{watch_writable}};
}

sub _watch {
  my $self = shift;
  my $watchtype = shift;
  croak "watch_$watchtype expects an even number of arguments after the type" unless @_ % 2 == 0;
  for (my $i = 0; $i < @_; $i+=2) {
    my ($fh, $cb) = ($_[$i], $_[$i+1]);
    croak "watch_$watchtype expects the second value of each pair to be a coderef, but element $i was not" unless ref $cb eq 'CODE';
    if ($self->{"watch_$watchtype"}{$fh}) {
      carp "watch_$watchtype was given a filehandle at index $i which is already being watched; ignoring!";
      next;
    }
    $self->{"select_$watchtype"}->add($fh);
    $self->{"watch_$watchtype"}{$fh} = {fh=>$fh, cb=>$cb};
  }
}

sub unwatch_readable {
  my $self = shift;
  $self->_unwatch(readable => @_);
}

sub unwatch_writable {
  my $self = shift;
  $self->_unwatch(writable => @_);
}

sub _unwatch {
  my $self = shift;
  my $watchtype = shift;
  foreach my $fh (@_) {
    $self->{"select_$watchtype"}->remove($fh);
    delete $self->{"watch_$watchtype"}{$fh};
  }
}

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;
  }
}

sub start {
  my $self = shift;

  if (ref $self->{listen}) {
    # if we got a server, make sure it's valid by clearing errors and checking errors anyway; if there's still an error, it's closed
    $self->{listen}->clearerr;
    croak "failed to start websocket server; the TCP server provided via 'listen' is invalid. (is the listening socket is closed? are you trying to reuse a server that has already shut down?)"
       if $self->{listen}->error;
  } else {
    # if we merely got a port, set up a reasonable default tcp server
    $self->{listen} = IO::Socket::INET->new(
      Listen    => 5,
      LocalPort => $self->{listen},
      Proto     => 'tcp',
      ReuseAddr => 1,
    ) || croak "failed to listen on port $self->{listen}: $!";
  }

  $self->{select_readable}->add($self->{listen});

  $self->{conns} = {};
  my $silence_nextcheck = $self->{silence_max} ? (time + $self->{silence_checkinterval}) : 0;
  my $tick_next = $self->{tick_period} ? (time + $self->{tick_period}) : 0;

  while (%{$self->{conns}} || $self->{listen}->opened) {
    my $silence_checktimeout = $self->{silence_max} ? ($silence_nextcheck - time) : undef;
    my $tick_timeout = $self->{tick_period} ? ($tick_next - time) : undef;
    my $timeout = min(grep {defined} ($silence_checktimeout, $tick_timeout));

    my ($ready_read, $ready_write, undef) = IO::Select->select($self->{select_readable}, $self->{select_writable}, undef, $timeout);
    foreach my $fh ($ready_read ? @$ready_read : ()) {
      if ($fh == $self->{listen}) {
        my $sock = $self->{listen}->accept;
        next unless $sock;
        my $conn = new Net::WebSocket::Server::Connection(socket => $sock, server => $self);
        $self->{conns}{$sock} = {conn=>$conn, lastrecv=>time};
        $self->{select_readable}->add($sock);
        $self->{on_connect}($self, $conn);
      } elsif ($self->{watch_readable}{$fh}) {
        $self->{watch_readable}{$fh}{cb}($self, $fh);
      } elsif ($self->{conns}{$fh}) {
        my $connmeta = $self->{conns}{$fh};
        $connmeta->{lastrecv} = time;
        $connmeta->{conn}->recv();
      } else {
        warn "filehandle $fh became readable, but no handler took responsibility for it; removing it";
        $self->{select_readable}->remove($fh);
      }
    }

    foreach my $fh ($ready_write ? @$ready_write : ()) {
      if ($self->{watch_writable}{$fh}) {
        $self->{watch_writable}{$fh}{cb}($self, $fh);
      } else {
        warn "filehandle $fh became writable, but no handler took responsibility for it; removing it";
        $self->{select_writable}->remove($fh);
      }
    }

    if ($self->{silence_max}) {
      my $now = time;
      if ($silence_nextcheck < $now) {
        my $lastcheck = $silence_nextcheck - $self->{silence_checkinterval};
        $_->{conn}->send('ping') for grep { $_->{lastrecv} < $lastcheck } values %{$self->{conns}};

        $silence_nextcheck = $now + $self->{silence_checkinterval};
      }
    }

    if ($self->{tick_period} && $tick_next < time) {
      $self->{on_tick}($self);
      $tick_next += $self->{tick_period};
    }
  }
}

sub connections { map {$_->{conn}} values %{$_[0]{conns}} }

sub shutdown {
  my ($self) = @_;
  $self->{on_shutdown}($self);
  $self->{select_readable}->remove($self->{listen});
  $self->{listen}->close();
  $_->disconnect(1001) for $self->connections;
}

sub disconnect {
  my ($self, $fh) = @_;
  $self->{select_readable}->remove($fh);
  $fh->close();
  delete $self->{conns}{$fh};
}

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

__END__