Files and Streams

Handlers run on the polling thread, so a reply or a pull that finishes later runs on a worker: request.detach() hands the request over as a ucall.DetachedRequest, which reads and replies like a Request and alone has the three verbs below.

§ Serving a File

import concurrent.futures

import ucall

workers = concurrent.futures.ThreadPoolExecutor()
server = ucall.Server(ucall.Interface("0.0.0.0:8080", reply_directories=["/srv/videos"]))


def send_video(detached: ucall.DetachedRequest, name: str):
    deferred = detached.reply_parts([ucall.Path(0, name)])
    deferred.wait()  # or: await deferred


@server.get("/videos/{name}")
def video(request: ucall.Request):
    workers.submit(send_video, request.detach(), request.path_parameter("name"))

reply_parts concatenates bytes-like objects, ucall.Path(directory, path, offset=0, length=0) and ucall.File(descriptor, offset=0, length=0) into one body whose Content-Length is measured from the files. A Path resolves beneath the named reply_directories entry, and the event loop opens, reads and closes it unbuffered. A File is the caller’s regular file opened with os.O_DIRECT, or an object whose fileno() answers one; UCall never closes it, and the Deferred keeps it open until the reply is final. deferred.status() reads the outcome without waiting, deferred.wait() blocks without the interpreter lock and raises the failure, such as Status.NOT_FOUND after a 404, and await deferred runs that wait on the running loop’s default executor. Releasing an unfinished Deferred waits for it, which is why only a detached request makes one.

§ Streaming a Reply

def send_rows(detached: ucall.DetachedRequest, rows):
    stream = detached.reply_stream(headers=("content-type: text/csv",))
    for row in rows:
        stream.append([row.encode()])
    stream.end()

append queues more parts and raises ucall.Error with Status.WOULD_BLOCK while the peer lags behind what is queued; end() answers None or a Deferred as reply_parts does, and a stream released without end() ends itself.

§ Streaming an Upload

import hashlib


def store(detached: ucall.DetachedRequest):
    digest = hashlib.sha256()
    for chunk in detached.body_chunks():  # or: async for chunk in detached.body_chunks()
        digest.update(chunk)
    detached.reply(digest.hexdigest().encode())


@server.post("/uploads", body=ucall.BodyDelivery.STREAMED)
def upload(request: ucall.Request):
    workers.submit(store, request.detach())

An endpoint registered with body=ucall.BodyDelivery.STREAMED, on add_endpoint, route and every verb decorator, runs its handler once the headers arrived. body_chunks() yields each chunk as a memoryview valid until the next pull, which releases it, and waits for bytes without the interpreter lock. async for over it returns a chunk already decoded at once and runs the wait for one still arriving on the running loop’s default executor, as await deferred does.