PAGI-Server
view release on metacpan or search on metacpan
lib/PAGI/Server/Connection.pm view on Meta::CPAN
package PAGI::Server::Connection;
use strict;
use warnings;
our $VERSION = '0.002006';
use Future;
use Future::AsyncAwait;
use Scalar::Util qw(weaken refaddr);
use Protocol::WebSocket::Handshake::Server;
use Protocol::WebSocket::Frame;
use Digest::SHA qw(sha1_base64);
use Encode;
use URI::Escape qw(uri_unescape);
use IO::Async::Timer::Countdown;
use IO::Async::Timer::Periodic;
use Time::HiRes qw(gettimeofday tv_interval);
use PAGI::Server::AsyncFile;
use PAGI::Server::ConnectionState;
use PAGI::Server::TransportState;
use constant FILE_CHUNK_SIZE => 65536; # 64KB chunks for file streaming
# Per-second cache for CLF timestamp in access log (same pattern as HTTP1::format_date)
my $_cached_log_timestamp;
my $_cached_log_time = 0;
# =============================================================================
# Unrecognized Event Type Handler (PAGI spec compliance)
# =============================================================================
# Per main.mkdn: "Servers must raise exceptions if... The type field is unrecognized"
sub _unrecognized_event_type {
my ($type, $protocol) = @_;
die "Unrecognized event type '$type' for $protocol protocol\n";
}
# =============================================================================
# Header Validation (CRLF Injection Prevention)
# =============================================================================
# RFC 7230 Section 3.2.6: Field values MUST NOT contain CR or LF
sub _validate_header_value {
my ($value) = @_;
if ($value =~ /[\r\n\0]/) {
die "Invalid header value: contains CR, LF, or null byte\n";
}
return $value;
}
sub _validate_header_name {
my ($name) = @_;
if ($name =~ /[\r\n\0]/) {
die "Invalid header name: contains CR, LF, or null byte\n";
}
if ($name =~ /[[:cntrl:]]/) {
die "Invalid header name: contains control characters\n";
}
return $name;
}
# RFC 6455 Section 11.3.4: Subprotocol must be a token (no whitespace, separators)
sub _validate_subprotocol {
my ($value) = @_;
if ($value =~ /[\r\n\0\s]/) {
die "Invalid subprotocol: contains CR, LF, null, or whitespace\n";
}
# Token characters only (roughly)
if ($value !~ /^[\w\-\.]+$/) {
die "Invalid subprotocol: must be alphanumeric, dash, underscore, or dot\n";
}
return $value;
}
=head1 NAME
PAGI::Server::Connection - Per-connection state machine
=head1 SYNOPSIS
lib/PAGI/Server/Connection.pm view on Meta::CPAN
return { type => 'http.disconnect' } unless $weak_self;
my $ss = $weak_self->{h2_streams}{$stream_id};
return { type => 'http.disconnect' } unless $ss;
# Check queue first
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
# If body is already complete, return final body event
if ($ss->{body_complete}) {
my $body = $ss->{body};
$ss->{body} = '';
return {
type => 'http.request',
body => $body,
more => 0,
};
}
# Wait for body data
if (!$ss->{body_pending}) {
$ss->{body_pending} = Future->new;
}
await $ss->{body_pending};
# Re-fetch stream state (may have changed)
$ss = $weak_self->{h2_streams}{$stream_id};
return { type => 'http.disconnect' } unless $ss;
# Check queue after waking
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
my $body = $ss->{body};
$ss->{body} = '';
return {
type => 'http.request',
body => $body,
more => $ss->{body_complete} ? 0 : 1,
};
})->();
return $future;
};
}
sub _h2_create_send {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
my $status;
my @response_headers;
# Streaming state for deferred data provider pattern.
# The send queue lives on per-stream state ($ss->{send_queue} /
# $ss->{send_queue_bytes}) so the h2 transport handle can measure it;
# $eof_pending / $streaming_started stay closure-local.
my $eof_pending = 0;
my $streaming_started = 0;
# Data callback for nghttp2's streaming response.
# Returns ($data, $eof) when data is available, or undef to defer.
my $data_callback = sub {
my ($cb_stream_id, $max_len) = @_;
my $ss = $weak_self && $weak_self->{h2_streams}{$stream_id};
return undef unless $ss;
my $q = $ss->{send_queue} ||= [];
if (@$q) {
my $chunk = shift @$q;
# Respect max_len — XS truncates without preserving remainder
if (length($chunk) > $max_len) {
unshift @$q, substr($chunk, $max_len);
$chunk = substr($chunk, 0, $max_len);
}
$ss->{send_queue_bytes} -= length($chunk);
# Per-stream backpressure: once this stream's queue falls below the
# low watermark, release any producer blocked in
# _h2_wait_for_stream_drain. This callback runs inside nghttp2's
# extract(), so resolve on the next loop tick — completing the Future
# resumes the awaiting producer synchronously, and it must not call
# resume_stream/_h2_write_pending re-entrantly into nghttp2.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{stream_drain_waiters} && @{$ss->{stream_drain_waiters}}) {
my @waiters = splice @{$ss->{stream_drain_waiters}};
$weak_self->{server}->loop->later(sub {
$_->done for grep { !$_->is_ready } @waiters;
});
}
# Fire the app's on_drain hysteresis callbacks once this stream's
# queue falls below the low watermark. Like the waiters above, this
# runs inside nghttp2's extract(), and an on_drain callback may call
# $send to resume its source — which would re-enter nghttp2. Splice
# the fires out first (so they can't double-fire), then invoke them on
# the next loop tick.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{transport_drain_fires} && @{$ss->{transport_drain_fires}}) {
my @fires = splice @{$ss->{transport_drain_fires}};
$weak_self->{server}->loop->later(sub {
$_->() for @fires;
});
}
my $eof = (!@$q && $eof_pending) ? 1 : 0;
return ($chunk, $eof);
}
# Queue empty but EOF pending — signal end of stream
if ($eof_pending) {
return ('', 1);
}
# Queue empty, more data expected — defer (NGHTTP2_ERR_DEFERRED in C layer)
return undef;
};
return async sub {
my ($event) = @_;
return unless $weak_self;
return if $weak_self->{closed};
my $type = $event->{type} // '';
if ($type eq 'http.response.start') {
my $ss = $weak_self->{h2_streams}{$stream_id} or return;
return if $ss->{response_started};
$ss->{response_started} = 1;
$ss->{connection_state}->_mark_response_started if $ss->{connection_state};
$status = $event->{status} // 200;
@response_headers = map {
[_validate_header_name($_->[0]), _validate_header_value($_->[1])]
} @{$event->{headers} // []};
# Server-supplied Date header (HTTP/1.1 parity) — add if the app didn't.
unless (grep { lc($_->[0]) eq 'date' } @response_headers) {
push @response_headers, ['date', $weak_self->{protocol}->format_date];
}
}
elsif ($type eq 'http.response.body') {
my $ss = $weak_self->{h2_streams}{$stream_id} or return;
return unless $ss->{response_started};
my $body = $event->{body} // '';
my $more = $event->{more} // 0;
if ($more) {
if (!$streaming_started) {
# First streaming chunk — submit with data callback
$streaming_started = 1;
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
# Synchronous: we're in the app's send path (not nghttp2's
# extract), so on_high_water can fire here to tell the app to
# pause its source.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status,
headers => \@response_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
} else {
# Subsequent chunk — backpressure check then push and resume.
# Bound on THIS stream's send queue (per-stream), not the
# shared TCP buffer which is meaningless across multiplexed
# streams.
if (($ss->{send_queue_bytes} // 0) >= $weak_self->{write_high_watermark}) {
await $weak_self->_h2_wait_for_stream_drain($stream_id);
return unless $weak_self;
return if $weak_self->{closed};
return unless $weak_self->{h2_streams}{$stream_id};
}
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
# Synchronous — app send path, not nghttp2 extract.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
}
} else {
if ($streaming_started) {
# Final chunk on an already-streaming response. Bound on THIS
# stream's send queue (per-stream), not the shared TCP buffer.
if (($ss->{send_queue_bytes} // 0) >= $weak_self->{write_high_watermark}) {
await $weak_self->_h2_wait_for_stream_drain($stream_id);
return unless $weak_self;
return if $weak_self->{closed};
return unless $weak_self->{h2_streams}{$stream_id};
}
$eof_pending = 1;
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
# Synchronous — app send path, not nghttp2 extract.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
} else {
# Non-streaming: single response (unchanged one-shot path)
$weak_self->{h2_session}->submit_response($stream_id,
status => $status,
headers => \@response_headers,
body => $body,
);
$weak_self->_h2_write_pending;
}
}
}
else {
_unrecognized_event_type($type, 'http');
}
};
}
# =============================================================================
# HTTP/2 WebSocket over HTTP/2 (RFC 8441)
# =============================================================================
sub _h2_create_websocket_scope {
my ($self, $stream_id, $stream_state) = @_;
my $pseudo = $stream_state->{pseudo};
my $headers = $stream_state->{headers};
my $full_path = $pseudo->{':path'} // '/';
my ($path, $query_string) = split(/\?/, $full_path, 2);
$query_string //= '';
# Match HTTP/1.1 pipeline: URI::Escape + UTF-8 decode with fallback
my $unescaped = uri_unescape($path);
my $decoded_path = eval { decode('UTF-8', $unescaped, Encode::FB_CROAK) }
// $unescaped;
# Extract subprotocols from headers
my @subprotocols;
for my $header (@$headers) {
my ($name, $value) = @$header;
if ($name eq 'sec-websocket-protocol') {
push @subprotocols, map { s/^\s+|\s+$//gr } split /,/, $value;
}
}
return {
type => 'websocket',
pagi => {
version => '0.3',
spec_version => '0.3',
},
http_version => '2',
scheme => $self->_get_ws_scheme,
path => $decoded_path,
raw_path => $path,
query_string => $query_string,
root_path => '',
headers => $headers,
(defined $self->{client_host}
? (client => [$self->{client_host}, $self->{client_port}])
: ()
),
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# Wait for events
while (1) {
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
return { type => 'websocket.disconnect', code => 1006, reason => '' }
if $weak_self->{closed};
if (!$ss->{body_pending}) {
$ss->{body_pending} = Future->new;
}
await $ss->{body_pending};
$ss = $weak_self->{h2_streams}{$stream_id};
return { type => 'websocket.disconnect', code => 1006, reason => '' }
unless $ss;
}
})->();
return $future;
};
}
sub _h2_create_websocket_send {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
return async sub {
my ($event) = @_;
return unless $weak_self;
return if $weak_self->{closed};
my $ss = $weak_self->{h2_streams}{$stream_id};
return unless $ss;
my $type = $event->{type} // '';
if ($type eq 'websocket.accept') {
return if $ss->{ws_accepted};
# HTTP/2 WebSocket: respond with 200 (not 101)
my @headers;
if (my $subprotocol = $event->{subprotocol}) {
$subprotocol = _validate_subprotocol($subprotocol);
push @headers, ['sec-websocket-protocol', $subprotocol];
}
if (my $extra = $event->{headers}) {
push @headers, map {
[_validate_header_name($_->[0]), _validate_header_value($_->[1])]
} @$extra;
}
$ss->{ws_accepted} = 1;
$ss->{response_started} = 1;
$ss->{ws_frame} = Protocol::WebSocket::Frame->new(
max_payload_size => $weak_self->{max_ws_frame_size},
);
# Submit 200 response with streaming body that defers
$weak_self->{h2_session}->submit_response($stream_id,
status => 200,
headers => \@headers,
body => sub { return undef }, # defer until submit_data
);
$weak_self->_h2_write_pending;
# Process any data that arrived before accept
if (length($ss->{body}) > 0) {
my $buffered = $ss->{body};
$ss->{body} = '';
$weak_self->_h2_process_ws_frames($stream_id, $ss, $buffered);
}
}
elsif ($type eq 'websocket.send') {
return unless $ss->{ws_accepted};
my $frame;
if (defined $event->{text}) {
$frame = Protocol::WebSocket::Frame->new(
buffer => $event->{text},
type => 'text',
);
}
elsif (defined $event->{bytes}) {
$frame = Protocol::WebSocket::Frame->new(
buffer => $event->{bytes},
type => 'binary',
);
}
else {
return;
}
my $bytes = $frame->to_bytes;
$weak_self->{h2_session}->submit_data($stream_id, $bytes, 0);
$weak_self->_h2_write_pending;
}
elsif ($type eq 'websocket.http.response.start') {
return if $ss->{ws_accepted};
return if $ss->{ws_denial_started};
$ss->{ws_denial_started} = 1;
$ss->{ws_denial_status} = $event->{status} // 403;
$ss->{ws_denial_headers} = [
map { [_validate_header_name($_->[0]), _validate_header_value($_->[1])] }
@{$event->{headers} // []}
];
$ss->{ws_denial_body} = '';
}
elsif ($type eq 'websocket.http.response.body') {
return unless $ss->{ws_denial_started};
return if $ss->{response_started};
$ss->{ws_denial_body} .= $event->{body} // '';
return if $event->{more}; # more chunks coming — keep buffering
$ss->{response_started} = 1;
$weak_self->{h2_session}->submit_response($stream_id,
status => $ss->{ws_denial_status},
headers => $ss->{ws_denial_headers},
body => $ss->{ws_denial_body},
lib/PAGI/Server/Connection.pm view on Meta::CPAN
return;
}
my $code = $event->{code} // 1000;
my $reason = $event->{reason} // '';
my $frame = Protocol::WebSocket::Frame->new(
type => 'close',
buffer => pack('n', $code) . $reason,
);
# Send close frame + END_STREAM
$weak_self->{h2_session}->submit_data($stream_id, $frame->to_bytes, 1);
$weak_self->_h2_write_pending;
}
return;
};
}
# =============================================================================
# HTTP/2 SSE (Server-Sent Events over HTTP/2)
# =============================================================================
sub _h2_create_sse_scope {
my ($self, $stream_id, $stream_state) = @_;
my $pseudo = $stream_state->{pseudo};
my $headers = $stream_state->{headers};
my $full_path = $pseudo->{':path'} // '/';
my ($path, $query_string) = split(/\?/, $full_path, 2);
$query_string //= '';
# Match HTTP/1.1 pipeline: URI::Escape + UTF-8 decode with fallback
my $unescaped = uri_unescape($path);
my $decoded_path = eval { decode('UTF-8', $unescaped, Encode::FB_CROAK) }
// $unescaped;
return {
type => 'sse',
pagi => {
version => '0.3',
spec_version => '0.3',
},
http_version => '2',
method => $pseudo->{':method'} // 'GET',
scheme => $pseudo->{':scheme'} // $self->_get_scheme,
path => $decoded_path,
raw_path => $path,
query_string => $query_string,
root_path => '',
headers => $headers,
(defined $self->{client_host}
? (client => [$self->{client_host}, $self->{client_port}])
: ()
),
server => [$self->{server_host}, $self->{server_port}],
state => keys %{$self->{state}} ? { %{$self->{state}} } : {},
extensions => $self->_get_extensions_for_scope,
# Per-stream outbound flow-control handle. Like the h2 streaming scope,
# it measures THIS stream's send queue (h2 multiplexes many streams over
# one connection, so the shared TCP buffer is meaningless per stream).
'pagi.transport' => ($stream_state->{transport_state} = $self->_h2_transport_state($stream_state)),
};
}
sub _h2_create_sse_receive {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
my $sse_disconnect = sub {
return {
type => 'sse.disconnect',
reason => 'client_closed',
};
};
return sub {
return Future->done($sse_disconnect->()) unless $weak_self;
return Future->done($sse_disconnect->()) if $weak_self->{closed};
my $ss = $weak_self->{h2_streams}{$stream_id};
return Future->done($sse_disconnect->()) unless $ss;
my $future = (async sub {
return $sse_disconnect->() unless $weak_self;
my $ss = $weak_self->{h2_streams}{$stream_id};
return $sse_disconnect->() unless $ss;
# Check queue first
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
# First call returns sse.request with body
if (!$ss->{sse_request_sent}) {
$ss->{sse_request_sent} = 1;
return {
type => 'sse.request',
body => $ss->{body},
more => 0,
};
}
# Wait for disconnect
while (1) {
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
return $sse_disconnect->()
if $weak_self->{closed};
if (!$ss->{body_pending}) {
$ss->{body_pending} = Future->new;
}
await $ss->{body_pending};
$ss = $weak_self->{h2_streams}{$stream_id};
return $sse_disconnect->() unless $ss;
}
})->();
return $future;
};
}
sub _h2_create_sse_send {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
# Streaming state for the data-provider pull pattern. The send queue lives on
# per-stream state ($ss->{send_queue} / $ss->{send_queue_bytes}) so the
# pagi.transport handle can measure THIS stream's backlog. $streaming_started
# stays closure-local.
my $streaming_started = 0;
# Data callback for nghttp2's streaming response. Pulls from the per-stream
# queue; SSE responses stay open, so this never signals EOF (returns eof=0),
# or undef to defer when the queue is empty.
my $data_callback = sub {
my ($cb_stream_id, $max_len) = @_;
my $ss = $weak_self && $weak_self->{h2_streams}{$stream_id};
return undef unless $ss;
my $q = $ss->{send_queue} ||= [];
if (@$q) {
my $chunk = shift @$q;
# Respect max_len — XS truncates without preserving remainder
if (length($chunk) > $max_len) {
unshift @$q, substr($chunk, $max_len);
$chunk = substr($chunk, 0, $max_len);
}
$ss->{send_queue_bytes} -= length($chunk);
# Per-stream backpressure: once this stream's queue falls below the
# low watermark, release any producer blocked in
# _h2_wait_for_stream_drain. This runs inside nghttp2's extract(), so
# resolve on the next loop tick — completing the Future resumes the
# awaiting producer synchronously, and it must not re-enter nghttp2.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{stream_drain_waiters} && @{$ss->{stream_drain_waiters}}) {
my @waiters = splice @{$ss->{stream_drain_waiters}};
$weak_self->{server}->loop->later(sub {
$_->done for grep { !$_->is_ready } @waiters;
});
}
# Fire the app's on_drain hysteresis callbacks once this stream's
# queue falls below the low watermark. Deferred for the same reason:
# an on_drain callback may call $send, which would re-enter nghttp2.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{transport_drain_fires} && @{$ss->{transport_drain_fires}}) {
my @fires = splice @{$ss->{transport_drain_fires}};
$weak_self->{server}->loop->later(sub {
$_->() for @fires;
});
}
return ($chunk, 0); # SSE streams never EOF via data_callback
}
# Queue empty. If the application closed this stream (sse.close), emit a
# final empty DATA frame with END_STREAM to terminate it; otherwise defer.
return ('', 1) if $ss->{sse_closing};
# Queue empty — defer (NGHTTP2_ERR_DEFERRED in the C layer)
return undef;
};
return async sub {
my ($event) = @_;
return unless $weak_self;
my $ss = $weak_self->{h2_streams}{$stream_id};
return unless $ss;
my $type = $event->{type} // '';
# After an application-initiated sse.close on THIS stream, a second
# sse.close is a no-op and any other send raises. Tracked per-stream
# ($ss), since HTTP/2 multiplexes many SSE streams over one connection.
if ($ss->{sse_close_sent}) {
return if $type eq 'sse.close';
die "cannot send '$type' after sse.close\n";
}
# After an sse.http.response.start (decline), only the decline body may
# follow; a stream event is a programming error (first-send-wins).
if ($ss->{sse_decline_started} && $type !~ /^sse\.http\.response\./) {
die "cannot send '$type' after sse.http.response.start\n";
}
return if $weak_self->{closed};
# Reset SSE idle timer on send activity
$weak_self->_reset_sse_idle_timer;
# Dev-mode event validation (PAGI spec compliance)
if ($weak_self->{validate_events}) {
require PAGI::Server::EventValidator;
PAGI::Server::EventValidator::validate_sse_send($event);
}
if ($type eq 'sse.start') {
return if $ss->{response_started};
$ss->{response_started} = 1;
my $status = $event->{status} // 200;
my $headers = $event->{headers} // [];
# Ensure Content-Type is text/event-stream
my $has_content_type = 0;
for my $h (@$headers) {
if (lc($h->[0]) eq 'content-type') {
$has_content_type = 1;
last;
}
}
my @final_headers;
for my $h (@$headers) {
push @final_headers, [_validate_header_name($h->[0]), _validate_header_value($h->[1])];
}
if (!$has_content_type) {
push @final_headers, ['content-type', 'text/event-stream'];
}
push @final_headers, ['cache-control', 'no-cache'];
# Server-supplied Date header (HTTP/1.1 parity) — the h1 SSE path adds
# this too; add it unless the app supplied one.
unless (grep { lc($_->[0]) eq 'date' } @final_headers) {
push @final_headers, ['date', $weak_self->{protocol}->format_date];
}
$streaming_started = 1;
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status,
headers => \@final_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
# Protocol-specific keepalive writer (HTTP/2 DATA frames). Keepalive
# bytes are counted in the per-stream backlog so buffered_amount stays
# accurate, but they do not poke the watermark callbacks — a server
# heartbeat is not an application send.
$weak_self->{sse_keepalive_writer} = sub {
my ($text) = @_;
return unless $weak_self;
return if $weak_self->{closed};
my $ss = $weak_self->{h2_streams}{$stream_id} or return;
push @{$ss->{send_queue} ||= []}, $text;
$ss->{send_queue_bytes} = ($ss->{send_queue_bytes} // 0) + length $text;
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
};
# Start SSE idle timer if configured
$weak_self->_start_sse_idle_timer;
}
elsif ($type eq 'sse.send') {
return unless $ss->{response_started};
# Per-stream backpressure: bound on THIS stream's queue, not the
# shared TCP buffer (meaningless across multiplexed h2 streams).
if (($ss->{send_queue_bytes} // 0) >= $weak_self->{write_high_watermark}) {
await $weak_self->_h2_wait_for_stream_drain($stream_id);
return unless $weak_self;
return if $weak_self->{closed};
return unless $weak_self->{h2_streams}{$stream_id};
}
my $sse_data = _format_sse_event($event);
push @{$ss->{send_queue} ||= []}, $sse_data;
$ss->{send_queue_bytes} = ($ss->{send_queue_bytes} // 0) + length $sse_data;
# Synchronous — app send path, not nghttp2 extract — so on_high_water
# may fire here to tell the app to pause its source.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
}
elsif ($type eq 'sse.comment') {
return unless $ss->{response_started};
my $comment = _format_sse_comment($event);
push @{$ss->{send_queue} ||= []}, $comment;
$ss->{send_queue_bytes} = ($ss->{send_queue_bytes} // 0) + length $comment;
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
}
elsif ($type eq 'sse.keepalive') {
my $interval = $event->{interval} // 0;
my $comment = $event->{comment};
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# Seek to offset if specified
if ($offset && $offset > 0) {
seek($fh, $offset, 0) or die "Cannot seek: $!";
}
# For filehandles, we can't easily use the worker pool (can't pass fh across fork).
# Use blocking reads in small chunks - not ideal but practical.
# TODO: Consider IO::Async::FileStream for better event loop integration.
my $remaining = $length; # undef means read to EOF
my $stream = $self->{stream};
while (1) {
my $to_read = FILE_CHUNK_SIZE;
if (defined $remaining) {
$to_read = $remaining if $remaining < $to_read;
last if $to_read <= 0;
}
my $bytes_read = read($fh, my $chunk, $to_read);
last if !defined $bytes_read; # Error
last if $bytes_read == 0; # EOF
$self->{_response_size} += $bytes_read;
if ($chunked) {
my $len = sprintf("%x", length($chunk));
$stream->write("$len\r\n$chunk\r\n");
}
else {
$stream->write($chunk);
}
if (defined $remaining) {
$remaining -= $bytes_read;
}
}
# Send final chunk if chunked encoding
if ($chunked) {
$stream->write("0\r\n\r\n");
}
}
1;
__END__
=head1 SSE OVER HTTP/2
SSE events (C<sse.start>, C<sse.send>, C<sse.comment>, C<sse.keepalive>)
work transparently over both HTTP/1.1 and HTTP/2. Applications do not need
to change their SSE handling code based on protocol version.
=head2 How It Works
When a request arrives with C<Accept: text/event-stream>, the connection
detects it as SSE regardless of HTTP version. Over HTTP/1.1, SSE data is
sent using chunked Transfer-Encoding. Over HTTP/2, SSE data is sent as
DATA frames via the C<submit_response_streaming>/C<data_callback> mechanism.
This difference is transparent to the application.
The C<http_version> field in the scope hash will be C<'2'> for HTTP/2
connections, allowing applications to distinguish if needed.
=head2 SSE Idle Timeout over HTTP/2
The C<sse_idle_timeout> setting applies at the B<connection level>,
not per-stream. Over HTTP/1.1, this is a non-issue since each SSE
stream occupies its own TCP connection. Over HTTP/2, where multiple
streams share a single connection, an idle SSE stream timeout will
close the B<entire connection>, terminating all active streams.
=head3 Trade-offs
B<Pros:>
=over 4
=item * Simple, consistent behavior across protocols
=item * Matches the approach used by Go (net/http2), Rust (hyper/Axum),
Java (Netty/Reactor Netty/Vert.x), Python (Hypercorn), and gRPC
=item * No additional complexity in stream lifecycle management
=back
B<Cons:>
=over 4
=item * Closing the connection affects all multiplexed HTTP/2 streams,
not just the idle SSE stream
=item * Clients multiplexing SSE + REST on one HTTP/2 connection may
see unexpected disconnects on their REST requests
=back
B<Recommendation:> Use SSE keepalive comments (C<sse.keepalive> event)
with an interval shorter than C<sse_idle_timeout> to prevent the timer
from firing. This is the industry-standard approach used across all
major frameworks. For production deployments behind reverse proxies
(Envoy, Nginx, HAProxy), align your keepalive interval with the
proxy's stream idle timeout.
B<Note:> Per-stream idle timeout (using HTTP/2 RST_STREAM to close
only the idle SSE stream) is a future enhancement. Only Node.js
(http2stream.setTimeout) and Envoy (stream_idle_timeout) implement
this among mainstream servers.
=head1 SEE ALSO
L<PAGI::Server>, L<PAGI::Server::Protocol::HTTP1>
=head1 AUTHOR
John Napiorkowski E<lt>jjnapiork@cpan.orgE<gt>
( run in 0.487 second using v1.01-cache-2.11-cpan-995e09ba956 )