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

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

Equations

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

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

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'✝

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.

Equations

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

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.

Equations

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

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

@[instance_reducible]

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

Equations

• Std.Http.Body.Stream.instNextChunkAsync = { nextChunk := fun (stream : Std.Http.Body.Stream) => stream.recv }

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

Equations

• stream.drain drainLimit closeStream = Std.Http.Body.Stream.drain.loop✝ stream drainLimit closeStream 0

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

Equations

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

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

@[instance_reducible]

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

Equations

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

@[instance_reducible]

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

Equations

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

@[instance_reducible]

instance Std.Http.Body.instStream : Body Stream

Equations

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

@[instance_reducible]

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

Equations

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

@[instance_reducible]

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

Equations

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

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

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