Metadata-Version: 2.4
Name: socket-netty
Version: 0.4.0
Summary: Asynchronous networking library for Python, inspired by Netty
Author: Button
Keywords: networking,asyncio,netty,tcp,udp,game-server,protocol
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Networking
Classifier: Framework :: AsyncIO
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: protolib>=0.4.3

# socket-netty

Asynchronous networking library for Python, closely inspired by
[Netty](https://netty.io) (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

```bash
pip install socket-netty
```

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

```bash
pip install -e .
```

`socket_netty` declares [`protolib`](https://pypi.org/project/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

```python
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:

```python
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:

```python
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

```python
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

```python
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:

```python
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:

```python
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:

```python
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

```python
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

```python
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`:

```python
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)

```python
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

```python
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)

```python
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:

```python
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:

```python
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):

```python
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)

```python
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.):

```python
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`](https://pypi.org/project/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`:

```python
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
```

```python
# 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

```python
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)

```python
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

```python
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

```bash
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](#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.
