pig_transport

Shared HTTP transport primitives.

A stream is opened as a lightweight process hand-off. The caller gets a cancellable opaque handle immediately; the upstream head is delivered as an event once the first byte commits a 2xx response. The relay never sends body chunks to the sink before start and emits at most one terminal.

Types

Event

</>

Events delivered to a consumer of a stream handle.

pub type Event {
  Committed(status: Int, headers: List(#(String, String)))
  Rejected(
    status: Int,
    headers: List(#(String, String)),
    body: BitArray,
  )
  Failed(reason: String)
  Chunk(data: BitArray)
  Done
  StreamError(reason: String)
  Cancelled
}

Constructors

```
Committed(status: Int, headers: List(#(String, String)))
```
A 2xx response committed after its first body byte (or an empty body).

Rejected(
  status: Int,
  headers: List(#(String, String)),
  body: BitArray,
)

A complete non-2xx response body, which is not committed for streaming.

```
Failed(reason: String)
```
A failure before a streaming response committed.
```
Chunk(data: BitArray)
```
An ordered body chunk after the consumer starts the relay.
```
Done
```
The one successful terminal for a committed stream.
```
StreamError(reason: String)
```
The one failed terminal for a committed stream.
```
Cancelled
```
The one cancellation terminal when cancellation can be delivered.

Request

</>

A single upstream request.

pub type Request {
  Request(
    method: String,
    url: String,
    headers: List(#(String, String)),
    body: String,
    timeout_ms: Int,
  )
}

Constructors

Request(
  method: String,
  url: String,
  headers: List(#(String, String)),
  body: String,
  timeout_ms: Int,
)

Response

</>

The outcome of one buffered upstream request.

pub type Response {
  Response(
    status: Int,
    headers: List(#(String, String)),
    body: BitArray,
  )
  TransportError(reason: String)
}

Constructors

Response(
  status: Int,
  headers: List(#(String, String)),
  body: BitArray,
)

```
TransportError(reason: String)
```

SourceControl

</>

Cancellation sent to a source adapter before its process is stopped.

pub type SourceControl {
  CancelSource
}

Constructors

```
CancelSource
```

SourceEvent

</>

Raw events produced by a stream adapter and consumed by the relay.

pub type SourceEvent {
  SourceReady(control: process.Subject(SourceControl))
  SourceHead(status: Int, headers: List(#(String, String)))
  SourceChunk(data: BitArray)
  SourceDone
  SourceError(reason: String)
}

Constructors

```
SourceReady(control: process.Subject(SourceControl))
```
Provides a source-owned cancellation subject for adapters that can close their underlying connection before their process is killed.

SourceHead(status: Int, headers: List(#(String, String)))

```
SourceChunk(data: BitArray)
```
```
SourceDone
```
```
SourceError(reason: String)
```

StreamHandle

opaque </>

An opaque, idempotently cancellable stream handle.

pub opaque type StreamHandle

Transport

</>

A synchronous adapter plus a streaming source adapter.

pub type Transport {
  Transport(
    sync: fn(Request) -> Response,
    stream: fn(Request, process.Subject(SourceEvent)) -> Nil,
  )
}

Constructors

Transport(
  sync: fn(Request) -> Response,
  stream: fn(Request, process.Subject(SourceEvent)) -> Nil,
)

Values

cancel

</>

pub fn cancel(handle: StreamHandle) -> Nil

Cancel a stream. Repeated cancellation is harmless and produces no second terminal event.

events

</>

pub fn events(handle: StreamHandle) -> process.Subject(Event)

Get the head subject, primarily for consumers that use a selector.

open

</>

pub fn open(
  transport: Transport,
  request: Request,
) -> StreamHandle

Open a stream without performing upstream IO on the caller process.

The returned handle’s events subject delivers Committed, Rejected, or Failed for the head decision. Call start only after Committed; body chunks and exactly one terminal then go to the sink.

receive

</>

pub fn receive(
  handle: StreamHandle,
  timeout_ms: Int,
) -> Result(Event, Nil)

Receive the next lifecycle event from a handle’s head subject.

start

</>

pub fn start(
  handle: StreamHandle,
  sink: process.Subject(Event),
) -> Nil

Start forwarding body chunks to sink.

Repeating the same start is harmless. A different sink gets a deterministic pre-start terminal instead of being left waiting for a stream it cannot own.

sync

</>

pub fn sync(transport: Transport, request: Request) -> Response

Perform one synchronous upstream request through a transport.