Chapter 108 Streaming Bodies: Upload, Download, and Async Files

Chapter Orientation

HTTP's body is the battlefield of "content" versus "memory": a 2 GB upload, a streaming download, a response generated on demand — body size should not equal memory footprint is the starting point of the whole design. xhttpbody unifies five body sources: the fixed family (Empty/Copy/Borrow/Take/Reference — replayable, known length), the bounded producer stream (HttpBodyStreamCreate — task-thread production, budget-bounded, the foundation of SSE), file bodies (http_body_file — async file IO made into a body), custom sources (producer callbacks), and transformed bodies. On the consuming side xhttpbodyreader is a single-consumer streaming reader, and xhttpbodychunk is an independent data lease (ChunkRelease to return, releasable later than the Reader — the perfect partner of TCP reference sending). Chapter 104's server-side Body callbacks and Chapter 107's SSE producer end get their complete foundation story here.

108.1引入

Three body scenarios with memory-shape demands. Uploading 2 GB: the server must not buffer it all before processing — Chapter 104's streaming Body callbacks already solved the receiving side; but what about the client's sending side — reading 2 GB into memory before the POST? File bodies let it "read as you send". Streaming-generated downloads: response length unknown, content produced on demand (database cursors, live transcoding) — producer-stream bodies let it "generate as you send", capped by budget. Replay needs: Chapter 100's 307/308 redirects require a replayable body — the fixed family is replayable by nature; what about producer streams? The Body that HttpBodyStreamCreate produces is consumed once — scenarios that must resend on redirect use the fixed family or file bodies.

The lease is this chapter's most important new concept: the Chunk that Next yields owns an independent data lease — the data is not copied, but you must ChunkRelease to return it; a Chunk may outlive the Reader (sitting in a reference-send queue while the body object is already destroyed — managed by reference counting). This is the same idea as Chapter 96's WebSocket reference sending (released only when the write completes), landing on the body layer.

108.2概念

Five sources and replayable semantics

Fixed-family details: Copy places the descriptor and the copy in one compact allocation sized to the actual length (no second allocation for the same body); Borrow does not extend the external memory's lifecycle (the original buffer must stay alive during the send); Reference must supply a release procedure (if you only need borrowing, use Borrow). xrtHttpBodyView borrows contiguous bytes of a fixed body — sources that cannot guarantee contiguity return false (file/streaming/transformed may be non-contiguous).

Reader: single consumer and leases

Aliasing discipline (contract-literal): the output Chunk's descriptor/buffer/length slots must not overlay the underlying bytes of the Reader, the Body, or a fixed body — an aliasing error advances the Reader not at all and clears no output length, and a fixed Body stays immutable through replay and concurrent opens. This defense makes "one body object shared by multiple Calls" safe (Chapter 99's frozen-semantics body references).

The bounded producer stream: budgets and finishes

xhttpbodystreamconfig's MaxBytes/MaxChunks is the producer-side budget (reservations + queuing + active leases all covered) — SSE's AGAIN threshold comes exactly from here. Three finishes: destruction of the last stream reference = queued content then a normal EOF; Close idempotently pre-closes the input; Fail discards undelivered content and hands a stable Cause to the consuming side. The Write/WriteRef/WriteTake three write forms correspond to copy/reference/take-over — the same family as Chapter 68's five send gears.

File bodies: async IO as a body

The http_body_file family wraps Chapter 72's completion-port file IO into a body: xrtHttpBodyFileConfigInit + file-body creation — reads take the async file path (driven by the Engine's Workers), no blocking, no whole-file reads; xrtHttpBodyFileFuture/FileRangeFuture provide the "wait for the body to be ready" Future form. Range variants naturally support resumable upload/download. File bodies are replayable (reopen the file, reread) — the safe combination of large files + redirect resends.

The complete shapes of upload and download

• Upload (client): a file body + Chapter 99's builder SetBody — the client memory of a 2 GB upload is roughly one read block; the server receives via Chapter 104's streaming Body callbacks.
• Download (client): Chapter 99's ResponseBodyLimit manages the budget, streaming consumption (Body callbacks or the Reader) — memory decoupled from download size; Chapter 93's decoders chain onto the consumption path.
• Generated responses (server): a producer-stream body + Respond — a database cursor yields step by step; once the budget caps out, AGAIN makes the producer wait for the consuming side. Running all of this over a TLS service (Chapter 86), request-head parsing goes through Chapter 90's parser TLS binding (http1_tls) — no connection-level buffer across records; the streaming discipline runs end to end.

108.3示例

First complete program: producer-stream write-out and read-back

The program below is from examples/http/body_stream — the minimal loop of a bounded producer stream:

#include <stdio.h>
#include <xhttp.h>



/* 读取完整异步正文；AGAIN 只通过公开 Future 契约继续。 */
static bool exampleReadBody(xhttpbody* pBody)
{
	xhttpbodyreader* pReader = xrtHttpBodyOpen(pBody);
	xhttpbodychunk Chunk;
	xhttpbodystatus Status;

	if ( pReader == NULL ) {
		return false;
	}
	for ( ;; ) {
		Status = xrtHttpBodyNext(pReader, 4, &Chunk);
		if ( Status == XHTTP_BODY_DATA ) {
			(void)fwrite(Chunk.Data, 1, Chunk.Size, stdout);
			xrtHttpBodyChunkRelease(&Chunk);
			continue;
		}
		if ( Status == XHTTP_BODY_AGAIN ) {
			xfuture* pReady = xrtHttpBodyReaderWait(pReader);

			if ( (pReady == NULL) ||
				(xrtFutureWait(pReady) != XWAIT_OK) ||
				(xrtFutureState(pReady) != XFUTURE_RESOLVED) ) {
				xrtFutureDestroy(pReady);
				xrtHttpBodyReaderDestroy(pReader);
				return false;
			}
			xrtFutureDestroy(pReady);
			continue;
		}
		break;
	}
	xrtHttpBodyReaderDestroy(pReader);
	return Status == XHTTP_BODY_EOF;
}



/* 创建有界生产流，写入两段数据并由最后一个生产端发布 EOF。 */
int main(void)
{
	xhttpbodystreamconfig Config;
	xhttpbodystream* pStream = NULL;
	xhttpbody* pBody;
	bool bResult;

	xrtHttpBodyStreamConfigInit(&Config);
	Config.MaxBytes = 1024;
	Config.MaxChunks = 16;
	pBody = xrtHttpBodyStreamCreate(&Config, &pStream);
	if ( (pBody == NULL) ||
		(xrtHttpBodyStreamWrite(
			pStream, XRT_BYTES_LITERAL("first chunk\n")
		) != XHTTP_BODY_STREAM_OK) ||
		(xrtHttpBodyStreamWrite(
			pStream, XRT_BYTES_LITERAL("second chunk\n")
		) != XHTTP_BODY_STREAM_OK) ) {
		xrtHttpBodyStreamDestroy(pStream);
		xrtHttpBodyDestroy(pBody);
		return 1;
	}
	xrtHttpBodyStreamDestroy(pStream);
	bResult = exampleReadBody(pBody);
	xrtHttpBodyDestroy(pBody);
	return bResult ? 0 : 1;
}

What just happened. (1) Config.MaxBytes=1024/MaxChunks=16 sets the budget, then HttpBodyStreamCreate yields the object pair in one call: xhttpbody (handed to the request/response) and xhttpbodystream (held by the producer). (2) Two Write segments, then the stream destroyed — last-reference destruction = normal EOF: queued content delivers first, EOF follows. (3) The read-back loop (the same skeleton as Chapter 107's client): Next(8, &Chunk) segments with a small cap, DATA prints + ChunkRelease, AGAIN waits on xrtHttpBodyReaderWait's Future, EOF finishes. (4) The two output lines mean "the byte stream the producer wrote arrives at the consumer whole" — the body object is the decoupling point between production and consumption: the two ends may live on different threads at different moments. In real deployments the producer is a task thread (SSE events/cursor yields) and the consumer is the HTTP transport.

Second complete program: file-body upload

The second program is from examples/http/body_file — the upload shape for large files:

#include <stdio.h>

#include <xhttp.h>



/* 等待 Future 成功并返回借用值。 */
static ptr exampleFutureValue(xfuture* pFuture)
{
	if ( (pFuture == NULL) ||
		(xrtFutureWait(pFuture) != XWAIT_OK) ||
		(xrtFutureState(pFuture) != XFUTURE_RESOLVED) ) {
		return NULL;
	}
	return xrtFutureValue(pFuture);
}



/* 使用统一正文 Reader 异步读取文件内容。 */
int main(void)
{
	static const char sPath[] = "xrt-http-body-file-example.tmp";
	xtaskpoolconfig Config = { 1, 16, 0 };
	xtaskpool* pPool;
	xfile File;
	xfuture* pPrepare;
	xhttpbody* pBody;
	xhttpbodyreader* pReader;
	xhttpbodychunk Chunk;
	xhttpbodystatus Status;

	(void)xrtFileDelete(sPath);
	xrtClearError();
	File = xrtOpen(
		sPath,
		XFILE_WRITE | XFILE_CREATE | XFILE_EXCLUSIVE
	);
	if ( (File == NULL) ||
		!xrtWriteFull(File, "HTTP file body\n", 15, NULL) ||
		!xrtClose(File) ) {
		return 1;
	}
	pPool = xrtTaskPoolCreate(&Config);
	if ( pPool == NULL ) {
		(void)xrtFileDelete(sPath);
		return 1;
	}
	/* NULL 配置按需读取，默认单次申请不超过 64 KiB。 */
	pPrepare = xrtHttpBodyFileFuture(pPool, sPath, NULL);
	pBody = (xhttpbody*)exampleFutureValue(pPrepare);
	pReader = pBody != NULL ? xrtHttpBodyOpen(pBody) : NULL;
	while ( pReader != NULL ) {
		Status = xrtHttpBodyNext(pReader, 4096, &Chunk);
		if ( Status == XHTTP_BODY_AGAIN ) {
			xfuture* pReady = xrtHttpBodyReaderWait(pReader);

			if ( exampleFutureValue(pReady) == NULL ) {
				xrtFutureDestroy(pReady);
				break;
			}
			xrtFutureDestroy(pReady);
			continue;
		}
		if ( Status == XHTTP_BODY_EOF ) {
			break;
		}
		if ( Status == XHTTP_BODY_ERROR ) {
			break;
		}
		(void)fwrite(Chunk.Data, 1, Chunk.Size, stdout);
		xrtHttpBodyChunkRelease(&Chunk);
	}
	xrtHttpBodyReaderDestroy(pReader);
	xrtFutureDestroy(pPrepare);
	(void)xrtTaskPoolDestroy(pPool);
	(void)xrtFileDelete(sPath);
	return 0;
}

What just happened. (1) xrtHttpBodyFileConfigInit configures path and options, then the file body is created — the length comes from file metadata, the content is read asynchronously on demand: client memory independent of file size (contrast Copy's whole read — 2 GB is 2 GB). (2) File bodies are replayable (reopen and reread) — Chapter 100's 307/308 redirect resend requirement is satisfied here by nature; a fixed streaming producer is not (consumed once). (3) The xrtHttpBodyFileFuture Future form brings "wait for the file to be ready/read done" into Chapter 58's waiting system. (4) Companions: examples/http/body (the fixed family's five sources), body_compose (composed bodies), body_inflate/deflate (transformed bodies — the sending-side compression matching Chapter 93's decoding), server_body_async (server-side async body responses) — the body family's full inspection face.

108.4契约

• Five sources: the fixed family (replayable, known length), the producer stream (consumed once, bounded), files (replayable, async reads), custom, transformed — one unified xhttpbody shape.
• Fixed-family discipline: Copy single allocation (descriptor + copy in one compact block); Borrow extends no external lifetime; Reference must carry a release procedure; View succeeds only for sources guaranteeing contiguity.
• Producer stream: MaxBytes/MaxChunks budget covers reservations + queuing + leases; EOF/Close/Fail three finishes; Write three forms (copy/reference/take-over).
• Reader: single consumer; DATA (lease)/AGAIN (Wait Future)/EOF three states; the Chunk is an independent lease (ChunkRelease; releasable later than Reader/Body — reference-send friendly).
• Aliasing defense: outputs never overlay the body's underlying bytes; aliasing errors advance nothing and break nothing — fixed Bodies stay immutable across replay and concurrent opens.
• File bodies: async reads, no blocking, no whole reads; replayable; Range variants; Future readiness form.
• Replay matrix: 307/308 resend → fixed family/files; generate once → producer stream.
• Budget ownership: the client's ResponseBodyLimit (Chapter 99) caps the representation body; the producer-stream budget caps the producer — two layers, each bounded.
• Trimming: HTTP_BODY/HTTP_BODY_STREAM/HTTP_BODY_FILE independent macros.

Common Mistakes

uint8 Buf[256];
fill(Buf);
xrtHttpRequestSetBody(pReq, xrtHttpBodyBorrow(view(Buf)));  /* stack buffer */
send_async(pReq);   /* the stack frame dies after return: the send reads garbage */

/* either Copy (small bodies) or Reference + release callback (your own heap buffer) */
xrtHttpRequestSetBody(pReq, xrtHttpBodyCopy(view(Buf)));
/* or xrtHttpBodyReference(view(HeapBuf), release_cb, ctx) - released automatically when the send completes */

for ( each_item ) {
	if ( send(stream, item) == AGAIN ) {
		continue;   /* dropped - the client misses this segment */
	}
}

for ( each_item ) {
	while ( send(stream, item) == AGAIN ) {
		wait_writable(stream);   /* the waiting template from Chapter 107, Pitfall 2 */
	}
}

while ( next(&Reader, &Chunk) == DATA ) {
	queue_push(g_Q, Chunk);   /* lease not returned: budget exhausted */
}

while ( next(&Reader, &Chunk) == DATA ) {
	if ( want_keep(Chunk) ) {
		queue_push(g_Q, copy_chunk(Chunk));  /* keep a copy */
	}
	xrtHttpBodyChunkRelease(&Chunk);          /* return the lease */
}

Programming Exercises

Summary