SocketFlow

A high-performance, dependency-free TCP networking library for Python with advanced features like compression, event handling, bidirectional keepalive, and more.

Features

• Zero Dependencies - Uses only Python's standard library
• Bidirectional Keepalive - Both client and server independently monitor connection health
• TCP-Level Keepalive - OS-managed keepalive for reliable connection detection
• Compression - Support for zlib, lzma, and bz2 compression
• Event-Driven Architecture - Flexible event dispatcher for handling server/client events
• Blueprint System - Organize your code with reusable blueprints
• Middleware Support - Add custom middleware to request/response processing
• Path-Based Routing - Route messages to specific handlers using paths
• Per-Path Serialisation - block=True runs a path one message at a time per client, for handlers that must not interleave
• Status Codes - Attach HTTP-style status codes to replies, defaulting to 200
• Automatic Error Reporting - Unknown paths return 404, handler crashes return 500
• Efficient Buffer Handling - O(N) buffer processing with offset pattern
• Type Hints - Full type annotations for better IDE support
• Cross-Platform - Works on Windows, Linux, and macOS

Installation

pip install socketflow

Full documentation available at: https://socketflow.dev/

Or install from source:

git clone https://github.com/ayammaximilian/socketflow.git
cd socketflow
pip install .

Quick Start

Server Example

from socketflow import TcpServer, EventType

# Create server
server = TcpServer(
    host="127.0.0.1",
    port=8080,
    keepalive_interval=30.0,
    keepalive_max_missed=3,
    compress=True
)

# Register event handler
@server.event(EventType.Server.MESSAGE)
def handle_message(data):
    print(f"Received: {data.data}")
    # Handlers do NOT reply by returning. Send the reply explicitly,
    # echoing the data_id so the caller can match it to its request.
    server.send_client(data.client_addr, "Response", data.data_id)

# Start server
server.start()
server.wait()  # Keep server running

Client Example

from socketflow import TcpClient, EventType

# Create client
client = TcpClient(
    host="127.0.0.1",
    port=8080,
    keepalive_interval=30.0,
    keepalive_max_missed=3,
    compress=True
)

# Register event handler BEFORE connecting
@client.event(EventType.Client.MESSAGE)
def handle_message(data):
    # Only reached for messages sent without wait_response.
    print(f"Received: {data.data}")

# Connect to server
client.connect()

# Send message and wait for the reply
response = client.send("Hello, Server!", wait_response=True, wait_response_timeout=5)
print(f"Server response: {response.data}")

# Disconnect
client.disconnect()

Output:

Received: Hello, Server!
Server response: Response

How replies work

This trips people up, so it is worth being explicit:

You want	Do this
Server answers a request	server.send_client(data.client_addr, reply, data.data_id)
Client waits for the answer	client.send(msg, wait_response=True)
Fire and forget, no answer	client.send(msg)
Client handles messages on its own	@client.event(EventType.Client.MESSAGE)
Reply with a status code	add status_code=404 to send_client
Check the outcome on the client	reply.status_code (see below)

Three things to remember:

• Returning a value from a handler does nothing. Handlers must call send_client (or send_client_async) to reply.
• A reply that matches a pending wait_response request never reaches Client.MESSAGE. It resolves the send() call instead. Your event handler only sees messages that nobody was waiting for.

If the server never replies, wait_response=True raises NoResponse once wait_response_timeout expires rather than hanging forever. Always pass a timeout.

Status Codes and Error Handling

Every reply carries a status code, the same idea as an HTTP status. It defaults to 200, so nothing changes unless you set one:

@server.path("sf_download_size")
def get_size(data):
    path = data.data["file_path"]
    if not os.path.exists(path):
        return server.send_client(
            data.client_addr,
            {"error": "Not Found", "detail": "no such file"},
            data.data_id,
            status_code=404,
        )
    return server.send_client(
        data.client_addr, {"size": os.path.getsize(path)}, data.data_id
    )

The caller reads it off the reply:

reply = client.send(
    {"file_path": file_path},
    path="sf_download_size",
    wait_response=True,
    wait_response_timeout=timeout,
)

if reply.status_code == 404:
    return False, reply.data["detail"]
return int(reply.data["size"])

send_async works the same way: request.result().status_code. Handlers receive incoming messages with data.status_code, and a non-waiting message still carries it.

Two details worth knowing:

• A 200 is never written to the wire. The code is only attached to the frame when it differs from the default, so normal traffic stays byte-for-byte unchanged and older peers keep working. They simply never send a status code, and always appear as 200.
• A non-2xx code is not an error. The exchange completes normally, .data and .data_id are intact, and nothing is raised. You decide what a code means.

Automatic error reporting

You do not have to catch anything to get the common failures reported back. SocketFlow handles these itself:

Situation	Status	data
No handler registered for the path	404	{"error": "Not Found", "detail": "..."}
A handler or middleware raised	500	{"error": "Internal Server Error", "detail": "ValueError: ..."}
Handler replied, then raised	the first reply is kept	—
Anything else	200 unless you set one	your payload

So this needs no try/except to be useful:

@server.path("sf_download_size")
def get_size(data):
    raise Exception("fake error")   # client gets 500 with the reason

The client stops waiting immediately instead of hanging until the timeout, and the real cause arrives instead of a generic "no response". This is a big difference from an unanswered request, which still raises NoResponse on timeout.

Log both cases centrally with one handler:

@server.event(EventType.Global.ERROR)
def on_error(data):
    print(f"{data.context}: {data.error}")

data.error is the original exception, or a PathNotFound for an unknown path.

Notes:

• Automatic replies are only sent when the request used wait_response=True, since otherwise nobody is waiting. A fire-and-forget message to an unknown path is still logged through EventType.Global.ERROR.
• A reply you already sent wins, so a handler that replies and then raises will not produce a confusing double answer.
• A missing path is reported fast. If a client requests a path the server does not implement, it fails on the first attempt instead of waiting out the full timeout.

Non-Blocking Request/Reply

send_async returns a request handle immediately instead of blocking the calling thread:

request = client.send_async("Hello, Server!", path="echo", timeout=5.0)
print("Sent, not waiting")

# Check later
if request.done():
    print(request.result())

# Or wait only as long as you want
try:
    response = request.result(timeout=2)
except NoResponse:
    print("Timed out")

Handles also support callbacks and cancellation:

def on_reply(handle):
    print("Reply:", handle.result().data)

request = client.send_async("Hello!", path="echo")
request.add_done_callback(on_reply)

# Give up on a request that is no longer needed
request.cancel()

Handle methods: done(), result(timeout=None), exception(timeout=None), cancel(), cancelled(), add_done_callback(fn), and the data_id attribute.

The server works the same way:

request = server.send_client_async(client_addr, "push", path="push")
response = request.result(timeout=5)

Notes:

• send(..., wait_response=True) still works and is now built on the same handle.
• Timeouts run on one shared timer thread per client/server, not one thread per request.
• Cancelling or timing out removes the request from tracking immediately.

Pending request tracking

Every in-flight request is held in pending_responses until it completes. The default timeout=30.0 is what keeps this bounded: when it expires, the request is removed automatically even if no reply ever arrives.

request = client.send_async("hello", path="echo", timeout=30.0)

Use timeout=None only when a reply is genuinely optional. In that case the request is kept until the connection closes, so len(client.pending_responses) grows with every unanswered request:

request = client.send_async("fire-and-forget", path="notify", timeout=None)
print(len(client.pending_responses))  # grows while replies are missing

To keep a hard ceiling, cancel explicitly or check the count before sending.

Installation

SocketFlow has no required dependencies. It runs on the Python standard library alone, so installing it never pulls anything else in.

pip install socketflow

Optional compression codecs

Four codecs are built in: zlib, lzma, bz2, and gzip. Two more are available through extras:

pip install socketflow[zstd]      # adds zstandard
pip install socketflow[brotli]    # adds Brotli
pip install socketflow[all]       # adds both

Then use them like any other codec:

server = TcpServer(compression_type="zstd")

If a codec is not installed, you get a message telling you exactly what to do:

Compression method 'zstd' is not available because 'zstandard' is not
installed. Install it with 'pip install socketflow[zstd]', or choose one of:
lzma, bz2, zlib, gzip.

Check what your machine can do:

from socketflow.global_side.compression import MultiCompressor
print(MultiCompressor.available_methods())

Logging and Metrics

Two tools for seeing what your server is doing. Both are off by default, so adding them changes nothing until you ask for them.

Logging

Instead of reading log lines and guessing, each entry carries labeled fields.

from socketflow import logs

logs.configure(level="INFO")              # human-readable, to stderr
logs.configure(level="INFO", json_output=True)   # one JSON object per line

Sample output:

{"time": "2026-09-26T02:00:08.616Z", "level": "INFO", "logger": "socketflow.server",
 "message": "client connected", "client_identity": "test-client", "active_clients": 1}

Send logs somewhere else by giving it a sink:

logs.configure(level="DEBUG", sinks=[logs.MemorySink(limit=500)])
logs.configure(level="DEBUG", sinks=[logs.CallbackSink(my_function)])
logs.configure(level="DEBUG", sinks=[])   # silence completely

Level	Shows
DEBUG	Every message, sent and received
INFO	Connections, server start/stop, drains
WARNING	Rejected connections, disconnects
ERROR	Failures

Write your own logs the same way:

log = logs.get_logger("my.app").bind(service="billing")
log.info("charge accepted", order_id=123)   # every line now has service=billing

Long messages

message is always written out in full. A 50,000-character message is logged as 50,000 characters — it is not silently shortened, and JSON output stays valid.

To stop one huge value from flooding a sink, set a limit:

logs.configure(level="INFO", json_output=True, max_message_length=2000)

Anything longer is cut and marked:

"message": "A very long value ... [truncated 48213 chars]"

The limit applies to message and to any string field. It is off by default (None), so nothing changes unless you ask for it.

Payloads are never logged

SocketFlow logs metadata about messages, never their content. Sending a 2 MB message produces {"message": "message received", "path": "echo"} — the data is not written anywhere.

The thing to watch is your own code. log.info(f"got {payload}") will happily log a 2 MB payload. Log identifiers and sizes, not bodies.

Metrics

Counters, gauges, and timing histograms. The server records them automatically.

print(server.metrics.counter_value("messages_received_total", path="echo"))
print(server.metrics.gauge_value("connections_active"))
print(server.metrics_snapshot())     # nested dict, JSON-friendly
print(server.metrics_text())        # Prometheus format

Metric	Type	Meaning
connections_accepted_total	counter	Clients that connected
connections_rejected_total	counter	Clients turned away
connections_active	gauge	Connected right now
messages_received_total	counter	By path
messages_sent_total	counter	To clients
bytes_sent_total	counter	Bytes written
errors_total	counter	By context (server.handle_data, client.receive, …)
backpressure_total	counter	Rejections from full queues

Feeding Prometheus

metrics_text() is already Prometheus format, so expose it over HTTP:

# In test_server.py
class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = server.metrics_text().encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/plain; version=0.0.4")
        self.end_headers()
        self.wfile.write(body)

Then scrape http://your-host:9090/metrics:

socketflow_connections_accepted_total 1
socketflow_connections_active 1
socketflow_messages_received_total{path="echo"} 4

The catch

Labels create one time series per value. If you label with something unique — a data_id, a full URL with an ID, a raw socket address — your metrics will grow endlessly and can slow the server down. Label with small, bounded values like path or reason.

The library labels by path and reason only, so this is safe unless you add your own.

Mutual TLS (Client Certificates)

Like a door that checks both badges. The server proves who it is, and the client proves who it is. After the handshake, the server knows the client's name.

server = TcpServer(
    tls_enabled=True,
    tls_certfile="server.crt",
    tls_keyfile="server.key",
    tls_client_ca="clients.crt",           # CA that signs client certs
    tls_require_client_cert=True,          # no cert, no entry
)

client = TcpClient(
    tls_enabled=True,
    tls_ca_certs="ca.crt",
    tls_server_hostname="server.example.com",
    tls_certfile="me.crt",                 # my badge
    tls_keyfile="me.key",
)

Handlers see who is talking:

@server.path("whoami")
def whoami(message):
    print(message.client_identity)   # "laptop-07"
    server.send_client(message.client_addr, message.client_identity, message.data_id)

@server.event(EventType.Server.CLIENT_CONNECT)
def on_connect(data):
    print(f"{data.client_identity} joined")   # "laptop-07"

Or outside a handler:

server.get_client_identity(client_addr)   # "laptop-07" or None

Required vs optional

Setting	No client cert	Bad client cert
tls_require_client_cert=True	Rejected 🔒	Rejected 🔒
tls_client_ca only (optional)	Allowed, client_identity is None	Rejected 🔒
No tls_client_ca	Normal one-way TLS	Normal one-way TLS

The catch

Certificates expire. If one does, the client is locked out until you issue a new one. Plan renewal before the expiry date, and keep the CA private key safe — anyone holding it can mint a certificate your server will trust.

Protocol Version Negotiation

Like a phone that only works on certain networks. Both sides say which versions they speak, then agree on one.

It is off by default, so existing code is unaffected. Turn it on by passing protocol_version or a version range:

server = TcpServer(protocol_version=1)          # speaks exactly version 1
client = TcpClient(protocol_version=1)          # speaks exactly version 1
client.connect()
print(client.negotiated_protocol_version)        # 1

Ranges let old and new builds talk to each other:

server = TcpServer(min_protocol_version=1, max_protocol_version=3)
client = TcpClient(min_protocol_version=1, max_protocol_version=5)

client.connect()
print(client.negotiated_protocol_version)        # 3, the highest both support

If there is no overlap, the connection is refused with a clear reason:

client = TcpClient(min_protocol_version=7, max_protocol_version=9)

try:
    client.connect()
except ProtocolVersionError as error:
    print(error)
    # No common protocol version: client supports 7-9, server supports 1-1

The check runs during the handshake, before authentication, so a mismatched peer never reaches your credentials.

	Old	New
Mismatched versions	Weird errors, or silent breakage	Clear ProtocolVersionError
Upgrading server	May break old clients	Old clients keep working
Knowing what runs	Guesswork	negotiated_protocol_version

Notes:

• Both sides must opt in. If only one side negotiates, no version is recorded and the handshake proceeds as before.
• The server stores the agreed version per connection; the client exposes it as negotiated_protocol_version.

Graceful Shutdown

Draining lets in-flight work finish before connections are closed, so replies that are already being produced are not lost.

# Stop accepting new clients, wait for handlers and queued replies, then close.
finished = server.drain(timeout=10.0)
print("drained cleanly:", finished)

While draining:

• New connections are refused immediately.
• Existing clients stay connected.
• Running and queued handlers are allowed to finish.
• Replies those handlers queued are flushed to the socket.

drain() returns True if everything finished before the timeout, False if the timeout expired first. It does not close connections by itself.

# Combined stop, with draining
server.stop(drain=True, drain_timeout=10.0)

# shutdown() drains by default
server.shutdown()                       # drains, then stops
server.shutdown(drain=False)            # immediate, no drain

To observe a drain starting:

@server.event(EventType.Server.DRAINING)
def on_draining(data):
    print(f"draining: {data.connected_clients} clients still connected")

server.draining is True while a drain is in progress.

Notes:

• Always pass a drain_timeout. A handler that blocks forever will make draining wait until the timeout, then return False.
• Draining waits for handler tasks, not for clients to disconnect. Long-lived clients that send nothing will still hold connections open until stop() closes them.

Connection Protection

Use TLS together with token or username/password authentication:

server = TcpServer(
    host="0.0.0.0",
    port=8080,
    tls_enabled=True,
    tls_certfile="server.crt",
    tls_keyfile="server.key",
    auth_token="replace-with-a-secret",
)

client = TcpClient(
    host="server.example.com",
    port=8080,
    tls_enabled=True,
    tls_ca_certs="ca.crt",
    tls_server_hostname="server.example.com",
    auth_token="replace-with-a-secret",
)

The client must trust the server certificate and use the matching server name. Authentication happens before normal application messages are accepted. The client also waits for the server's final handshake_ok confirmation. Username/password authentication is also available through auth_username and auth_password.

Configuration

Server Options

Parameter	Type	Default	Description
host	str	"127.0.0.1"	Server host address
port	int	8080	Server port
compression_type	str	"zlib"	Codec: zlib, lzma, bz2, gzip (built in), or zstd, brotli (optional)
compression_level	int	6	Compression level (1-9)
compress	bool	True	Enable compression
keepalive_interval	float	30.0	Keepalive interval in seconds
keepalive_max_missed	int	3	Max missed keepalives before disconnect
recv_buffer_size	int	65536	Receive buffer size
send_buffer_size	int	65536	Send buffer size
max_frame_size	int	8388608	Maximum inbound/outbound frame size
max_outbound_queue_bytes	int	16777216	Per-client queued outbound bytes
max_pending_writes	int	1000	Maximum queued outbound frames per client
max_dispatch_workers	int	32	Maximum application handler workers
max_pending_tasks	int	1000	Maximum queued/running handler tasks; also the cap on messages waiting in one block=True queue
dispatch_queue_timeout	float	1.0	Seconds to wait when dispatch capacity is exhausted
max_connections	int	1000	Maximum concurrently accepted clients
allow_pickle	bool	False	Explicitly allow legacy pickle payloads from trusted peers
tls_enabled	bool	False	Encrypt the connection with TLS
tls_certfile	str	None	Server TLS certificate file
tls_keyfile	str	None	Server TLS private key file
tls_client_ca	str	None	CA used to verify client certificates (mutual TLS)
tls_require_client_cert	bool	False	Reject clients that do not present a certificate
tls_handshake_timeout	float	10.0	TLS handshake timeout
auth_enabled	bool	False	Require the authentication handshake
auth_token	str	None	Shared authentication token
auth_username	str	None	Username for authentication
auth_password	str	None	Password for authentication
auth_timeout	float	30.0	Authentication timeout
require_handshake	bool	True	Require the security handshake before messages
handshake_timeout	float	None	Full TLS/server-ready/auth handshake timeout; defaults to the connection timeout
allow_legacy_clients	bool	False	Accept pre-handshake (0.1.x) clients; accepts pickled payloads inbound only
max_memory_bytes	int	268435456	Shared memory budget for connection buffers and queued writes
max_decompressed_size	int	16777216	Maximum size after decompression
use_event_loop	bool	True	Use the shared selector loop for server connections

Client Options

Parameter	Type	Default	Description
host	str	"127.0.0.1"	Server host address
port	int	8080	Server port
compression_type	str	"zlib"	Codec: zlib, lzma, bz2, gzip (built in), or zstd, brotli (optional)
compression_level	int	6	Compression level (1-9)
compress	bool	True	Enable compression
keepalive_interval	float	30.0	Keepalive interval in seconds
keepalive_max_missed	int	3	Max missed keepalives before disconnect
connection_timeout	float	10.0	Connection timeout in seconds
recv_buffer_size	int	65536	Receive buffer size
send_buffer_size	int	65536	Send buffer size
max_frame_size	int	8388608	Maximum inbound/outbound frame size
max_outbound_queue_bytes	int	16777216	Maximum queued outbound bytes
max_pending_writes	int	1000	Maximum queued outbound frames
max_dispatch_workers	int	32	Maximum application handler workers
max_pending_tasks	int	1000	Maximum queued/running handler tasks; also the cap on messages waiting in one block=True queue
dispatch_queue_timeout	float	1.0	Seconds to wait when dispatch capacity is exhausted
allow_pickle	bool	False	Explicitly allow legacy pickle payloads from trusted peers
tls_enabled	bool	False	Encrypt the connection with TLS
tls_ca_certs	str	None	Trusted server CA certificate file
tls_server_hostname	str	None	Name used to verify the server certificate
tls_certfile	str	None	Client certificate presented for mutual TLS
tls_keyfile	str	None	Private key for the client certificate
auth_enabled	bool	False	Send the authentication handshake
auth_token	str	None	Shared authentication token
auth_username	str	None	Username for authentication
auth_password	str	None	Password for authentication
auth_timeout	float	30.0	Authentication timeout
require_handshake	bool	True	Require the security handshake before messages
handshake_timeout	float	None	Full TLS/server-ready/auth handshake timeout; defaults to the authentication timeout
allow_legacy_server	bool	False	Allow a pre-handshake (0.1.x) server; peer is auto-detected
legacy_probe_timeout	float	1.0	Grace period for detecting a 0.1.x server when allow_legacy_server is set
max_memory_bytes	int	67108864	Shared memory budget for this client's buffers and queued writes
max_decompressed_size	int	16777216	Maximum size after decompression

Transport Reliability

• Inbound frames are length-prefixed and validated against max_frame_size.
• Each connection has one serialized outbound writer, preventing frame interleaving.
• Outbound queues are bounded by both bytes and frame count; overload raises Backpressure.
• Application handlers use a bounded worker pool instead of one thread per message.
• Pending responses are isolated per client connection on the server.
• max_memory_bytes limits buffered receive data and queued outgoing data across all connections.
• max_decompressed_size rejects compressed messages that expand beyond the allowed size.
• Server connections use one shared selectors event loop by default; application handlers still use the bounded worker pool.
• TCP_NODELAY is enabled for low-latency request/response traffic.
• Use shutdown() when permanently closing a client or server; use stop()/disconnect() when a restartable lifecycle is needed.
• Compressed application payloads use safe JSON/bytes serialization by default; allow_pickle=True is only for explicitly trusted legacy peers.
• allow_legacy_clients / allow_legacy_server widen what is decoded, not what is encoded. A connection that completes the version handshake always receives the safe JSON format; only a connection positively identified as pre-handshake 0.1.x is sent pickle.
• Server side: a client is identified as legacy when it sends application data before the handshake.
• Client side: a server is identified as legacy when no __server_ready__ arrives within legacy_probe_timeout (default 1.0 s), so a modern server on the other end still receives the safe JSON format.

API Reference

TcpServer

Methods

• start() - Start the server
• stop(drain=False, drain_timeout=10.0) - Stop the server and disconnect all clients
• drain(timeout=10.0, reason="shutdown") - Finish in-flight work, then report success
• shutdown(drain=True, drain_timeout=10.0) - Permanently stop the server and dispatcher
• draining - True while a drain is in progress
• wait() - Block until server stops
• start_and_wait() - Start server and block
• send_client(client_addr, data, data_id=None, path=None, wait_response=False, wait_response_timeout=30.0, status_code=200) - Send data to specific client
• send_client_async(client_addr, data, data_id=None, path=None, timeout=30.0, status_code=200) - Send without blocking, returns a request handle
• disconnect_client(client_addr) - Disconnect a specific client
• get_connected_clients() - Get number of connected clients
• get_client_identity(client_addr) - Certificate common name for a client, or None
• metrics - The MetricsRegistry recording this server's metrics
• metrics_snapshot() - All metrics as a nested dict
• metrics_text() - All metrics in Prometheus text format
• is_connected(client_addr) - Check if client is connected
• event(event_type) - Decorator to register event handler
• path(path, middleware=None, block=False) - Decorator to register path handler; block=True serialises messages for this path per client
• register_blueprint(blueprint) - Register a blueprint

TcpClient

Methods

• connect() - Connect to server
• disconnect() - Disconnect from server
• shutdown() - Permanently disconnect and stop the dispatcher
• send(data, data_id=None, path=None, wait_response=False, wait_response_timeout=30.0, status_code=200) - Send data to server
• send_async(data, data_id=None, path=None, timeout=30.0, status_code=200) - Send without blocking, returns a request handle
• metrics - The MetricsRegistry recording this client's metrics
• wait() - Block until client disconnects
• connect_and_wait() - Connect and block
• is_connected() - Check if connected
• event(event_type) - Decorator to register event handler
• path(path, middleware=None, block=False) - Decorator to register path handler; block=True serialises messages for this path per peer
• register_blueprint(blueprint) - Register a blueprint

Events

Server Events

• EventType.Server.START - Server started
• EventType.Server.DRAINING - Server started draining in-flight work
• EventType.Server.STOP - Server stopped
• EventType.Server.CLIENT_CONNECT - Client connected
• EventType.Server.CLIENT_DISCONNECT - Client disconnected
• EventType.Server.MESSAGE - Message received from client

Client Events

• EventType.Client.CONNECT - Connected to server
• EventType.Client.DISCONNECT - Disconnected from server
• EventType.Client.MESSAGE - Message received from server

Global Events

• EventType.Global.ERROR - Error occurred, including failed path handlers and unknown paths. data.error is the exception, data.context names the path.

Path-Based Routing

Send messages to specific handlers using paths:

# Server
@server.path("/user/login")
def handle_login(data):
    # Handle login
    pass

@server.path("/user/register")
def handle_register(data):
    # Handle registration
    pass

# Client
client.send(data, path="/user/login")

Requesting a path with no registered handler returns 404 with a Not Found payload instead of waiting for the timeout. A handler that raises returns 500 with the reason. See Status Codes and Error Handling.

Serialising a path with block=True

block=True means handle one message at a time for this path — a queue ticket for a single client. It is a rule about messages, not about the network.

Despite the name, it does not:

• block the network or hold the socket still,
• make the client wait for a reply (that is wait_response=True on the sender),
• move the handler onto the main thread.

Either way the handler runs on a background worker drawn from the thread pool. What block=True adds is a queue for that path+client pair, so only one message for that client is inside the handler at a time. The queue is released even if the handler raises.

# Default (block=False): two messages from one client may run at the same time,
# so a read-modify-write sequence can interleave with itself.
@server.path("/counter/increment")
def increment(data):
    ...

# block=True: the same client is never inside this handler twice at once.
@server.path("/counter/reset", block=True)
def reset(data):
    ...

The queue is keyed on the path and the client together, which is what keeps it from becoming a global bottleneck:

Situation	Runs in parallel?
Same path, same client	No — strictly one at a time, in arrival order
Same path, different clients	Yes — each client gets its own queue
Different paths, same client	Yes — separate queues
Different paths, different clients	Yes — separate queues

All paths matching one parameterised pattern share a single queue per client, so /item/1 and /item/2 from the same client are serialised against each other, not just against themselves.

A waiting message does not occupy a worker thread. Each blocking path+client pair gets a queue, and a single worker drains it one message at a time, so a large backlog costs queue space instead of pool capacity. That means one slow client cannot exhaust the pool and stall unrelated paths or unrelated clients. Messages are handled in arrival order. A per-key queue holds at most max_pending_tasks messages; beyond that, sending raises DispatcherError rather than growing without bound.

Blueprints

Organize your code with blueprints:

from socketflow import Blueprint

user_bp = Blueprint("user")

@user_bp.path("/login")
def login(data):
    pass

@user_bp.path("/register")
def register(data):
    pass

# Register blueprint
server.register_blueprint(user_bp)

Keepalive

SocketFlow implements bidirectional keepalive at two levels:

1. Application-Level Keepalive - Custom ping/pong messages
2. TCP-Level Keepalive - OS-managed keepalive probes

Both client and server independently monitor connection health based on their own configurations.

Compression

Support for multiple compression algorithms:

• zlib - Fast compression, good balance
• lzma - High compression ratio, slower
• bz2 - Good compression, moderate speed

Error Handling

SocketFlow provides custom exception types:

• NotConnected - Connection not established
• ConnectionTimeout - Connection attempt timed out
• KeepaliveTimeout - Keepalive timeout
• CompressionError - Compression/decompression error
• InvalidData - Invalid message format
• NoResponse - No response received within timeout
• MessageHandlerError - Message handling error
• Backpressure - Outbound or dispatch capacity was exhausted
• DispatcherError - Dispatcher queue or lifecycle failure
• AuthenticationError - Connection credentials were missing or invalid
• TlsError - TLS setup or certificate verification failed
• HandshakeError - The security handshake did not complete

License

MIT License - see LICENSE file for details

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

• GitHub Issues: https://github.com/ayammaximilian/socketflow/issues
• Documentation: https://socketflow.dev/

Requirements

• Python 3.7+
• No external dependencies (uses only standard library)