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 : Async.Async Stream

Creates a rendezvous body stream.

def Std.Http.Body.Stream.tryRecv (stream : Stream) : 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.tryRecvBody (stream : Stream) : Async.Async (Option (Option Chunk))

Non-blocking receive for the Body typeclass. Returns none when no producer is waiting and the channel is still open, some (some chunk) when data is ready, or some none at end-of-stream (channel closed with no pending producer).

def Std.Http.Body.Stream.recv (stream : Stream) : 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) : Async.Async Unit

Closes the channel.

@[inline]

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

Checks whether the channel is closed.

@[inline]

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

Gets the known size if available.

@[inline]

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

Sets known size metadata.

def Std.Http.Body.Stream.recvSelector (stream : Stream) : 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 → β → Async.Async (ForInStep β)) : Async.Async β

Iterates over chunks until the channel closes.

@[inline]

def Std.Http.Body.Stream.forIn' {β : Type} (stream : Stream) (acc : β) (step : Chunk → β → Async.ContextAsync (ForInStep β)) : 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 Async.Async

@[implicit_reducible]

instance Std.Http.Body.Stream.instNextChunkContextAsync : NextChunk 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) : 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) : Async.Async Bool

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

def Std.Http.Body.Stream.interestSelector (stream : Stream) : 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 → Async.Async Unit) : 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) : Async.Async Stream

Creates a body from a fixed byte array.

def Std.Http.Body.empty : 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 Async.Async Stream Chunk

@[implicit_reducible]

instance Std.Http.Body.instForInContextAsyncStreamChunk : ForIn 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 (Async.ContextAsync (Response Stream)) (Async.ContextAsync (Response Any))

@[implicit_reducible]

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

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

Builds a request with a streaming body generator.

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

Builds a response with a streaming body generator.