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
ServerBootstrap	Yes
Channel	Yes	TCP
ChannelPipeline	Yes
ChannelHandler	Yes	inbound + outbound
ChannelHandlerContext	Yes
EventLoop	Yes
EventLoopGroup	Yes
ChannelFuture	Yes	+ ChannelPromise
ByteBuf	Yes
Allocator	Yes	Unpooled + Pooled
TCP	Yes
UDP	Yes	DatagramChannel
Socket options	Yes	ChannelOption
Backpressure	Yes	WriteBufferWaterMark
Concurrency	Yes	ChannelExecutor
TLS/SSL	Yes	SslContextBuilder
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
  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.

Socket options

from socket_netty import ServerBootstrap, ChannelOption

server = (
    ServerBootstrap()
    .child_handler(init_channel)
    .option(ChannelOption.SO_REUSEADDR, True)
    .option(ChannelOption.SO_BACKLOG, 128)
    .option(ChannelOption.TCP_NODELAY, True)  # low latency, typical for games
)
await server.bind("0.0.0.0", 9000)

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)

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())

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
    ...

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.

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).
• PooledByteBufAllocator recycles buffers by size "bucket" (powers of 2), useful in high-traffic game servers to reduce GC pressure.
• 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.