socketify.py

When writing an extension for socketify.py, you can augment the core objects using decorators and property methods:

Adding Methods

Use the @request.method, @response.method, or @ws.method decorators. The first argument of the decorated function must be self to allow access to the object's internal state (e.g., self.get_header(), self.publish(), or self.write_header()).

Adding Properties

Use request.property(name, default_value) to attach data to the request object that can be accessed later in the request lifecycle (e.g., in middlewares or route handlers).

Example usage:

def extension(request, response, ws):
    @request.method
    async def get_user(self):
        return {"name": "Test"}

    request.property("cart", [])

@request.method
async def get_user(self):
    token = self.get_header("token")
    return { "name": "Test" } if token else { "name": "Anonymous" }

request.property("cart", [])

Corking is a performance optimization mechanism that packs multiple send operations into a single syscall or SSL block. This prevents excessive networking and poor performance caused by many small, individual network writes.

Automatic Corking

• Default Behavior: Sockets are corked by default in most simple cases and within all registered business logic callbacks (e.g., when a socket opens or a message is received).
• Limitations:
  • Third-party libraries: Callbacks registered to other libraries (like libhiredis) will not be corked automatically because the library cannot control their execution flow.
  • Async/Await Segments: In coroutines, automatic corking only applies to the first segment of the coroutine (the code before the first await). Once an await is encountered, the execution is handed over to the asyncio event loop, and the subsequent segments are no longer under the library's automatic corking control.

Manual Corking

To ensure efficiency when sending data after asynchronous operations, you must manually wrap your sending logic in a function or lambda using res.cork() or ws.cork_send().

Important: You cannot use async inside a cork block. Instead, perform all your asynchronous work first, then use the cork block to execute the final synchronous sending operations.

async def home(res, req):
    auth = req.get_header("authorization")
    user = await do_auth(auth)
    
    res.cork(lambda res: res.end(f"Hello {user.name}"))

The res.send_chunk method is a high-level abstraction for managing streaming backpressure. It works by:

1. Creating a Future to track the chunk's status.
2. Registering an on_writable callback that calls res.try_end with the remaining slice of the buffer once the socket is ready to write.
3. Registering an on_aborted callback to set the aborted flag and resolve the Future with (False, True) if the client disconnects.
4. Attempting an immediate res.try_end. If it fails (due to backpressure), it stores the current write offset via self.get_write_offset() so the next chunk can be sliced correctly.

def send_chunk(self, buffer, total_size):
        self._chunkFuture = self.loop.create_future()
        self._lastChunkOffset = 0

        def is_aborted(self):
            self.aborted = True
            try:
                if not self._chunkFuture.done():
                    self._chunkFuture.set_result(
                        (False, True)
                    )  # if aborted set to done True and ok False
            except:
                pass

        def on_writeble(self, offset):
            # Here the timeout is off, we can spend as much time before calling try_end we want to
            (ok, done) = self.try_end(
                buffer[offset - self._lastChunkOffset : :],
                total_size
            )
            if ok:
                self._chunkFuture.set_result((ok, done))
            return ok

        self.on_writable(on_writeble)
        self.on_aborted(is_aborted)

        if self.aborted:
            self._chunkFuture.set_result(
                (False, True)
            )  # if aborted set to done True and ok False
            return self._chunkFuture

        (ok, done) = self.try_end(buffer, total_size)
        if ok:
            self._chunkFuture.set_result((ok, done))
            return self._chunkFuture
        # failed to send chunk
        self._lastChunkOffset = self.get_write_offset()

        return self._chunkFuture

To avoid backpressure spikes and memory issues when sending large payloads, do not use res.end(huge_buffer). Instead, use res.send_chunk to stream data in parts.

res.send_chunk(buffer, total_size) returns a Future. It internally manages res.on_writable and res.on_aborted callbacks to handle flow control.

When iterating through chunks, check the returned tuple (ok, done):

• ok: Indicates if the chunk was successfully accepted for sending.
• done: Indicates if the entire response has been completed.

If ok is False or done is True, you should stop reading from your data source (e.g., break the loop) as the connection may have been aborted.

async def home(res, req):
    res.write_header("Content-Type", "audio/mpeg")
   
    filename = "./file_example_MP3_5MG.mp3"
    total = os.stat(filename).st_size
    
    async with aiofiles.open(filename, "rb") as fd:
        while not res.aborted:
            buffer = await fd.read(16384)  #16kb buffer
            (ok, done) = await res.send_chunk(buffer, total)
            if not ok or done: #if cannot send probably aborted
                break