NAME
Linux::Event::Stream - subclass-defined native buffered streams
SYNOPSIS
use v5.36;
use Linux::Event::Loop;
package EchoStream;
use parent 'Linux::Event::Stream';
use Linux::Event::Framer 'Delimiter', "\n";
sub on_message ($stream, $message) {
$stream->send($message);
}
sub on_eof ($stream) {
$stream->end;
}
sub on_error ($stream, $error) {
warn "$error\n";
}
package main;
my $loop = Linux::Event::Loop->new;
my $stream = $loop->add(EchoStream->new(
fh => $socket, # required
data => { user_id => 42 }, # optional
));
$loop->run;DESCRIPTION
Linux::Event::Stream is a resource-owning object backed by the native buffered byte-stream engine above Linux::Event::Loop. It is a base class rather than a configurable Stream type. Applications define behavior once in a subclass and construct lightweight per-connection instances containing only changing connection state.
The first construction of each subclass resolves its inherited callback CVs, framer and TLS declarations, parser configuration, and Stream policy into one cached descriptor. XS stores that descriptor once and every connection's native state references it. Construction therefore avoids per-object callback hashes, framer objects, repeated validation, and repeated native configuration copies.
DEFINING A STREAM TYPE
A raw subclass defines on_data and does not declare a framer:
package ByteStream;
use parent 'Linux::Event::Stream';
sub on_data ($stream, $bytes) {
$stream->write($bytes);
}A framed subclass imports one native built-in and defines on_message:
package LineStream;
use parent 'Linux::Event::Stream';
use Linux::Event::Framer 'Delimiter', "\n";
sub on_message ($stream, $message) {
$stream->send($message);
}Framed and raw modes are mutually exclusive. A subclass with no framer must define on_data; a framed subclass must define on_message. The base class cannot be instantiated directly.
CONSTRUCTOR
new(fh => $fh, loop => $loop, data => $value)
fh is required for an already-established Stream. Stream takes ownership of the filehandle and sets it nonblocking. Supply loop to attach before new returns, or omit it and attach with $loop->add($stream). Both forms are primary APIs and add returns the same Stream. data is optional per-connection state. Use detach to transfer a still-open plain handle back to the application; TLS Streams cannot be detached.
A TLS-declared Stream normally obtains its role from connect or Listener acceptance. Supplying an already-connected fh is ambiguous, so that advanced form also requires tls_role => 'client' or 'server'. See Linux::Event::TLS.
idle_timeout, read_timeout, and write_timeout override the subclass's cached inactivity defaults for this Stream. Each is non-negative seconds and zero explicitly disables that policy. deadline accepts a hash reference containing exactly one of after or at plus a non-empty operation label. Relative construction deadlines begin when the Stream becomes usable.
Socket policy such as tcp_nodelay, keepalive, and buffer sizes may also be supplied here for an established fh. Constructor values override the subclass policy. Settings omitted from both places leave the kernel value unchanged.
Callbacks, framing, and buffer policy are class behavior and are not accepted as constructor options.
connect(host => 'example.com', port => 443)
my $stream = MyStream->connect(
host => '127.0.0.1', # required
port => 9999, # required
timeout => 10, # default
local_host => '127.0.0.1', # optional source address
local_port => 0, # optional source port
tcp_nodelay => 1, # optional
);
$loop->add($stream);Returns one Stream that survives connection setup, optional TLS negotiation, established I/O, and close. Supply loop to start connecting immediately. Otherwise the state is unattached until $loop->add($stream).
Exactly one of host/port, unix, or packed sockaddr/family is required. timeout is seconds, defaults to 10, and may be zero to disable the deadline. data, established timeout overrides, and deadline are passed to the Stream. A class declaring Linux::Event::TLS automatically uses client TLS and defaults certificate hostname verification to host. write and send may queue output before attachment or readiness. Hostname resolution runs in the Loop's private native worker pool; socket establishment and staggered IPv6/IPv4 attempts are nonblocking.
Most outbound connections should omit local_host and local_port; Linux then chooses the source address and ephemeral source port. local_host selects a numeric IPv4 or IPv6 source address, while local_port selects the source port. They do not replace the remote host and port. bind_device optionally constrains the socket to a Linux interface before local binding and connection. It may require kernel privilege.
ATTACHMENT AND OWNERSHIP
A Stream is attached once, to one Loop. loop => $loop and $loop->add($stream) perform the same attachment. A terminal Stream cannot be reused or reattached. The Stream owns its filehandle until close, graceful completion, or detach transfers a plain handle back to the caller.
TLS DECLARATION
A subclass opts into TLS declaratively, after establishing Stream inheritance:
package SecureStream;
use parent 'Linux::Event::Stream';
use Linux::Event::TLS
ca_file => '/etc/ssl/certs/ca-certificates.crt', # optional
verify => 1, # default
alpn => ['http/1.1'], # optional
handshake_timeout => 10, # default
shutdown_timeout => 5; # defaultSecureStream->connect(host => 'example.com', port => 443) selects client TLS. A Listener that names SecureStream selects server TLS and therefore requires cert_file and key_file in the declaration. on_ready has one meaning in both roles: the plain connection or TLS handshake is ready for application traffic.
TLS is an acquisition declaration, not protocol inheritance. It creates fresh connection state only when a Stream is constructed. transition_to changes protocol callbacks and framing but never installs, removes, or replaces the active byte transport.
CLASS STREAM OPTIONS
A subclass that needs non-default Stream settings may define stream_options. It runs once when the class descriptor is built, not once per connection:
sub stream_options ($class) {
return (
read_size => 32_768, # optional
high_watermark => 2 * 1024 * 1024, # optional
low_watermark => 512 * 1024, # optional
max_pending_bytes => 8 * 1024 * 1024, # optional
max_buffer => 16 * 1024 * 1024, # optional
idle_timeout => 120, # optional; seconds
read_timeout => 30, # optional; seconds
write_timeout => 10, # optional; seconds
);
}The defaults are 65,536 bytes per read, a 1 MiB high watermark, a 256 KiB low watermark, no hard pending-output limit, and an 8 MiB maximum framed input buffer. Set max_pending_bytes to a positive byte count to impose a hard limit; zero keeps the default unlimited policy. Established timeout defaults are zero, which disables them. Constructor values take precedence for one Stream and survive protocol transitions; non-overridden values change to the target subclass's defaults.
SOCKET CONFIGURATION
Socket policy may be cached with the other stream_options:
sub stream_options ($class) {
return (
tcp_nodelay => 1, # optional
keepalive => 1, # optional
keepalive_idle => 60, # optional; seconds
keepalive_interval => 10, # optional; seconds
keepalive_count => 5, # optional
tcp_user_timeout => 15, # optional; seconds
send_buffer => 262_144, # optional; bytes
receive_buffer => 262_144, # optional; bytes
);
}The same names are accepted by new and connect for one Stream. Instance construction wins over class policy; an option omitted from both places is not set at all. This is deliberately different from inventing library defaults that overwrite Linux tuning silently.
tcp_nodelay, keepalive, keepalive_idle, keepalive_interval, keepalive_count, and tcp_user_timeout are valid only for IPv4 or IPv6 TCP sockets. Send and receive buffer sizing also applies to Unix Streams. Public timeout values are seconds; tcp_user_timeout is converted to the Linux millisecond value with a positive sub-millisecond duration rounded up.
For outbound sockets, built-in policy and the optional hook below run after socket and before local bind or remote connect. Accepted and adopted sockets apply policy before transport setup, including a TLS handshake.
An advanced subclass may define one cached socket hook:
use Socket qw(IPPROTO_TCP TCP_QUICKACK);
sub configure_socket ($stream, $fh, $role, $address) {
setsockopt($fh, IPPROTO_TCP, TCP_QUICKACK, pack('i', 1))
or die "setsockopt(TCP_QUICKACK): $!";
}$role is connect, accepted, or adopted. $address is the remote candidate or peer when one is available. A hook exception becomes a structured socket_configuration Error. It never falls back silently to another configuration.
These options are also live getters/setters on an established Stream:
my $enabled = $stream->tcp_nodelay; # current kernel value
$stream->tcp_nodelay(1); # enable and return effective value
my $bytes = $stream->send_buffer(262_144);The complete live set is tcp_nodelay, keepalive, keepalive_idle, keepalive_interval, keepalive_count, tcp_user_timeout, send_buffer, and receive_buffer. Linux may round or double buffer requests, so setters return the value read back from the kernel. Socket policy is acquisition policy; transition_to does not reapply it.
CALLBACKS
Subclasses may define these ordinary named methods:
sub on_data ($stream, $bytes) { # required for raw Stream
$stream->write($bytes);
}
sub on_message ($stream, $message) { # required for framed Stream
$stream->send($message);
}
sub on_drain ($stream) { # optional
$stream->data->{blocked} = 0 if $stream->data;
}
sub on_eof ($stream) { $stream->end } # optional
sub on_error ($stream, $error) { # optional
warn "$error\n";
}
sub on_close ($stream) { # optional
$stream->data->{closed} = 1 if $stream->data;
}
sub on_ready ($stream) { # optional
$stream->write("ready\n");
}
sub on_transport_ready ($stream) { # optional
say "transport " . $stream->transport_name . " is ready";
}The resolved CVs are cached and invoked directly; readiness dispatch does not perform Perl method lookup. Inheritance works normally, so a derived Stream type may reuse callbacks and framing from its parent. Per-user or per-connection permissions belong in data, which callbacks access through $stream->data. on_ready is called once when an asynchronously acquired Stream becomes usable. For TLS, this means handshake and verification have completed. Accepted Streams are ready after Listener attachment. on_transport_ready is the lower-level provider-ready notification and runs immediately before on_ready for an asynchronous non-plain transport. Most applications need only on_ready.
Application callback exceptions are not swallowed.
METHODS
write($bytes)
Writes immediately when possible and queues any remainder. Returns false after queued bytes exceed the high watermark; the bytes were still accepted. Wait for on_drain before producing more.
$bytes must be a byte string. Encode character text before writing it.
When max_pending_bytes is nonzero, a write whose unsent remainder would put the native queue above that limit is not queued. Stream reports an output_limit Linux::Event::Error through on_error and closes. The error's pending_bytes and limit accessors describe the attempted queue size and configured bound. An immediate kernel write may already have sent a prefix before its remainder is found to exceed the limit; Stream never adds that remainder to its queue. The ordinary false return remains reserved for accepted cooperative backpressure.
send($payload)
Available only to framed subclasses. Applies the subclass's declared outbound wire framing and then uses write. Serialization remains separate.
pause_read / resume_read
Disable and re-enable input readiness without destroying the Stream.
transition_to($class, input => $bytes)
Changes a live connection to another loaded Linux::Event::Stream subclass. The same object is reblessed into $class, and the same filehandle, native registration, native connection state, output queue, backpressure state, lifecycle state, and data are retained. Future callbacks, send framing, parser rules, and class Stream policy come from the target subclass's cached descriptor.
Unread bytes already held by a framed parser are preserved and reinterpreted by the target parser. A raw on_data callback may pass the unconsumed suffix of its current chunk with input => $bytes:
sub on_data ($stream, $bytes) {
my ($request, $remaining) = parse_upgrade($bytes);
return if !$request;
$stream->write(upgrade_response());
$stream->transition_to(
'My::WebSocketStream',
input => $remaining, # optional raw unconsumed suffix
);
return;
}Existing native input is ordered before the explicit input suffix. Complete target frames may be delivered before transition_to returns when the method is called outside input dispatch. During on_data or on_message, target dispatch begins after the old callback returns. Code should normally return immediately after requesting a transition.
Read pause is retained and continues to gate preserved input. Queued output keeps its original byte ordering; only later send calls use the new outbound framing. If preserved input exceeds the target class's max_buffer, or existing queued output exceeds its nonzero max_pending_bytes, the transition fails atomically and the old type remains active. Transitioning to the already-active class is rejected.
This method changes Stream protocol behavior; it does not replace the active byte transport. In particular, a TLS Stream remains TLS across protocol transitions.
end($final_bytes = undef)
Drains queued output and ends the transport's writable side. Plain Streams use shutdown(SHUT_WR); TLS providers send close_notify. Peer EOF and the local writable half-close remain independent.
close
Immediately cancels native readiness and closes the owned descriptor. Queued output may be lost. Returns the Stream.
detach
Cancels Stream ownership and returns the still-open filehandle for the plain transport. on_close is not called because the underlying resource remains open. Non-plain transports reject detach because the descriptor carries provider-owned wire state rather than application plaintext.
pending_bytes / is_write_blocked
Report native output-queue and flow-control state.
idle_timeout / read_timeout / write_timeout
Return this Stream's effective established inactivity policy in seconds. idle_timeout resets on successful read or write transport progress. read_timeout resets on inbound bytes, is suspended by pause_read, and starts a fresh interval on resume_read. write_timeout exists only while output remains queued and resets on successful write progress.
set_deadline(after => 5, operation => 'response')
Set or replace the one explicit overall-operation deadline and return the Stream. Supply after => $seconds or at => $monotonic_deadline, together with operation => $name. An overall deadline never resets because of I/O. Calling this before establishment stores the policy; relative time begins when the Stream becomes usable, while at remains an absolute CLOCK_MONOTONIC value.
clear_deadline
Remove the explicit operation deadline and return the Stream. Inactivity policies remain active.
deadline / deadline_operation
Return the active absolute operation deadline and its label. A detached relative deadline has no absolute value yet, so deadline returns undef.
All established deadline categories start after plain or TLS transport readiness. Resolver, connection, TLS handshake, and TLS shutdown time use their existing separate owners. Expiration reports a timeout Linux::Event::Error through on_error and closes normally through on_close. The Error includes timeout and deadline context. Every deadline-enabled Stream uses at most one private Timer in the Loop's shared timerfd/native heap.
transport / transport_name / is_transport_ready
Returns the configured provider object, active native byte transport name, and whether its asynchronous setup has completed. Ordinary filehandle-backed Streams have no provider object, report plain, and are immediately ready.
selected_alpn / tls_protocol / tls_cipher / tls_stats
Return negotiated ALPN, TLS protocol, cipher, and native TLS counters for a TLS Stream. The scalar information methods return undef for a plain Stream; tls_stats returns undef when no TLS provider is active.
tcp_nodelay / keepalive / keepalive_idle / keepalive_interval / keepalive_count / tcp_user_timeout / send_buffer / receive_buffer
With no argument, return the effective Linux socket value. With one argument, set the option and return the value read back from the kernel. These methods require an established socket; TCP-only methods reject a Unix Stream.
is_read_paused / is_read_eof / is_write_ended / is_closed
Report Stream lifecycle state.
data([$value])
Gets or replaces per-connection application state.
loop / state / peer / local
Return the owning Loop, unattached, connecting, active, detached, or closed lifecycle state, the lazy remote peer when available, and the local socket address. Outbound and adopted Streams populate both addresses when the kernel supplies them; accepted Streams receive their peer from Listener.
last_error
Returns the Linux::Event::Error that caused closure, or undef when the Stream has not failed.
FRAMING POLICY
Framed Stream types use native built-ins declared through Linux::Event::Framer. Arbitrary per-connection framer objects and the old custom Perl next_frame contract are intentionally unsupported. Unusual protocols can buffer and parse raw on_data bytes. Generally useful framing families should be implemented as native Linux::Event built-ins.
PERFORMANCE
Native code drains reads, detects built-in frame boundaries, performs immediate writes, drains segmented queues with writev, and accounts for backpressure. The class descriptor moves immutable callbacks and parser configuration out of each connection. Perl is entered for semantic on_data or on_message delivery and lifecycle policy.