File Coverage

blib/lib/Terse/WebSocket.pm

Criterion	Covered	Total	%
statement	9	83	10.8
branch	0	32	0.0
condition	0	10	0.0
subroutine	3	7	42.8
pod	1	4	25.0
total	13	136	9.5

package Terse::WebSocket;

use base 'Terse';
use MIME::Base64;
use Protocol::WebSocket::Handshake::Server;

sub new {
	my ($class, $t) = @_;
	my $self = $class->SUPER::new();
	my $version = '';
	my $env =  $t->request->env;
	$self->psgix = $env->{'psgix.io'};
	$self->handshake = Protocol::WebSocket::Handshake::Server->new_from_psgi($t->request->env);
	$self->handshake->parse();
	return $self;
}

sub start {
	my ($self, $t, $cbs, $responder) = @_;
	my $writer = eval { $responder->([101, $self->handshake->res->headers]); };
	$cbs->{($@ ? 'error' : 'connect')}->($self, $responder, $@);
	my $reset_rate = $t->websocket_reset_rate ||= 100000;
	eval {
		my $ping_rate = $reset_rate;
		while (1) {
			$ping_rate--;
			my $response;
			if ($ping_rate < 0) {
				$ping_rate = $reset_rate;
				syswrite $self->psgix, $self->handshake->build_frame( type => 'ping' )->to_bytes;
				$response = $self->recieve() while($ping_rate-- > 0 && !$response);
				if (!$response || $response ne 'pong') {
					last;
				}
				$ping_rate = $reset_rate;
			}
			$response = $self->recieve();
			if ($response) {
				if ($response =~ m/^invalid_(length|version|host|required_key)$/) {
					$cbs->{error}->($self, $response, $responder);
					last;
				} else {
					$ping_rate = $reset_rate;
					$cbs->{recieve}->($self, $response, $responder);
				}
			}
		}
	};
	$cbs->{error}->($self, $responder, $@) if ($@);
	$cbs->{disconnect}->($self, $responder) if $cbs->{disconnect};
	delete $t->{_application}->websockets->{$t->sid->value} if $cbs->{close_delete};
	$responder->([200, []]);
}

sub send {
	my ($self, $message) = @_;
	my $frame = $self->handshake->build_frame;
	$frame->append($message);
	my $pg = $self->psgix;
	syswrite $pg, $frame->to_bytes;
	return $self;
}

sub recieve {
	my ($self, @ENCODED) = @_;
	my $length;
	if (! scalar @ENCODED) {
		return shift @{ $self->next_frame } if scalar @{ $self->next_frame ||= [] };
		my $pg = $self->psgix;
		my $content = "";
		$length = sysread($pg, $content, 8192);
		return unless $length;
		$length = sysread($pg, $content, 8192, length($content)) while $length >= 8192;
		@ENCODED = map { unpack "C", $_ } split //, $content;
	}
	my @bits = split //, sprintf("%b\n", $ENCODED[0]);
        $self->fin = $bits[0];
        $self->rsv = [@bits[1 .. 3]];
	$self->op = shift @ENCODED;
	if ($ENCODED[0] == 254) {
		my @length = splice @ENCODED, 0, 3;
		$length = ((($length[0] + 2) * $length[1]) + $length[2]);
	} else {
		$length = shift @ENCODED;
		$length -= 128;
	}
	return pack "C*", join("", @ENCODED) if (scalar @ENCODED == $length);
	my @MASK = splice @ENCODED, 0, 4;
	if (scalar @ENCODED > $length) {
		my $next = $self->recieve(splice @ENCODED, $length, scalar @ENCODED);
		return $next if ($next eq 'invalid_length');
		unshift @{ $self->next_frame }, $next;
	}
	return 'invalid_length' if (scalar @ENCODED != $length);
	return join "", map { pack "C", ($ENCODED[$_] ^ $MASK[$_ % 4]) } 0 .. $#ENCODED;
}

1;

__END__;