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 : Internal.IO.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) : 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.

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

Equations

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

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

Equations

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

@[implicit_reducible]

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

Equations

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

def Std.Http.Body.Stream.hasInterest (stream : Stream) : Internal.IO.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) : 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.

Equations

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

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.

Equations

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

def Std.Http.Body.fromBytes (content : ByteArray) : Internal.IO.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 : 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.

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 Internal.IO.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 Internal.IO.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 (Internal.IO.Async.ContextAsync (Response Stream)) (Internal.IO.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 (Internal.IO.Async.Async (Response Stream)) (Internal.IO.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 → Internal.IO.Async.Async Unit) : Internal.IO.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 → Internal.IO.Async.Async Unit) : Internal.IO.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)