🔌 Socket I/O (xsocket)

---

Duck provides its own socket implementation: duck.utils.xsocket.
This is a custom, non-blocking replacement for Python’s built-in socket, built with async-first principles 🚀.

With xsocket, you get:

• ✅ Non-blocking calls in async mode.

• ⚡ Seamless async-ready APIs (async_... methods).

• 🔐 SSL/TLS support via ssl_xsocket.

• 🔄 Compatibility with sync and async environments.

---

📦 Getting Started

To use xsocket for Socket I/O, import from duck.utils.xsocket.io.

---

👨‍💻 Synchronous Example

import socket
import threading

from duck.utils.xsocket import create_xsocket, xsocket, ssl_xsocket
from duck.utils.xsocket.io import SocketIO

def handle_conn(sock: xsocket, addr: tuple[str, int]):
    # Do sock.do_handshake() if you are using ssl_xsocket
    print("New connection", addr)
    
    request = SocketIO.receive_full_request(sock, timeout=1)
    
    # Parse request data here
    
    SocketIO.send(b"200 OK\r\n\r\n")  # You can set timeout with the timeout argument
    
    # Close connection
    SocketIO.close(sock)


def serve_forever():
    server_sock = create_xsocket(family=socket.AF_INET, type=socket.SOCK_STREAM)
    server_sock.bind(("0.0.0.0", 5000))
    server_sock.listen(5)
   
    while True:
        sock, addr = server_sock.accept()
        sock = xsocket(sock)  # Or use ssl_xsocket for HTTPS
        threading.Thread(target=handle_conn, args=[sock, addr]).start()


if __name__ == '__main__':
    serve_forever()

---

⚡ Asynchronous Example

import socket
import asyncio

from duck.utils.xsocket import create_xsocket, xsocket, ssl_xsocket
from duck.utils.xsocket.io import SocketIO
from duck.utils.asyncio import create_task  # Better than default asyncio.create_task


async def handle_conn(sock: xsocket, addr: tuple[str, int]):
    # Do await sock.async_do_handshake() if you are using ssl_xsocket
    print("New connection", addr)
    
    request = await SocketIO.async_receive_full_request(sock, timeout=1)
    
    # Parse & process request here
    
    await SocketIO.async_send(b"200 OK\r\n\r\n")  # Supports timeout argument
    
    # Close connection
    SocketIO.close(sock)


async def serve_forever():
    server_sock = create_xsocket(family=socket.AF_INET, type=socket.SOCK_STREAM)
    server_sock.bind(("0.0.0.0", 5000))
    server_sock.listen(5)
   
    while True:
        sock, addr = await server_sock.async_accept()
        sock = xsocket(sock)  # Or use ssl_xsocket for HTTPS
        create_task(handle_conn(sock, addr))


if __name__ == '__main__':
    asyncio.run(serve_forever())

---

📝 Notes

• 🔒 Use ssl_xsocket for secure HTTPS/TLS connections.

• ⏱ Most I/O functions accept a timeout argument.

• 🔄 Works seamlessly with Duck’s async runtime (duck.utils.asyncio.create_task).

• ⚡ Perfect for custom servers, proxies, or protocol implementations.

---

🚀 Performance Tips

• ♻ Reuse sockets where possible to avoid repeated creation overhead.

• 📏 Tune timeout values depending on expected network latency.

• 🧩 Handle partial reads/writes — especially for large frames or slow clients.

• 🛡 SSL handshakes can be expensive; reuse ssl_xsocket connections if possible.

• 🔗 Consider thread pools or async tasks to prevent blocking other connections.

---

✨ With xsocket, you’re not tied to blocking sockets anymore.
Build fast, scalable, async-native servers with just a few lines of code 🔌⚡

---

👉 Next: Learn how this integrates with higher-level protocols like ⚡ ASGI or 🌐 WebSockets.