/ 
Net-WebSocket-EVx-0.21
/ Net::WebSocket::EVx

NAME

Net::WebSocket::EVx - Perl wrapper around Wslay websocket library

DESCRIPTION

Net::WebSocket::EVx is a websocket module based on EV and Alien::Wslay. It is a fork of Net::WebSocket::EV, which looks abandoned; the main differences are the use of Alien::Wslay and RSV bit support (for compressed transfers, for example).

The module does not perform the HTTP handshake. Complete the upgrade yourself, then hand the socket to new().

SYNOPSIS

use EV;
use Net::WebSocket::EVx;

# $fh is a non-blocking socket whose websocket handshake is already done
my $ws; $ws = Net::WebSocket::EVx->new({
    fh          => $fh,
    rsv         => WS_RSV_NONE,   # no extension negotiated
    on_msg_recv => sub {
        my ($rsv, $opcode, $msg, $status_code) = @_;
        $ws->queue_msg($msg);     # echo it back
    },
    on_close    => sub { my ($code) = @_; undef $ws; close $fh },
});

EV::run;

A complete PSGI echo server with compression

use strict; use experimental 'signatures';
use Net::WebSocket::EVx;
use Compress::Raw::Zlib qw'Z_SYNC_FLUSH Z_OK MAX_WBITS';
use Digest::SHA1 'sha1_base64';

use constant {
    ws_max_size => (1<<31)-1,
    ws_guid => '258EAFA5-E914-47DA-95CA-C5AB0DC85B11',
    ws_inflate_tail => pack(C4 => 0, 0, 255, 255),
    crlf => "\015\012"
};

sub ($env) {
    return [200, ['access-control-allow-origin', $env->{uc'http_origin'} // '*'], []] unless
        ($env->{uc'http_connection'}//'') eq 'upgrade' && ($env->{uc'http_upgrade'}//'') eq 'websocket';
    return [400, [], ['expecting ws v13 handshake']] unless
        ($env->{uc'http_sec_websocket_version'}//'') eq '13' && $env->{uc'http_sec_websocket_key'};
    return [500, [], []] unless exists $env->{'psgix.io'};
    my ($deflate, $inflate);
    if (($env->{uc'http_sec_websocket_extensions'} // '') =~ /permessage-deflate/) {
        $deflate = Compress::Raw::Zlib::Deflate->new(WindowBits => -MAX_WBITS);
        $inflate = Compress::Raw::Zlib::Inflate->new(WindowBits => -MAX_WBITS, Bufsize => ws_max_size, LimitOutput => 1);
    }
    sub {
        my $io = $env->{'psgix.io'};
        my $key = sha1_base64($env->{uc'http_sec_websocket_key'}.ws_guid);
        my $handshake = join crlf,
            'HTTP/1.1 101 Switching Protocols', 'connection: upgrade', 'upgrade: websocket',
            $deflate ? 'sec-websocket-extensions: permessage-deflate' : (),
            "sec-websocket-accept: $key=", crlf;
        my $got = syswrite $io, $handshake;
        die "failed to write ws handshake in one go: $!" unless $got and $got == length $handshake;
        open(my $fh, '+<&', $io) or die $!;
        my $srv; $srv = Net::WebSocket::EVx->new({
            fh => $fh, max_recv_size => ws_max_size,
            rsv => $deflate ? WS_RSV1_BIT : WS_RSV_NONE,
            on_msg_recv => sub ($rsv, $opcode, $msg, $status_code) {
                $srv->queue_msg($msg), return unless $rsv && $inflate; # plain echo
                return unless $inflate->inflate(($msg .= ws_inflate_tail), my $out) == Z_OK;
                return unless $deflate->deflate($out, $msg) == Z_OK && $deflate->flush($msg, Z_SYNC_FLUSH) == Z_OK;
                substr $msg, -4, 4, ''; # cut deflated tail
                $srv->queue_msg_ex($msg);
            },
            on_close => sub ($code) { undef $_ for $io, $fh, $srv, $inflate, $deflate } });
        return
    }
}

Run it via Twiggy or Feersum, which support psgix.io:

plackup -l $(realpath app.sock) -s Feersum app.psgi

Put nginx in front:

http {
  upstream app { server unix:app.sock; }
  map $http_upgrade $connection_upgrade { default upgrade; '' ''; }
  server {
    listen 127.0.0.1:5000;
    location / {
      proxy_pass http://app;
      proxy_ignore_client_abort on;
      proxy_set_header Host $http_host;
      proxy_set_header X-Forwarded-For $http_x_forwarded_for;
      proxy_set_header X-Forwarded-Proto $http_x_forwarded_proto;
      proxy_set_header Upgrade $http_upgrade;
      proxy_set_header Connection $connection_upgrade;
    }
  }
}

Run it:

/usr/sbin/nginx -p . -e nginx.err -c nginx.conf

EXPORTS

WS_FRAGMENTED_DATA, WS_FRAGMENTED_EOF, WS_FRAGMENTED_ERROR, WS_RSV_NONE and WS_RSV1_BIT are exported by default.

METHODS

new( \%params )

Returns a new websocket. The parameter hash becomes the object itself, so the callback keys stay live: replacing $ws->{on_msg_recv} later takes effect on the next message.

fh or fd

Filehandle or numeric file descriptor of the socket to use. The socket must be in non-blocking mode, and its websocket handshake must already be complete.

The descriptor is duplicated with dup(2), so the module owns and closes only its own copy. Ownership of what you pass in stays with you: keep the handle alive for as long as the websocket is in use, and close it yourself afterwards.

Croaks unless a defined, non-negative descriptor can be resolved - a missing, undefined or already closed fh is an error rather than a silent fallback.

type

Either client or server.

Default: server

buffering

If set to 0, disables buffering: on_msg_recv is then always called with an empty $msg and you use the on_frame_recv_* callbacks instead. Useful for handling large binary data without buffering it in memory.

Default: 1

max_recv_size

Maximum message or frame size. See wslay_event_config_set_max_recv_msg_length.

rsv

Reserved bits used by queue_msg_ex() and queue_fragmented_ex() when the caller omits their rsv argument. A per-call argument always wins.

Default: WS_RSV1_BIT

RFC 6455 section 5.2 requires the RSV bits to be 0 unless an extension that defines them has been negotiated, so a connection without permessage-deflate should be created with rsv => WS_RSV_NONE. The default is kept for backwards compatibility.

allowed_rsv

Bitmask of reserved bits accepted on incoming frames; a frame carrying any other RSV bit fails the connection. Only WS_RSV_NONE and WS_RSV1_BIT are meaningful to wslay.

Default: WS_RSV1_BIT

on_msg_recv

Called when a complete message has been received. Close messages are not reported here - use on_close. When buffering is disabled, $msg is always empty.

my ($rsv, $opcode, $msg, $status_code) = @_;
on_close

Called when the connection is closed.

my ($close_code) = @_;
genmask

Client mode only. Must return a scalar of exactly $len bytes to mask the message with; any other length is an error and aborts the connection.

If not specified, a rand() based generator seeded from perl's entropy source is used. That is adequate for the RFC's "unpredictable to intermediaries" requirement, but it is not cryptographically strong - supply your own callback if you need that.

my ($len) = @_;
on_frame_recv_start

Called when a frame header has been received.

my ($fin, $rsv, $opcode, $payload_length) = @_;
on_frame_recv_chunk

Called for each portion of frame data received.

my ($data) = @_;
on_frame_recv_end

Called at the end of a frame. No arguments.

queue_msg( message, opcode )

Queues a message. opcode is optional, default 1 (text). Returns 0 on success or a negative wslay error code on failure.

Control frames go through the same call: $ws->queue_msg($payload, 0x9) sends a ping, and wslay answers incoming pings with a pong by itself.

queue_msg_ex( message, opcode, rsv )

As queue_msg(), but with explicit reserved bits. rsv is optional and defaults to the connection's rsv setting (see new()).

queue_fragmented( callback, opcode )

Queues a fragmented message whose data comes from a callback. opcode is optional, default 2 (binary).

my ($len) = @_;

The callback must return one or two elements: ($data) or ($data, $status). With a single scalar, the status defaults to WS_FRAGMENTED_DATA. Status can be:

WS_FRAGMENTED_DATA

A data chunk. Wslay re-invokes the callback continuously while it returns this. Other events still run, but you will see 100% CPU if the callback keeps returning an empty scalar while waiting for data. To avoid that, call stop_write() when you have nothing more to send and start_write() when the next portion is ready.

WS_FRAGMENTED_EOF

End of message.

WS_FRAGMENTED_ERROR

Error; the callback is not called again and the connection is aborted.

queue_fragmented_ex( callback, opcode, rsv )

As queue_fragmented(), but with explicit reserved bits. rsv is optional and defaults to the connection's rsv setting (see new()).

wait( cb )

Registers a callback to be run when the send queue becomes empty. Setting a new one replaces any callback not yet fired.

queued_count()

Returns the number of messages in the send queue.

start() and stop()

Resume or suspend all websocket IO.

start_write() and stop_write()

Resume or suspend the write side only. Use this for flow control while sending a fragmented message, as described under queue_fragmented().

start_read() and stop_read()

Resume or suspend the read side only. Data already in the socket buffer is delivered once reading is resumed.

shutdown_read() and shutdown_write()

Permanently disable reading or writing at the wslay protocol level. Unlike stop_*/start_*, which only pause the event watcher, a shutdown is irreversible for the lifetime of the connection.

close( status_code, reason_data )

Queues a close frame. Both arguments are optional. reason_data must not exceed 123 bytes; anything longer is rejected by wslay and croaks. (Wslay's own header says "less than 123 bytes", but the implementation accepts 123.)

Possible attack vector: a client can hold the connection open after receiving the close frame, and open many such connections. To close for certain, call close(), wait() until the close frame has been sent, then close $ws->{fh} yourself.

That handshake cannot complete while writing is suspended: call start_write() first if you have used stop_write(), otherwise the close frame is never sent and wait() never fires.

ERRORS AND EXCEPTIONS

The queueing methods return a negative wslay error code for conditions you are expected to handle - in practice only WSLAY_ERR_NO_MORE_MSG (-302), returned when queueing after a close frame has been queued.

Programming errors croak instead: an invalid argument (an over-long close reason, for instance), WSLAY_ERR_NOMEM, and calling any method on a connection that is already closed or on an object that has been destroyed. Wrap sends in eval if you cannot rule those out.

A callback that dies does not unwind through wslay and libev. The exception is held until the C state is consistent, then rethrown from whichever call entered the module - an EV::run, or the queue_*/start_*/stop_* method you called. While an exception is pending, the remaining callbacks for that event are skipped.

Whether the connection survives depends on which callback died. on_msg_recv, on_frame_recv_*, on_close and the wait() callback are advisory, so the connection is left intact and you can carry on using it. genmask and the queue_fragmented source callback produce bytes that wslay is in the middle of framing, so a failure there - dying, or returning the wrong length - aborts the connection.

The object must not outlive the event loop it was created on. Watchers are disarmed automatically if the default loop is destroyed first, but any websocket still open at that point stops doing IO.

Unix only. The WIN32 guards in the XS are incomplete: socket errors are read from errno rather than WSAGetLastError(), and descriptors are closed with close() rather than closesocket().

SEE ALSO

EV, Alien::Wslay, Net::WebSocket::EV, RFC 6455, RFC 7692 (permessage-deflate)

AUTHOR

Yegor Korablev <egor@cpan.org>

LICENSE

This is free software; you can redistribute it and/or modify it under the same terms as Perl itself.