magmalake flare_mlake

flare_mlake · API · tcp

flare.tcp.stream

TCP stream: a connected, bidirectional byte stream.

TcpStream owns a RawSocket and exposes blocking read/write operations. TCP_NODELAY is set automatically on every new stream so callers get low-latency behaviour by default (Nagle's algorithm is disabled).

Design rules enforced here:

  • write retries transparently on EINTR.
  • write_all loops until every byte is sent.
  • read returning 0 means EOF — it is never converted to an error.
  • connect_timeout on macOS/arm64 delegates to flare_connect_timeout in libflare_tls.so to sidestep a Mojo ABI bug with variadic functions (e.g. fcntl) on that platform. On Linux, connect_timeout calls fcntl / connect / poll / getsockopt directly via external_call — OwnedDLHandle.get_function crashes on Linux when calling into a freshly-loaded shared library.

Structs

TcpStream

struct TcpStream

A connected TCP socket.

Owns a RawSocket and exposes blocking read/write operations. TCP_NODELAY is set automatically on every new stream so callers get low-latency behaviour without any configuration.

This type is Movable but not Copyable — a TCP connection cannot be duplicated without tracking both file descriptors.

The connection is closed automatically when the struct is destroyed.

Thread safety: Not thread-safe . Do not share a TcpStream across threads without external synchronisation.

Example:

from flare.tcp import TcpStream
from flare.net import SocketAddr

var stream = TcpStream.connect(SocketAddr.localhost(8080))
stream.write_all("hello".as_bytes())
var buf = List[UInt8](capacity=1024)
buf.resize(1024, 0)
var n = stream.read(buf.unsafe_ptr(), len(buf))
stream.close()

Implements Movable, Readable, _ChunkWriter

__init__

def __init__(out self, var socket: RawSocket, peer: SocketAddr)

Wrap an already-connected RawSocket.

Safety: socket must be connected before calling this constructor. TCP_NODELAY should already be set by the factory functions.

Args
socket: An open, connected file descriptor (ownership transferred).
peer: The remote address.

__deinit__

def __deinit__(deinit self)

connect

def connect(addr: SocketAddr) -> Self

Open a blocking TCP connection to addr.

Sets TCP_NODELAY automatically.

Example:

var stream = TcpStream.connect(SocketAddr.localhost(8080))
Args
addr: The remote socket address to connect to.

Raises ConnectionRefused: If the remote port actively refuses. ConnectionTimeout: If the OS connect timeout expires. ConnectionReset: If the peer sends a TCP RST during connect. NetworkError: For any other OS error.

def connect(host: String, port: UInt16) -> Self

Resolve host via DNS and open a blocking TCP connection.

Convenience overload that performs DNS resolution then calls TcpStream.connect(SocketAddr(...)) automatically.

Example:

var stream = TcpStream.connect("example.com", 80)
Args
host: Hostname or IP address string (e.g. "example.com").
port: TCP port number.

Raises DnsError: If host cannot be resolved. ConnectionRefused: If the remote port actively refuses. ConnectionTimeout: If the OS connect timeout expires. NetworkError: For any other OS error.

def connect(host: String, port: UInt16, timeout_ms: Int) -> Self

Resolve host via DNS (dual-stack) and connect with a timeout.

Tries each resolved address in order (IPv4 and IPv6), falling back to the next on failure.

Args
host: Hostname or IP address string.
port: TCP port number.
timeout_ms: Maximum time to wait in milliseconds.

Raises DnsError: If host cannot be resolved. ConnectionTimeout: If all addresses time out. ConnectionRefused: If all addresses refuse. NetworkError: For any other OS error.

connect_timeout

def connect_timeout(addr: SocketAddr, timeout_ms: Int) -> Self

Open a TCP connection, failing if it takes longer than timeout_ms.

On macOS/arm64 this delegates to flare_connect_timeout in libflare_tls.so to avoid a Mojo ABI bug where external_call silently corrupts variadic-function arguments (e.g. the third argument of fcntl) on that platform.

On Linux, the non-blocking connect + poll(POLLOUT) + getsockopt(SO_ERROR) sequence is implemented directly via external_call. OwnedDLHandle.get_function crashes on Linux when calling into a freshly-loaded shared library, so the C helper cannot be used there.

Example:

var stream = TcpStream.connect_timeout(SocketAddr.localhost(8080), 5000)
Args
addr: The remote socket address.
timeout_ms: Maximum time to wait in milliseconds (must be > 0).

Raises ConnectionTimeout: If the deadline expires before connecting. ConnectionRefused: If the port actively refuses. NetworkError: For any other OS error.

__enter__

def __enter__(var self) -> Self

Transfer ownership of self into the with block.

read

def read(mut self, buf: Pointer[UInt8], size: Int) -> Int

Read up to size bytes from the stream into buf.

Retries transparently on EINTR. Returns 0 on EOF (the peer closed the connection) — this is not an error.

Example:

var buf = List[UInt8](capacity=4096)
buf.resize(4096, 0)
var n = stream.read(buf.unsafe_ptr(), len(buf))
if n == 0:
    print("connection closed")
Args
buf: Destination buffer; the caller must provide at least size bytes of valid storage.
size: Maximum number of bytes to read.

Raises ConnectionReset: If the peer sent a TCP RST. Timeout: If a recv timeout was set and expired. NetworkError: For any other OS read error.

read_exact

def read_exact(mut self, buf: Pointer[UInt8], size: Int)

Read exactly size bytes into buf.

Loops over read() until size bytes have been received.

Example:

var exact = List[UInt8](capacity=8)
exact.resize(8, 0)
stream.read_exact(exact.unsafe_ptr(), 8)
Args
buf: Destination buffer; must have at least size bytes.
size: Exact number of bytes to read.

Raises NetworkError: If EOF is reached before size bytes are received, or on any other I/O error.

write

def write(self, data: Span[UInt8]) -> Int

Write up to len(data) bytes to the stream.

A single call may write fewer bytes than requested (partial write). Use write_all to guarantee all bytes are sent. Retries on EINTR.

Example:

var n = stream.write("hello".as_bytes())
Args
data: The bytes to send.

Raises BrokenPipe: If the peer has closed the read end. Timeout: If a send timeout was set and expired. NetworkError: For any other OS write error.

write_all

def write_all(self, data: Span[UInt8])

Write all of data, looping until every byte is sent.

Example:

stream.write_all("Hello, server!".as_bytes())
Args
data: The bytes to send completely.

Raises BrokenPipe: If the peer closes before all bytes are sent. NetworkError: For any other OS write error, or if send(2) returns 0 for a non-empty buffer (no progress is possible).

peer_addr

def peer_addr(self) -> SocketAddr

Return the remote socket address.

local_addr

def local_addr(self) -> SocketAddr

Return the local socket address assigned by the OS.

Raises NetworkError: If getsockname(2) fails.

shutdown_read

def shutdown_read(self)

Shut down the read half of the connection.

After this call, any subsequent read() returns 0 (EOF). The peer will receive a FIN indicating no more data will be read.

Raises NetworkError: If shutdown(2) fails.

shutdown_write

def shutdown_write(self)

Shut down the write half of the connection.

Sends a FIN to the peer, signalling that no more data will be written. The peer will see EOF on their next read().

Raises NetworkError: If shutdown(2) fails.

close

def close(mut self)

Close the connection explicitly. Idempotent.

set_nodelay

def set_nodelay(self, enabled: Bool)

Toggle TCP_NODELAY (Nagle's algorithm).

Args
enabled: True disables Nagle (low-latency); False re-enables it.

set_keepalive

def set_keepalive(self, enabled: Bool)

Toggle SO_KEEPALIVE.

Args
enabled: True to send keepalive probes on idle connections.

set_recv_timeout

def set_recv_timeout(self, ms: Int)

Set a per-read timeout.

After the timeout expires, read() raises Timeout.

Args
ms: Timeout in milliseconds. 0 disables the timeout.

set_send_timeout

def set_send_timeout(self, ms: Int)

Set a per-write timeout.

After the timeout expires, write() raises Timeout.

Args
ms: Timeout in milliseconds. 0 disables the timeout.