/ 
Linux-Event-0.115
/ Linux::Event::IO::Sock::Stream

NAME

Linux::Event::IO::Sock::Stream - asynchronous Linux SOCK_STREAM connections

SYNOPSIS

use v5.36;
use Linux::Event::Loop;
use Linux::Event::IO::Sock::Listener;
use Linux::Event::IO::Sock::Stream;

my $loop = Linux::Event::Loop->new;
my $server = Linux::Event::IO::Sock::Listener->new(
    loop => $loop,
    host => '127.0.0.1',
    port => 0,
    stream => {
        on_data => sub ($stream, $bytes) {
            $stream->write($bytes);
        },
    },
);

my $prefix = 'received';
my $client = Linux::Event::IO::Sock::Stream->connect(
    loop    => $loop,
    host    => '127.0.0.1',
    port    => $server->port,
    on_ready => sub ($stream) {
        $stream->write('hello');
    },
    on_data => sub ($stream, $bytes) {
        say "$prefix: $bytes";
        $stream->close;
        $server->close;
        $loop->stop;
    },
    on_error => sub ($stream, $error) {
        die "connection failed: $error\n";
    },
);

$loop->run;

DESCRIPTION

Linux::Event::IO::Sock::Stream is the public class for connected Linux SOCK_STREAM sockets. TCP over IPv4 or IPv6 and Unix-domain stream sockets use the same class; address family is connection configuration rather than a separate type hierarchy.

The class combines the common ordered-byte engine with socket acquisition, addresses, socket policy, kernel half-close semantics, and optional TLS. A concrete protocol subclass can supply named callbacks and declare class policy; constructor callbacks are an equally supported way to provide application behavior with normal Perl lexical scope.

CALLBACKS, SUBCLASSING, AND TUNING

Constructor callbacks make the public Stream leaf directly useful and preserve ordinary lexical scope. Subclassing remains one of Linux::Event's important distinguishing features because a protocol class can declare, once:

  • a native Linux::Event::Framer and its wire format;

  • reusable TLS defaults such as ALPN and transport timeouts, without making TLS part of the Stream class identity;

  • stream_tuning tuning for reads, fairness, batching, buffers, watermarks, limits, and established deadlines; and

  • socket policy and named, reusable callbacks.

Class policy and method callbacks are validated and cached once per subclass. A constructor callback overrides a same-named method for one connection and is retained once in that object's effective descriptor. This makes it natural to combine reusable high-performance protocol policy with per-connection lexical state without adding event-time method lookup or callback-style selection.

A Stream subclass is also an ordinary Perl class and may initialize and expose its own instance variables. Linux::Event does not interpret or manage subclass-owned state, and no separate state or initialization hook is required:

package StatefulConnection;
use parent 'Linux::Event::IO::Sock::Stream';

sub new ($class, %option) {
    my $self = $class->SUPER::new(%option);
    $self->{message_count} = 0;
    return $self;
}

sub message_count ($self, @value) {
    $self->{message_count} = $value[0] if @value;
    return $self->{message_count};
}

sub on_data ($self, $bytes) {
    $self->{message_count}++;
    ...;
}

Core operations leave unrelated subclass-owned entries alone. Subclass constructors remain responsible for their own state and should pass only Linux::Event constructor options to SUPER::new.

stream_tuning

Define stream_tuning as a class method on the Stream subclass. It returns key/value pairs, or one hash reference:

package TunedConnection;
use parent 'Linux::Event::IO::Sock::Stream';

sub stream_tuning ($class) {
    return (
        read_size         => 131_072,
        read_budget_bytes => 524_288,
        high_watermark    => 2_097_152,
        low_watermark     => 524_288,
        idle_timeout      => 60,
    );
}

These options also apply to Pipe and TTY subclasses. The complete Stream option set is:

  • read_size (default 65,536)

    Maximum bytes requested by one native read; a positive integer.

  • read_budget_bytes (default 65_536)

    Maximum bytes read during one readiness drain. The default bounds one readiness callback to 65,536 bytes so other Loop resources can run. Zero is an explicit opt-in to drain until the socket would block.

  • read_batch_bytes (default 0)

    For an unframed class, combine successful reads before on_data up to this non-negative byte target. Partial batches flush when the current drain ends; zero preserves normal read callback boundaries. It is invalid with framing.

  • message_batch_size (default 0)

    For a framed class, deliver arrays of at most this many messages to on_messages. Partial batches flush when the current drain ends; zero uses on_message. A positive value requires on_messages and framing.

  • max_buffer (default 8,388,608)

    Positive hard byte bound for retained input, an incomplete frame, and the aggregate payload retained for one message batch.

  • high_watermark (default 1,048,576)

    Non-negative pending-output byte level at which write or send begins returning false while still accepting the data.

  • low_watermark (default 262,144)

    Non-negative pending-output byte level at or below which on_drain fires after high-watermark backpressure. It must not exceed high_watermark.

  • max_pending_bytes (default 0)

    Hard non-negative pending-output byte limit. Zero means unbounded.

  • idle_timeout (default 0 seconds)

    Maximum inactivity interval since successful established input or output progress. Zero disables it.

  • read_timeout (default 0 seconds)

    Maximum interval without inbound progress while reading is active. Pausing input suspends it; zero disables it.

  • write_timeout (default 0 seconds)

    Maximum interval without output progress while data is queued. Zero disables it.

Byte counts are integers. Timeout values are finite non-negative seconds and may be fractional.

tune

tune changes the mutable ordered-byte policy of an existing Stream without reconstructing it:

$stream->tune(
    read_size         => 131_072,
    read_budget_bytes => 524_288,
    high_watermark    => 2_097_152,
    low_watermark     => 524_288,
    idle_timeout      => 30,
);

The supported keys are the same eleven values documented by stream_tuning: read_size, read_budget_bytes, read_batch_bytes, message_batch_size, high_watermark, low_watermark, max_pending_bytes, max_buffer, idle_timeout, read_timeout, and write_timeout.

Effective precedence is class stream_tuning() defaults, then Listener stream => { tuning => {...} } deployment overrides for accepted connections, then tune() on the live object.

Mutable values are copied into native per-Stream state when policy changes. Ordinary reads and writes do not consult Perl hashes or perform class-versus- instance resolution. Changing message batching settles work owned by the old batch policy first. Watermark changes immediately reconcile backpressure. Lowering max_pending_bytes or max_buffer does not discard bytes already queued or buffered; later growth must satisfy the new limit. Timeout changes re-arm or cancel established deadline state as needed.

Framer identity, callback structure, native-consumer identity, and transport kind are not tune() values. tune() returns the Stream and rejects calls on a closed Stream.

socket_options

Define socket_options as another class method on a Stream subclass. It also returns key/value pairs or one hash reference:

sub socket_options ($class) {
    return (
        tcp_nodelay      => 1,
        keepalive        => 1,
        tcp_user_timeout => 15,
    );
}

Unspecified options retain kernel defaults. The complete set is:

  • tcp_nodelay

    Boolean 0 or 1 controlling TCP_NODELAY; TCP only.

  • keepalive

    Boolean 0 or 1 controlling SO_KEEPALIVE; TCP only.

  • keepalive_idle

    Positive integer seconds before the first TCP keepalive probe.

  • keepalive_interval

    Positive integer seconds between TCP keepalive probes.

  • keepalive_count

    Positive integer number of failed TCP keepalive probes allowed.

  • tcp_user_timeout

    Finite non-negative seconds for TCP_USER_TIMEOUT; fractional values are rounded up to milliseconds. TCP only.

  • send_buffer

    Positive integer requested SO_SNDBUF size.

  • receive_buffer

    Positive integer requested SO_RCVBUF size.

Positive socket integers are at most 2,147,483,647. Constructor values override class policy for one connection. bind_device is a constructor option, not a socket_options key. configure_socket is the cached cold-path hook for Linux options not covered above.

OUTBOUND CONNECTIONS

connect constructs one connection object whose identity is retained through resolution, connection, optional TLS handshake, established I/O, and close:

my $stream = Client->connect(
    loop    => $loop,          # optional immediate attachment
    host    => 'example.com',  # TCP remote host
    port    => 443,            # TCP remote port
    timeout => 10,             # connection deadline; default 10
    data    => $state,         # optional application state
);

Use unix => $path for a filesystem Unix-domain stream socket. Advanced callers may supply a packed sockaddr with its numeric family.

loop is optional. Without it, connect returns a detached object that may later be passed to $loop->add($stream). Writes submitted before readiness use the normal bounded output queue and are delivered in order after the transport becomes usable.

Optional source-side controls include numeric local_host, local_port, and bind_device. Hostname resolution is asynchronous and uses the Loop's private native resolver service.

ADOPTED CONNECTED SOCKETS

new(fh => $socket) adopts an already connected SOCK_STREAM handle. The handle is validated, made nonblocking and close-on-exec, and uses the same established I/O path as an accepted or outbound connection. A TLS-declared class must also specify tls_role for an adopted handle because acquisition cannot infer client versus server role.

CALLBACKS

Callbacks may be methods, constructor coderefs, or a mixture:

my $database = ...;
my $stream = RawConnection->new(
    fh      => $socket,
    on_data => sub ($stream, $bytes) {
        process_bytes($database, $stream, $bytes);
    },
);

A constructor callback overrides the corresponding class method for that object. Supported names and signatures are on_data($stream, $bytes), on_message($stream, $message), on_messages($stream, $messages), on_ready($stream), on_transport_ready($stream), on_drain($stream), on_eof($stream), on_error($stream, $error), and on_close($stream). connect accepts the same callback options as new.

on_ready($stream) runs once when an outbound or accepted connection becomes application-ready. For TLS that means after handshake and verification, not merely after TCP connect. new(fh => ...) adopts a connection that is already ready and does not emit a later readiness callback.

on_transport_ready($stream) is the lower transport notification used by TLS or another native transport and runs immediately before on_ready. Plain connections have no separate transport phase.

A raw object requires on_data($stream, $bytes) as a method or constructor callback. The public Stream leaf can therefore be constructed directly for raw I/O. A framed class uses Linux::Event::Framer and requires on_message or, with explicit batching, on_messages; either may be supplied by the class or constructor.

Optional lifecycle callbacks include on_drain, on_eof, on_error, on_close, and on_transport_ready for transport-specific observation. Method defaults are resolved into an immutable class descriptor. Constructor input callbacks are retained once in native Stream state, producing one effective cached CV with no event-time lookup or method-versus-coderef branch. Lifecycle callbacks are likewise resolved once during construction. Closing or detaching the Stream releases its retained constructor callbacks.

FRAMING AND OUTPUT

write($bytes) sends raw ordered bytes. send($payload) applies the subclass's native framer. The native write engine attempts immediate output, queues only unsent bytes, enables writable readiness only while necessary, and uses high/low watermarks plus optional max_pending_bytes protection.

pause_read and resume_read control application reads. transition_to changes protocol callback/framing descriptors in place while retaining the live socket, transport, output queue, and unread native input according to the transition rules in docs/FRAMING.md. Native protocol extensions may also hand off from one native consumer provider to another without copying the retained input through Perl; adding or removing native-consumer mode during a live transition remains invalid.

SOCKET POLICY

A subclass may define socket_options for acquisition-time socket policy. The method shape and complete option contract appear near the top of this document. See docs/SOCKET-CONFIGURATION.md for application order and failure behavior.

ORDERED-BYTE POLICY AND DEADLINES

stream_tuning has the complete option contract listed near the top of this document. One explicit operation deadline may also be set or changed at runtime. Established timeout policy begins when the application transport is usable; DNS, connect, TLS handshake, and TLS shutdown retain separate lifecycle deadlines.

TLS

TLS is acquisition policy for a Stream socket rather than a separate Stream class identity. For accepted connections a Listener selects TLS in its generated Stream recipe:

my $listener = Linux::Event::IO::Sock::Listener->new(
    loop => $loop,
    host => '0.0.0.0',
    port => 9443,
    stream => {
        class => 'ServerConnection',
        tls => {
            cert_file => $cert_file,
            key_file  => $key_file,
            alpn      => ['my-protocol/1'],
        },
    },
);

The same ServerConnection class may be used by another Listener without a tls recipe and is then plain. A subclass may define tls_defaults() for reusable policy such as ALPN and handshake/shutdown timeouts, but those defaults do not activate TLS. The Listener prepares one reusable server context and each accepted TLS Stream receives independent connection state.

Existing class-level Linux::Event::TLS declarations remain available for explicit outbound client policy and adopted-handle compatibility. Outbound connect derives the default server name from host. Framing and callbacks always receive plaintext. See Linux::Event::TLS.

ADDRESSES AND LIFECYCLE

local and peer return lazy Linux::Event::Address values when available. fd, fh, state, pending_bytes, and last_error expose connection state without changing ownership.

end drains accepted output then performs the transport's writable half-close. close is immediate and terminal. detach transfers a plain connected socket only when no output is pending; encrypted transports cannot be detached safely.

SEE ALSO

Linux::Event::IO::Sock::Listener, Linux::Event::IO::Sock::Dgram, Linux::Event::Framer, Linux::Event::TLS, docs/SOCKET-CONNECTIONS.md, docs/ORDERED-BYTE-IO-DESIGN.md, docs/FIRST-CLASS-STREAM-CALLBACKS.md.