socket-netty

Asynchronous networking library for Python, closely inspired by Netty (Java). Built on top of asyncio, with no external dependencies.

Coverage relative to Netty

Netty	Python (socket-netty)	Notes
Bootstrap	Yes	+ child_option for per-connection socket options
ServerBootstrap	Yes
Channel	Yes	TCP, write/queue_write/flush
ChannelPipeline	Yes	add_first/last/before/after, replace, remove_first/last, first/last
ChannelHandler	Yes	inbound + outbound, channel_writability_changed, user_event_triggered
ChannelHandlerContext	Yes	is_removed()
EventLoop	Yes
EventLoopGroup	Yes
ChannelFuture	Yes	+ ChannelPromise
ByteBuf	Yes	get_*/set_*, little-endian, slice/copy/discard_read_bytes
Allocator	Yes	Unpooled + Pooled
TCP	Yes
UDP	Yes	DatagramChannel
Socket options	Yes	ChannelOption, option/child_option
Backpressure	Yes	WriteBufferWaterMark, channel_writability_changed
Concurrency	Yes	ChannelExecutor
TLS/SSL	Yes	SslContextBuilder
HTTP/1.1	Yes	HttpServerCodec/HttpClientCodec, chunked encoding, keep-alive, HttpObjectAggregator
HTTPS	Yes	Same HTTP pipeline + .ssl(...), TLS handled transparently by the transport
WebSocket	Yes	RFC 6455 server-side upgrade, all frame types, WSS (WebSocket over TLS)
Encoders / Decoders	Yes	length-based framing
Idle handlers	Yes	IdleStateHandler
Timeouts	Yes	Read/WriteTimeoutHandler
Exception handling	Yes	exception_caught via the pipeline, io.netty.*-style exception classes
Declarative packet codec	Yes	ProtolibCodec, via the protolib dependency

Installation

pip install socket-netty

Or, for local development from a clone of this repo:

pip install -e .

socket_netty declares protolib as an install dependency, so either install method pulls it in automatically — you don't need a separate install step to use ProtolibCodec below.

Structure

socket_netty/
  exceptions.py -> Netty-style exception hierarchy (DecoderException,
                    EncoderException, CorruptedFrameException, etc.)
  buffer/       -> ByteBuf, ByteBufAllocator (Unpooled/Pooled)
  handler/      -> ChannelHandler, codecs, protolib bridge, SSL,
                    timeouts/idle
  handler/http/ -> HTTP/1.1 (HttpServerCodec, HttpObjectAggregator) and
                    WebSocket (RFC 6455 handshake + frame codec)
  channel/      -> Channel, DatagramChannel, ChannelPipeline,
                    EventLoop/Group, ChannelFuture, ChannelOption,
                    flow control (backpressure + concurrency)
  bootstrap/    -> ServerBootstrap (server), Bootstrap (client)

Example: TCP echo server

import asyncio
from socket_netty import ServerBootstrap, ChannelInboundHandler

class EchoHandler(ChannelInboundHandler):
    async def channel_active(self, ctx):
        print("Client connected:", ctx.channel.remote_address())

    async def channel_read(self, ctx, msg):
        await ctx.write(msg)  # echo

    async def exception_caught(self, ctx, exc):
        print("Error:", exc)
        await ctx.close()

async def main():
    def init_channel(channel):
        channel.pipeline.add_last("echo", EchoHandler())

    server = await ServerBootstrap().child_handler(init_channel).bind("0.0.0.0", 9000)
    await server.serve_forever()

asyncio.run(main())

Exceptions

socket_netty raises the same exception hierarchy Netty does, under io.netty.*. Every exception's str() is prefixed with its fully-qualified Java-style path, so logs and exception_caught output look exactly like a real Netty stack:

from socket_netty import DecoderException

async def exception_caught(self, ctx, exc):
    print(exc)
    # io.netty.handler.codec.DecoderException: Failed to decode packet

Available exceptions (all subclass NettyException):

Class	Netty path
ChannelException	io.netty.channel.ChannelException
DuplicateHandlerNameError	io.netty.channel.ChannelPipelineException
CodecException	io.netty.handler.codec.CodecException
DecoderException	io.netty.handler.codec.DecoderException
EncoderException	io.netty.handler.codec.EncoderException
CorruptedFrameException	io.netty.handler.codec.CorruptedFrameException
TooLongFrameException	io.netty.handler.codec.TooLongFrameException
ReadTimeoutError	io.netty.handler.timeout.ReadTimeoutException
WriteTimeoutError	io.netty.handler.timeout.WriteTimeoutException
IndexOutOfBoundsError	io.netty.buffer.IndexOutOfBoundsException

LengthFieldBasedFrameDecoder and ByteToMessageCodec automatically wrap any unexpected error raised inside decode() into a DecoderException, and LengthFieldPrepender does the same for EncoderException — matching Netty's own behavior of never letting a raw internal error escape a codec unwrapped.

Pipeline management

Beyond add_first/add_last/remove/get, the pipeline supports inserting or swapping handlers at any position:

pipeline.add_before("game", "auth", AuthHandler())   # right before "game"
pipeline.add_after("auth", "logging", LoggingHandler())  # right after "auth"

pipeline.first()   # handler closest to the head, or None
pipeline.last()    # handler closest to the tail, or None
pipeline.names()   # names in actual pipeline order

pipeline.remove_first()
pipeline.remove_last()

# Common pattern: swap a login handler for the real game handler once
# a player authenticates.
pipeline.replace("auth", "game", GameHandler())

A handler can check ctx.is_removed() to see whether it's still part of the pipeline before touching ctx again from an async task started earlier (e.g. one kicked off from channel_read).

Socket options

from socket_netty import ServerBootstrap, ChannelOption

server = (
    ServerBootstrap()
    .child_handler(init_channel)
    .option(ChannelOption.SO_REUSEADDR, True)   # applies to the listening socket
    .option(ChannelOption.SO_BACKLOG, 128)      # applies to the listening socket
    .child_option(ChannelOption.TCP_NODELAY, True)  # applies to each ACCEPTED connection
)
await server.bind("0.0.0.0", 9000)

option() configures the listening socket itself. child_option() configures each accepted connection instead — this is what you want for per-connection tuning like TCP_NODELAY (disabling Nagle's algorithm for lower latency), since the listening socket never carries application traffic.

TLS/SSL

from socket_netty import ServerBootstrap, Bootstrap, SslContextBuilder

# Server
server_ctx = SslContextBuilder.for_server("cert.pem", "key.pem").build()
server = ServerBootstrap().child_handler(init_channel).ssl(server_ctx)
await server.bind("0.0.0.0", 9443)

# Client
client_ctx = SslContextBuilder.for_client().trust_manager("cert.pem").build()
channel = await Bootstrap().handler(init_client).ssl(client_ctx).connect("myserver.com", 9443)

HTTP / HTTPS

HttpServerCodec decodes incoming requests and encodes outgoing responses through a single pipeline handler. Pair it with HttpObjectAggregator if you don't need streaming and just want a complete FullHttpRequest per request:

from socket_netty import (
    ServerBootstrap, ChannelInboundHandler,
    HttpServerCodec, HttpObjectAggregator,
    FullHttpRequest, FullHttpResponse, HttpResponseStatus, HttpVersion, HttpUtil,
    LastHttpContent,
)

class HelloHandler(ChannelInboundHandler):
    async def channel_read(self, ctx, msg):
        if isinstance(msg, FullHttpRequest):
            body = f"Hello, {msg.uri}".encode()
            resp = FullHttpResponse(HttpResponseStatus.OK, body, HttpVersion.HTTP_1_1)
            resp.headers.set("Content-Type", "text/plain")
            HttpUtil.set_content_length(resp, len(body))
            HttpUtil.set_keep_alive(resp, HttpUtil.is_keep_alive(msg))
            await ctx.write_no_flush(resp)
            await ctx.write(LastHttpContent(b""))

def init_channel(channel):
    channel.pipeline.add_last("http", HttpServerCodec())
    channel.pipeline.add_last("aggregator", HttpObjectAggregator(1024 * 1024))  # max body size
    channel.pipeline.add_last("hello", HelloHandler())

server = await ServerBootstrap().child_handler(init_channel).bind("0.0.0.0", 8080)

HTTPS is the exact same pipeline with .ssl(...) added — TLS is handled transparently by the underlying transport, the HTTP handlers above don't change at all:

server_ctx = SslContextBuilder.for_server("cert.pem", "key.pem").build()
server = ServerBootstrap().child_handler(init_channel).ssl(server_ctx)
await server.bind("0.0.0.0", 8443)

Chunked request/response bodies are decoded/encoded transparently - HttpObjectAggregator reassembles a chunked body into a normal FullHttpRequest/FullHttpResponse (with Content-Length set and Transfer-Encoding removed) before your handler ever sees it. If you want to stream a large response instead of buffering it all, skip the aggregator and write HttpResponse + one or more HttpContent + LastHttpContent yourself.

Keep-alive follows the real HTTP/1.0 vs 1.1 rules via HttpUtil: HTTP/1.1 connections stay open by default (Connection: close closes them), HTTP/1.0 connections close by default (Connection: keep-alive keeps them open).

Not implemented (documented gap): the Expect: 100-continue request/response flow.

WebSocket

WebSocketServerProtocolHandler sits after the HTTP codec/aggregator and handles the RFC 6455 upgrade handshake automatically - once a client connects, it swaps the pipeline from HTTP handling to WebSocket frames for that connection:

from socket_netty import (
    ServerBootstrap, ChannelInboundHandler,
    HttpServerCodec, HttpObjectAggregator,
    TextWebSocketFrame, CloseWebSocketFrame,
    WebSocketServerProtocolHandler,
)

class EchoHandler(ChannelInboundHandler):
    async def channel_read(self, ctx, msg):
        if isinstance(msg, TextWebSocketFrame):
            await ctx.write(TextWebSocketFrame(text=f"echo: {msg.text}"))
        elif isinstance(msg, CloseWebSocketFrame):
            await ctx.write(CloseWebSocketFrame(1000, "bye"))
            await ctx.close()

def init_channel(channel):
    channel.pipeline.add_last("http", HttpServerCodec())
    channel.pipeline.add_last("aggregator", HttpObjectAggregator(1024 * 1024))
    channel.pipeline.add_last(
        "upgrade",
        WebSocketServerProtocolHandler(path="/ws", http_handler_names=["http", "aggregator"]),
    )
    channel.pipeline.add_last("echo", EchoHandler())

server = await ServerBootstrap().child_handler(init_channel).bind("0.0.0.0", 8080)

path restricts the upgrade to that specific request path, so the same server can serve normal HTTP on other paths. http_handler_names tells the handshake handler which HTTP-side handlers to remove from the pipeline once the connection becomes a WebSocket (there's no reliable way to infer this automatically, since you choose those names yourself). WSS (WebSocket over TLS) is, again, the same pipeline with .ssl(...) on the bootstrap — nothing else changes.

Frame types: TextWebSocketFrame, BinaryWebSocketFrame, ContinuationWebSocketFrame, PingWebSocketFrame, PongWebSocketFrame, CloseWebSocketFrame. Not implemented (documented gap): WebSocket extensions (permessage-deflate compression) and the client-side handshaker (WebSocketClientHandshaker)

• the server side above covers the common "game server exposes a WebSocket API/panel" case.

UDP

from socket_netty import DatagramBootstrap, ChannelInboundHandler

class UdpHandler(ChannelInboundHandler):
    async def channel_read(self, ctx, msg):
        data, addr = msg  # UDP delivers (bytes, (host, port))
        await ctx.write((b"pong", addr))

channel = await DatagramBootstrap().handler(
    lambda ch: ch.pipeline.add_last("udp", UdpHandler())
).bind("0.0.0.0", 9001)

Idle / Timeouts

from socket_netty import IdleStateHandler, ReadTimeoutHandler

async def on_idle(ctx, event):
    print("Idle channel:", event.state)
    await ctx.write(b"PING")  # e.g. a game protocol heartbeat

def init_channel(channel):
    idle = IdleStateHandler(reader_idle_seconds=30)
    idle.on_idle = on_idle
    channel.pipeline.add_last("idle", idle)
    channel.pipeline.add_last("read_timeout", ReadTimeoutHandler(60))  # closes if no data in 60s
    channel.pipeline.add_last("my_handler", MyHandler())

Custom (non-I/O) events

ChannelInboundHandler.user_event_triggered(ctx, event) lets an application signal something through the pipeline that isn't a network read - e.g. "player finished login" triggered from application code rather than channel_read:

class GameHandler(ChannelInboundHandler):
    async def user_event_triggered(self, ctx, event):
        if event == "login_complete":
            print("Player is ready:", ctx.channel.remote_address())

# From anywhere with access to the channel/ctx:
await ctx.fire_user_event_triggered("login_complete")

EventLoopGroup (real multi-threaded concurrency)

from socket_netty import EventLoopGroup

worker_group = EventLoopGroup(num_threads=4)
worker_group.start()

loop = worker_group.next_loop()  # round-robin
future = loop.submit(lambda: my_heavy_coroutine())
result = future.result(timeout=5)  # blocking, from any thread

ChannelFuture / ChannelPromise

from socket_netty import ChannelFuture

future = channel.new_future()
future.add_listener(lambda f: print("Finished:", f.is_success()))

# Still directly awaitable, Python-style:
result = await future

Backpressure (WriteBufferWaterMark)

from socket_netty import ServerBootstrap, WriteBufferWaterMark

server = (
    ServerBootstrap()
    .child_handler(init_channel)
    .water_mark(WriteBufferWaterMark(low=32*1024, high=64*1024))
)

# In the handler:
if not ctx.channel.is_writable():
    # pause sending more data until it becomes writable again
    ...

ChannelInboundHandler.channel_writability_changed(ctx) is fired automatically through the pipeline whenever writability flips, so you don't have to poll is_writable() to react to backpressure:

class MyHandler(ChannelInboundHandler):
    async def channel_writability_changed(self, ctx):
        if not ctx.channel.is_writable():
            print("Backpressure: pausing outgoing updates")
        else:
            print("Writable again: resuming")

Batched writes (write / queue_write / flush)

write() keeps its historical behavior: it queues a message through the outbound handlers and immediately flushes it (equivalent to Netty's writeAndFlush()). For a game server updating many entities per tick, queue_write() + a single flush() sends everything queued as one combined write, paying the pipeline traversal cost once instead of once per message:

for entity in updated_entities:
    await channel.queue_write(encode(entity))  # queued, not sent yet
await channel.flush()                          # sends everything queued, combined

The same pair (write_no_flush()/flush()) is available on ChannelHandlerContext for handlers that want the same control.

Concurrency (ChannelExecutor)

Guarantees a channel's messages are processed one at a time, in order, even across await points (avoids race conditions on shared handler state):

from socket_netty import ChannelExecutor

executor = ChannelExecutor()
executor.start()

async def channel_read(self, ctx, msg):
    await executor.submit(self._process(ctx, msg))

Allocator (buffer pool)

from socket_netty import PooledByteBufAllocator

allocator = PooledByteBufAllocator()
buf = allocator.buffer(256)
buf.write_int(42)
# ... use the buffer ...
allocator.release(buf)  # goes back to the pool, ready to be reused

Protocol framing (length-prefixed packets)

Very common in game protocols (Minecraft, Free Fire, etc.):

from socket_netty import LengthFieldBasedFrameDecoder, LengthFieldPrepender

def init_channel(channel):
    channel.pipeline.add_last("frame_decoder", LengthFieldBasedFrameDecoder(4))
    channel.pipeline.add_last("frame_prepender", LengthFieldPrepender(4))
    channel.pipeline.add_last("my_handler", MyHandler())

Protocol decoding with protolib (declarative packets)

LengthFieldBasedFrameDecoder/LengthFieldPrepender above only solve framing (where one packet ends and the next begins) — you still have to hand-write the code that turns those raw bytes into a meaningful packet. protolib solves that second half: you describe every packet's fields in a .yml/.json file, and ProtolibCodec plugs that description directly into the pipeline. channel_read then hands your handler a ready-made {"name": ..., "params": {...}} dict instead of raw bytes, and write accepts the same shape (or a (name, params) tuple) going out.

ProtolibCodec has two framing modes, controlled by framed:

from socket_netty import ServerBootstrap, ChannelInboundHandler, ProtolibCodec

# framed=True (default): the codec frames the stream itself via
# protolib's own PacketFramer (varint length-prefix, Minecraft-style).
# Don't also add LengthFieldBasedFrameDecoder in this mode.
def init_channel(channel):
    channel.pipeline.add_last(
        "protolib", ProtolibCodec("my_protocol.yml", state="play",
                                   direction_in="toServer", direction_out="toClient"),
    )
    channel.pipeline.add_last("game", GameHandler())

class GameHandler(ChannelInboundHandler):
    async def channel_read(self, ctx, msg):
        print(msg["name"], msg["params"])          # ready-made dict
        await ctx.write(("keep_alive", {"id": 1}))  # (name, params) out

# framed=False: for fixed-size / non-varint protocols (e.g. Minecraft
# Classic/ClassiCube), keep your own LengthFieldBasedFrameDecoder in
# front and let ProtolibCodec just parse/serialize the complete frame
# it's handed.
def init_channel(channel):
    channel.pipeline.add_last("frame_decoder", LengthFieldBasedFrameDecoder(1))
    channel.pipeline.add_last(
        "protolib", ProtolibCodec("classicube_protocol.yml", framed=False),
    )
    channel.pipeline.add_last("game", GameHandler())

protocol accepts a protolib.Protocol instance you already built, or anything Protocol(...) itself accepts (a .yml/.json path, an in-memory string, or a parsed dict) — ProtolibCodec builds the Protocol for you in that case. Any parsing/serialization error protolib raises is wrapped into DecoderException/EncoderException, same as the rest of the codecs in this library.

ByteBuf

from socket_netty import ByteBuf

buf = ByteBuf()
buf.write_varint(1000)
buf.write_string("hello")
buf.write_int(-42)

data = buf.to_bytes()

read_buf = ByteBuf.wrapped(data)
n = read_buf.read_varint()
s = read_buf.read_string()
i = read_buf.read_int()

Supports: byte, unsigned_byte, short, unsigned_short, int, unsigned_int, long, float, double, boolean, varint, varlong (Minecraft/protobuf style), string, and raw bytes — in both big-endian (default) and little-endian (*_le suffix, e.g. write_int_le/read_int_le) for every fixed-size numeric type.

Absolute get/set (don't move reader_index/writer_index)

buf = ByteBuf()
buf.write_int(0)              # placeholder for a length prefix
start = buf.writer_index
buf.write_string("hello world")
buf.set_int(0, buf.writer_index - start)  # patch the length in afterwards

Every fixed-size type has a get_*/set_* pair (get_int/set_int, get_short/set_short, etc., plus _le variants) for reading/writing at an absolute index without touching reader_index/writer_index — the classic "write payload, go back and patch the length prefix" pattern.

Views: slice / duplicate / copy

buf = ByteBuf(b"hello world")
buf.read_bytes(6)          # consume "hello "
view = buf.slice()         # shares memory with buf - "world"
independent = buf.copy()   # independent deep copy

slice()/duplicate()/retained_slice() share backing storage with the original buffer (writes through one are visible through the other) and can't grow past their fixed window. copy() is a fully independent deep copy. Note a CPython-specific limitation: while a slice is alive, the parent buffer can't grow, clear(), or discard_read_bytes() — doing so raises IndexOutOfBoundsError with a clear message; drop every reference to a slice once you're done with it.

Other utilities

• discard_read_bytes() — compacts the buffer, dropping the already-consumed prefix. Useful for long-lived accumulation buffers (frame decoders) so they don't grow indefinitely in memory.
• mark_writer_index() / reset_writer_index() — writer-side counterpart to mark_reader_index()/reset_to_mark().
• bytes_before(value) — position of the first occurrence of a byte ahead of reader_index, without moving it (handy for null-terminated C-style strings).
• ensure_writable(length) — pre-reserves capacity.

Tests

python3 tests/test_bytebuf.py
python3 tests/test_echo.py
python3 tests/test_core_extras.py      # EventLoop, ChannelFuture, Allocator
python3 tests/test_network_extras.py   # UDP, socket options, timeouts, idle
python3 tests/test_tls.py              # end-to-end TLS (generates temporary certs)
python3 tests/test_concurrency.py      # ChannelExecutor

Design notes

• The entire pipeline is 100% async/await.
• EventLoopGroup uses real OS threads (each with its own asyncio loop), unlike the rest of the library which normally runs on a single loop — it's the honest way to replicate Netty's model (thread pool) in Python.
• ByteBuf is a simple implementation backed by bytearray, with no manual refcounting (Python already has GC). slice()/duplicate() share memory via a memoryview instead of copying, but inherit a CPython-specific limitation from it: a bytearray can't be resized while a memoryview into it is alive, so the parent buffer can't grow/clear()/discard_read_bytes() while a slice of it still is.
• PooledByteBufAllocator recycles buffers by size "bucket" (powers of 2), useful in high-traffic game servers to reduce GC pressure.
• Per-connection events (data_received, connection_made, connection_lost, writability changes) are processed one at a time, in the exact order asyncio delivered them, via a small internal queue — this preserves TCP's in-order delivery guarantee at the application level even when a handler's processing of one message takes an await long enough for a later message to otherwise finish first.
• Exceptions mirror Netty's own io.netty.* hierarchy (see Exceptions above), so error messages and logs read the same as a real Netty stack trace.
• Designed for use both in game projects (custom servers, protocol reversing) and as a general-purpose library.