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

@[instance_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. If the stream has been closed with a terminal error, throws that error instead of returning none.

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. If the channel was closed with a terminal error (via closeWithError) after all buffered chunks have been consumed, that error is thrown instead of EOF.

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

Closes the channel.

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

Closes the channel only if no consumer is currently registered (i.e., the stream was abandoned without being read). If a consumer is waiting, the close is skipped so the pending recv can still receive data. Returns true if the stream was closed.

def Std.Http.Body.Stream.closeWithError (stream : Stream) (err : IO.Error) : Async.Async Unit

Closes the channel and records err as a terminal error. A subsequent recv that would have returned end-of-stream surfaces the error instead; already buffered chunks are still delivered first. This function does not overwrite an existing error.

@[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.

@[instance_reducible]

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

@[instance_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.drain {m : Type → Type} [Monad m] [NextChunk m] (stream : Stream) (drainLimit : Option UInt64 := none) (closeStream : m Unit := pure ()) : m Unit

Drains all remaining chunks from the stream, discarding their contents.

If drainLimit is given, at most that many bytes are consumed; when the limit is exceeded the stream is closed (via closeStream) and no error is raised.

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 closed when gen returns. If gen throws, the channel is closed with that terminal error so consumers observe the failure instead of a clean end-of-stream.

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.

@[instance_reducible]

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

@[instance_reducible]

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

@[instance_reducible]

instance Std.Http.Body.instStream : Body Stream

@[instance_reducible]

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

@[instance_reducible]

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

@[instance_reducible]

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

@[instance_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.