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

Equations

• Std.Http.Body.instTypeNameStream = Std.Http.Body.inst✝

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

Creates a rendezvous body stream.

Equations

• One or more equations did not get rendered due to their size.

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.

Equations

• stream.tryRecv = (Std.Http.Body.Stream.state✝ stream).atomically do Std.Http.Body.Channel.pruneFinishedWaiters✝ Std.Http.Body.Channel.tryRecv'✝

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).

Equations

• One or more equations did not get rendered due to their size.

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.

Equations

• stream.recv = do let __do_lift ← liftM (Std.Http.Body.Stream.recv'✝ stream) Std.Async.Async.ofAsyncTask __do_lift

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

Closes the channel.

Equations

• stream.close = (Std.Http.Body.Stream.state✝ stream).atomically Std.Http.Body.Channel.close'✝

@[inline]

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

Checks whether the channel is closed.

Equations

• stream.isClosed = (Std.Http.Body.Stream.state✝ stream).atomically do let __do_lift ← get pure (Std.Http.Body.Channel.State.closed✝ __do_lift)

@[inline]

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

Gets the known size if available.

Equations

• stream.getKnownSize = (Std.Http.Body.Stream.state✝ stream).atomically do let __do_lift ← get pure (Std.Http.Body.Channel.State.knownSize✝ __do_lift)

@[inline]

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

Sets known size metadata.

Equations

• One or more equations did not get rendered due to their size.

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).

Equations

• One or more equations did not get rendered due to their size.

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

Equations

• stream.forIn acc step = Std.Http.Body.Stream.forIn.loop✝ step stream acc

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

Equations

• stream.forIn' acc step = Std.Http.Body.Stream.forIn'.loop✝ step stream acc

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

Equations

• Std.Http.Body.Stream.instNextChunkAsync = { nextChunk := Std.Http.Body.Stream.recv }

@[implicit_reducible]

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

Equations

• One or more equations did not get rendered due to their size.

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).

Equations

• One or more equations did not get rendered due to their size.

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.

Equations

• One or more equations did not get rendered due to their size.

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

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

Equations

• stream.hasInterest = (Std.Http.Body.Stream.state✝ stream).atomically do Std.Http.Body.Channel.pruneFinishedWaiters✝ Std.Http.Body.Channel.hasInterest'✝

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.

Equations

• One or more equations did not get rendered due to their size.

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.

Equations

• Std.Http.Body.stream gen = do let s ← Std.Http.Body.mkStream Std.Async.background (tryFinally (gen s) s.close) pure s

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

Creates a body from a fixed byte array.

Equations

• One or more equations did not get rendered due to their size.

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.

Equations

• Std.Http.Body.empty = do let s ← Std.Http.Body.mkStream s.setKnownSize (some (Std.Http.Body.Length.fixed 0)) s.close pure s

@[implicit_reducible]

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

Equations

• Std.Http.Body.instForInAsyncStreamChunk = { forIn := fun {β : Type} => Std.Http.Body.Stream.forIn }

@[implicit_reducible]

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

Equations

• Std.Http.Body.instForInContextAsyncStreamChunk = { forIn := fun {β : Type} => Std.Http.Body.Stream.forIn' }

@[implicit_reducible]

instance Std.Http.Body.instStream : Body Stream

Equations

• One or more equations did not get rendered due to their size.

@[implicit_reducible]

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

Equations

• Std.Http.Body.instCoeStreamAny = { coe := Std.Http.Body.Any.ofBody }

@[implicit_reducible]

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

Equations

• One or more equations did not get rendered due to their size.

@[implicit_reducible]

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

Equations

• One or more equations did not get rendered due to their size.

@[implicit_reducible]

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

Equations

• One or more equations did not get rendered due to their size.

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.

Equations

• builder.stream gen = do let s ← Std.Http.Body.stream gen pure (builder.body s)

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.

Equations

• builder.stream gen = do let s ← Std.Http.Body.stream gen pure (builder.body s)