view release on metacpan or  search on metacpan

t/http2/07-websocket.t  view on Meta::CPAN

};

# ============================================================
# Extended CONNECT → app receives websocket scope
# ============================================================
subtest 'Extended CONNECT creates websocket scope' => sub {
    my @scopes;

    my $app = async sub {
        my ($scope, $receive, $send) = @_;
        push @scopes, $scope;

        if ($scope->{type} eq 'websocket') {
            # Accept the WebSocket connection
            await $send->({
                type => 'websocket.accept',
            });

            # Receive connect event
            my $event = await $receive->();

            # Wait for disconnect
            while ($event->{type} ne 'websocket.disconnect') {
                $event = await $receive->();
            }
        }
    };

    my ($conn, $stream, $client_sock, $server) = create_h2_connection(app => $app);

    my %response_headers;
    my $response_data = '';

    my $client = create_client(
        on_header => sub {
            my ($sid, $name, $value) = @_;
            $response_headers{$name} = $value;
            return 0;
        },
        on_data_chunk_recv => sub {
            my ($sid, $data) = @_;
            $response_data .= $data;
            return 0;
        },
    );

    complete_h2_handshake($client, $client_sock);

    # Send Extended CONNECT for WebSocket (RFC 8441)
    $client->submit_request(
        method    => 'CONNECT',
        path      => '/ws/chat',
        scheme    => 'https',
        authority => 'localhost',
        headers   => [
            [':protocol', 'websocket'],
            ['sec-websocket-version', '13'],
            ['sec-websocket-protocol', 'chat, superchat'],
            ['origin', 'https://localhost'],
        ],
        body      => sub { return undef },  # streaming: keep open
    );
    $client_sock->syswrite($client->mem_send);

    exchange_frames($client, $client_sock);

    ok(scalar @scopes >= 1, 'App was called');

    if (@scopes) {
        my $scope = $scopes[0];
        is($scope->{type}, 'websocket', 'scope type is websocket');
        is($scope->{http_version}, '2', 'http_version is 2');
        is($scope->{path}, '/ws/chat', 'path is /ws/chat');
        is($scope->{scheme}, 'ws', 'scheme is ws (WebSocket, no TLS in test)');
        ok(ref $scope->{headers} eq 'ARRAY', 'headers is array');
        ok(ref $scope->{subprotocols} eq 'ARRAY', 'subprotocols is array');
        is(scalar @{$scope->{subprotocols}}, 2, 'Two subprotocols');
        is($scope->{subprotocols}[0], 'chat', 'First subprotocol');
        is($scope->{subprotocols}[1], 'superchat', 'Second subprotocol');
    }

    # Server should have responded with 200 (not 101)
    is($response_headers{':status'}, '200', 'Server responded with 200');

    $stream->close_now;
    $loop->remove($server);
};

# ============================================================
# Bidirectional text message exchange
# ============================================================
subtest 'Bidirectional text message exchange' => sub {
    my @received_messages;

    my $app = async sub {
        my ($scope, $receive, $send) = @_;

        if ($scope->{type} eq 'websocket') {
            await $send->({ type => 'websocket.accept' });

            # Read connect event
            my $event = await $receive->();
            next unless $event->{type} eq 'websocket.connect';

            # Echo loop
            while (1) {
                $event = await $receive->();
                if ($event->{type} eq 'websocket.receive') {
                    push @received_messages, $event;
                    # Echo back
                    await $send->({
                        type => 'websocket.send',
                        text => "echo: $event->{text}",
                    });
                }
                elsif ($event->{type} eq 'websocket.disconnect') {
                    last;
                }
            }
        }
    };

    my ($conn, $stream_io, $client_sock, $server) = create_h2_connection(app => $app);

    my %response_headers;
    my $ws_data = '';  # Raw DATA frames received on the WS stream

    my $client = create_client(
        on_header => sub {
            my ($sid, $name, $value) = @_;
            $response_headers{$name} = $value;
            return 0;
        },
        on_data_chunk_recv => sub {
            my ($sid, $data) = @_;
            $ws_data .= $data;
            return 0;
        },
    );

    complete_h2_handshake($client, $client_sock);

    # Send Extended CONNECT
    my $ws_stream_id = $client->submit_request(
        method    => 'CONNECT',
        path      => '/ws/echo',
        scheme    => 'https',
        authority => 'localhost',
        headers   => [
            [':protocol', 'websocket'],
            ['sec-websocket-version', '13'],
        ],
        body      => sub { return undef },  # streaming: keep open
    );
    $client_sock->syswrite($client->mem_send);

    # Wait for 200 response
    exchange_frames($client, $client_sock);
    is($response_headers{':status'}, '200', 'Got 200 for WebSocket accept');

    # Send a WebSocket text frame via HTTP/2 DATA
    my $ws_frame = Protocol::WebSocket::Frame->new(
        buffer => 'Hello WebSocket',
        type   => 'text',
        masked => 1,
    );
    my $frame_bytes = $ws_frame->to_bytes;

    send_stream_data($client, $client_sock, $ws_stream_id, $frame_bytes);

    # Read the echo response
    $ws_data = '';
    exchange_frames($client, $client_sock);

    # Parse the server's WebSocket frame from the DATA
    ok(length($ws_data) > 0, 'Received data from server');
    if (length($ws_data) > 0) {
        my $response_frame = Protocol::WebSocket::Frame->new;
        $response_frame->append($ws_data);
        my $text = $response_frame->next_bytes;
        is($text, 'echo: Hello WebSocket', 'Got echoed message');
    }

    ok(scalar @received_messages >= 1, 'Server received message');
    is($received_messages[0]{text}, 'Hello WebSocket', 'Server got correct text');

    $stream_io->close_now;
    $loop->remove($server);
};

# ============================================================
# Close handshake
# ============================================================
subtest 'WebSocket close handshake over HTTP/2' => sub {
    my $disconnect_event;

    my $app = async sub {
        my ($scope, $receive, $send) = @_;

        if ($scope->{type} eq 'websocket') {
            await $send->({ type => 'websocket.accept' });
            my $event = await $receive->();  # websocket.connect

            # Wait for disconnect
            while (1) {
                $event = await $receive->();
                if ($event->{type} eq 'websocket.disconnect') {
                    $disconnect_event = $event;
                    last;
                }
            }
        }
    };

    my ($conn, $stream_io, $client_sock, $server) = create_h2_connection(app => $app);

    my %response_headers;
    my $ws_data = '';
    my $stream_closed = 0;

    my $client = create_client(
        on_header => sub {
            my ($sid, $name, $value) = @_;
            $response_headers{$name} = $value;
            return 0;
        },
        on_data_chunk_recv => sub {
            my ($sid, $data) = @_;
            $ws_data .= $data;
            return 0;
        },
        on_stream_close => sub {
            $stream_closed = 1;
            return 0;
        },
    );

    complete_h2_handshake($client, $client_sock);

    # Open WebSocket
    my $ws_stream_id = $client->submit_request(
        method    => 'CONNECT',
        path      => '/ws/close-test',
        scheme    => 'https',
        authority => 'localhost',
        headers   => [
            [':protocol', 'websocket'],
            ['sec-websocket-version', '13'],
        ],
        body      => sub { return undef },  # streaming: keep open
    );
    $client_sock->syswrite($client->mem_send);
    exchange_frames($client, $client_sock);

    is($response_headers{':status'}, '200', 'WebSocket accepted');

    # Send close frame (code=1000, reason="normal closure")
    my $close_frame = Protocol::WebSocket::Frame->new(
        type   => 'close',
        buffer => pack('n', 1000) . 'normal closure',
        masked => 1,
    );
    send_stream_data($client, $client_sock, $ws_stream_id, $close_frame->to_bytes);

    exchange_frames($client, $client_sock);

    # Server should have received the disconnect event
    ok(defined $disconnect_event, 'Server got disconnect event');
    if ($disconnect_event) {
        is($disconnect_event->{code}, 1000, 'Close code is 1000');
        is($disconnect_event->{reason}, 'normal closure', 'Close reason matches');
    }

    # Server should have sent close frame back
    if (length($ws_data) > 0) {
        my $frame = Protocol::WebSocket::Frame->new;
        $frame->append($ws_data);
        my $bytes = $frame->next_bytes;
        # Close frame echo
        ok(defined $bytes, 'Server sent close frame response');
    }

    $stream_io->close_now;
    $loop->remove($server);
};

done_testing;