view release on metacpan or  search on metacpan

lib/Net/WebSocket/EVx.pod  view on Meta::CPAN

    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 C<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

=head1 EXPORTS

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

=head1 METHODS

=head2 new( \%params )

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

=over

=item 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 C<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 C<fh> is an error rather than a silent fallback.

=item type

Either C<client> or C<server>.

Default: C<server>

=item buffering

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

Default: 1

=item max_recv_size

Maximum message or frame size. See
L<wslay_event_config_set_max_recv_msg_length|https://tatsuhiro-t.github.io/wslay/man/wslay_event_config_set_max_recv_msg_length.html>.

=item rsv

Reserved bits used by C<queue_msg_ex()> and C<queue_fragmented_ex()> when the