Body.Stream

This module defines a zero-buffer rendezvous body channel (Body.Stream) that supports both sending and receiving chunks.

There is no queue and no capacity. A send waits for a receiver and a receive waits for a sender. At most one blocked producer and one blocked consumer are supported.

structure Std.Http.Body.Stream : Type

A zero-buffer rendezvous body channel that supports both sending and receiving chunks.

• state : Std.Mutex Channel.State

instance Std.Http.Body.instNonemptyStream : Nonempty Stream

@[implicit_reducible]

instance Std.Http.Body.instTypeNameStream : TypeName Stream

def Std.Http.Body.mkStream : Internal.IO.Async.Async Stream

Creates a rendezvous body stream.

def Std.Http.Body.Stream.tryRecv (stream : Stream) : Internal.IO.Async.Async (Option Chunk)

Attempts to receive a chunk from the channel without blocking. Returns some chunk only when a producer is already waiting.

def Std.Http.Body.Stream.recv (stream : Stream) : Internal.IO.Async.Async (Option Chunk)

Receives a chunk from the channel. Blocks until a producer sends one. Returns none if the channel is closed and no producer is waiting.

def Std.Http.Body.Stream.close (stream : Stream) : Internal.IO.Async.Async Unit

Closes the channel.

@[inline]

def Std.Http.Body.Stream.isClosed (stream : Stream) : Internal.IO.Async.Async Bool

Checks whether the channel is closed.

@[inline]

def Std.Http.Body.Stream.getKnownSize (stream : Stream) : Internal.IO.Async.Async (Option Length)

Gets the known size if available.

@[inline]

def Std.Http.Body.Stream.setKnownSize (stream : Stream) (size : Option Length) : Internal.IO.Async.Async Unit

Sets known size metadata.

def Std.Http.Body.Stream.recvSelector (stream : Stream) : Internal.IO.Async.Selector (Option Chunk)

Creates a selector that resolves when a producer is waiting (or the channel closes).

@[inline]

def Std.Http.Body.Stream.forIn {β : Type} (stream : Stream) (acc : β) (step : Chunk → β → Internal.IO.Async.Async (ForInStep β)) : Internal.IO.Async.Async β

Iterates over chunks until the channel closes.

@[inline]

def Std.Http.Body.Stream.forIn' {β : Type} (stream : Stream) (acc : β) (step : Chunk → β → Internal.IO.Async.ContextAsync (ForInStep β)) : Internal.IO.Async.ContextAsync β

Context-aware iteration over chunks until the channel closes.

class Std.Http.Body.Stream.NextChunk (m : Type → Type) : Type

Abstracts over how the next chunk is received, allowing readAll to work in both Async (no cancellation) and ContextAsync (races with cancellation via doneSelector).

• nextChunk : Stream → m (Option Chunk)

  Receives the next chunk, stopping at EOF or (in ContextAsync) when the context is cancelled.

@[implicit_reducible]

instance Std.Http.Body.Stream.instNextChunkAsync : NextChunk Internal.IO.Async.Async

@[implicit_reducible]

instance Std.Http.Body.Stream.instNextChunkContextAsync : NextChunk Internal.IO.Async.ContextAsync

def Std.Http.Body.Stream.readAll {α : Type} {m : Type → Type} [FromByteArray α] [Monad m] [MonadExceptOf IO.Error m] [NextChunk m] (stream : Stream) (maximumSize : Option UInt64 := none) : m α

Reads all remaining chunks and decodes them into α.

Works in both Async (reads until EOF, no cancellation) and ContextAsync (also stops if the context is cancelled).

def Std.Http.Body.Stream.send (stream : Stream) (chunk : Chunk) (incomplete : Bool := false) : Internal.IO.Async.Async Unit

Sends a chunk.

If incomplete := true, the chunk is buffered and collapsed with subsequent chunks, and is not delivered to the receiver yet. If incomplete := false, any buffered incomplete pieces are collapsed with this chunk and the single merged chunk is sent.

def Std.Http.Body.Stream.hasInterest (stream : Stream) : Internal.IO.Async.Async Bool

Returns true when a consumer is currently blocked waiting for data.

def Std.Http.Body.Stream.interestSelector (stream : Stream) : Internal.IO.Async.Selector Bool

Creates a selector that resolves when consumer interest is present. Returns true when a consumer is waiting, false when the channel closes first.

def Std.Http.Body.stream (gen : Stream → Internal.IO.Async.Async Unit) : Internal.IO.Async.Async Stream

Creates a body from a producer function. Returns the stream immediately and runs gen in a detached task. The channel is always closed when gen returns or throws. Errors from gen are not rethrown here; consumers observe end-of-stream via recv = none.

def Std.Http.Body.fromBytes (content : ByteArray) : Internal.IO.Async.Async Stream

Creates a body from a fixed byte array.

def Std.Http.Body.empty : Internal.IO.Async.Async Stream

Creates an empty Stream body channel (already closed, no data).

Prefer Body.Empty when you need a concrete zero-cost type. Use this when the calling context requires a Stream specifically.

@[implicit_reducible]

instance Std.Http.Body.instForInAsyncStreamChunk : ForIn Internal.IO.Async.Async Stream Chunk

@[implicit_reducible]

instance Std.Http.Body.instForInContextAsyncStreamChunk : ForIn Internal.IO.Async.ContextAsync Stream Chunk

@[implicit_reducible]

instance Std.Http.Body.instStream : Body Stream

@[implicit_reducible]

instance Std.Http.Body.instCoeStreamAny : Coe Stream Any

@[implicit_reducible]

instance Std.Http.Body.instCoeResponseStreamAny : Coe (Response Stream) (Response Any)

@[implicit_reducible]

instance Std.Http.Body.instCoeContextAsyncResponseStreamAny : Coe (Internal.IO.Async.ContextAsync (Response Stream)) (Internal.IO.Async.ContextAsync (Response Any))

@[implicit_reducible]

instance Std.Http.Body.instCoeAsyncResponseStreamContextAsyncAny : Coe (Internal.IO.Async.Async (Response Stream)) (Internal.IO.Async.ContextAsync (Response Any))

def Std.Http.Request.Builder.stream (builder : Builder) (gen : Body.Stream → Internal.IO.Async.Async Unit) : Internal.IO.Async.Async (Request Body.Stream)

Builds a request with a streaming body generator.

def Std.Http.Response.Builder.stream (builder : Builder) (gen : Body.Stream → Internal.IO.Async.Async Unit) : Internal.IO.Async.Async (Response Body.Stream)

Builds a response with a streaming body generator.