Encoding & Decoding

Streams of bytes or text rarely arrive as the values you actually want to work with — they arrive as raw Uint8Array chunks, NDJSON lines, MessagePack frames, or Server-Sent Events. Effect models each codec as a Channel (a stage that consumes input chunks and produces output chunks) and lets you splice it into a Stream with Stream.pipeThroughChannel, so decoding/encoding becomes just another step in the pipeline.

import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

// Raw NDJSON text — one JSON object per line — as it might arrive from a file
// or socket.
const raw = Stream.make(
  '{"timestamp":"2025-06-01T00:00:00Z","level":"info","message":"start"}\n' +
    '{"timestamp":"2025-06-01T00:00:01Z","level":"error","message":"oops"}\n'
)

// `Ndjson.decodeString()` is a Channel that splits incoming strings on newlines
// and `JSON.parse`s each line. Splice it in with `Stream.pipeThroughChannel`.
export const decoded = raw.pipe(
  Stream.pipeThroughChannel(Ndjson.decodeString()),
  Stream.runCollect
)
// => [{ timestamp, level, message }, { ... }]

Note

Ndjson, Msgpack, and Sse live under effect/unstable/encoding. Unstable modules have stable behaviour but may see API tweaks before they graduate to the core namespace. The lower-level text codecs (Stream.decodeText, Stream.splitLines) and the schema-channel adapters (ChannelSchema) are part of the stable core effect package.

Text codecs on a Stream

Before any framing, you usually need to turn raw bytes into text (or text back into bytes). Stream.decodeText, Stream.encodeText, and Stream.splitLines operate directly on a Stream and need no channel plumbing.

import { Stream } from "effect"

// Bytes in, text out. Internally uses a streaming TextDecoder, so multi-byte
// characters that straddle a chunk boundary are handled correctly.
const bytes = Stream.make(
  new TextEncoder().encode("Hello "),
  new TextEncoder().encode("World")
)

export const text = bytes.pipe(Stream.decodeText(), Stream.runCollect)
// => ["Hello ", "World"]

// Text in, UTF-8 bytes out — the inverse direction.
export const encoded = Stream.make("Hello", " ", "World").pipe(
  Stream.encodeText,
  Stream.runCollect
)
// => [Uint8Array[72,101,108,108,111], Uint8Array[32], Uint8Array[87,111,114,108,100]]

// Split a text stream into lines, handling \n, \r, and \r\n across chunk
// boundaries. Trailing delimiters do not produce a spurious empty line.
export const lines = Stream.make("a\nb\r\n", "c\n").pipe(
  Stream.splitLines,
  Stream.runCollect
)
// => ["a", "b", "c"]

Stream.decodeText accepts an optional encoding, e.g. Stream.decodeText({ encoding: "utf-16le" }), forwarded to the underlying TextDecoder.

Splicing codecs in: pipeThroughChannel

Every codec in this page is a Channel. There are three ways to attach one to a stream; the first is what you will reach for almost every time.

Stream.pipeThroughChannel

Runs the stream’s chunks through a channel and adopts the channel’s error type. Upstream stream errors are passed into the channel, so the resulting error type is just the channel’s (E2).

import { Channel, Stream } from "effect"

// A trivial channel: stringify every chunk element.
const stringify = Channel.identity<
  readonly [number, ...Array<number>],
  never,
  unknown
>().pipe(Channel.map((chunk) => chunk.map(String)))

export const out = Stream.make(1, 2, 3).pipe(
  Stream.pipeThroughChannel(stringify),
  Stream.runCollect
)
// => ["1", "2", "3"]

Stream.pipeThroughChannelOrFail

Same splice, but upstream stream failures are not fed to the channel, so the resulting stream can fail with either the original stream error or the channel error (E | E2). Use this when the channel should only see successful values.

import { Channel, Stream } from "effect"

const stringify = Channel.identity<
  readonly [number, ...Array<number>],
  "StreamError",
  unknown
>().pipe(Channel.map((chunk) => chunk.map(String)))

export const out = Stream.make(1, 2, 3).pipe(
  Stream.rechunk(2),
  Stream.pipeThroughChannelOrFail(stringify),
  Stream.runCollect
)
// => ["1", "2", "3"]

Stream.pipeThrough

A different tool with a confusingly similar name: it pipes the stream through a Sink and emits the sink’s leftovers. Reach for this when a sink stops partway and you want the unconsumed tail as a stream — not for codecs.

import { Sink, Stream } from "effect"

export const leftovers = Stream.make(1, 2, 3, 4).pipe(
  Stream.pipeThrough(Sink.take(2)),
  Stream.runCollect
)
// => [3, 4]

A realistic pipeline

Because decode and encode are both just channels, the common decode → transform → re-encode shape reads as one linear pipeline. Here we decode NDJSON into validated values, filter, then re-encode.

import { Schema, Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

class LogEntry extends Schema.Class<LogEntry>("LogEntry")({
  timestamp: Schema.String,
  level: Schema.Literals(["info", "warn", "error"]),
  message: Schema.String
}) {}

const ndjsonInput = Stream.make(
  '{"timestamp":"2025-06-01T00:00:00Z","level":"info","message":"ok"}\n' +
    '{"timestamp":"2025-06-01T00:00:01Z","level":"error","message":"fail"}\n' +
    '{"timestamp":"2025-06-01T00:00:02Z","level":"warn","message":"slow"}\n'
)

export const errorsOnly = ndjsonInput.pipe(
  // 1. Decode each line into a validated LogEntry.
  Stream.pipeThroughChannel(Ndjson.decodeSchemaString(LogEntry)()),
  // 2. Keep only error-level entries (an ordinary stream transform).
  Stream.filter((entry) => entry.level === "error"),
  // 3. Re-encode the survivors back to NDJSON strings.
  Stream.pipeThroughChannel(Ndjson.encodeSchemaString(LogEntry)()),
  Stream.runCollect
)
// => ['{"timestamp":"2025-06-01T00:00:01Z","level":"error","message":"fail"}\n']

Caution

The schema-aware NDJSON/Msgpack constructors are curried: you call them with the schema, then call the returned function with no arguments to get the channel — Ndjson.decodeSchemaString(LogEntry)(). The trailing () fixes the optional input-error and done type parameters.

---

NDJSON — effect/unstable/encoding/Ndjson

Newline-delimited JSON: one complete JSON value per line. Decoders accumulate partial lines across chunks; encoders append a trailing newline as record framing. Byte helpers handle UTF-8 transport, string helpers handle already-decoded text, and Schema helpers validate or transform each record.

Ndjson.decodeString

Channel that splits incoming string chunks on newlines and JSON.parses each line into unknown. Pass { ignoreEmptyLines: true } to skip blank lines (otherwise a blank line fails as invalid JSON).

import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

export const out = Stream.make('{"a":1}\n', '{"a":2}\n').pipe(
  Stream.pipeThroughChannel(Ndjson.decodeString({ ignoreEmptyLines: true })),
  Stream.runCollect
)
// => [{ a: 1 }, { a: 2 }]

Ndjson.encodeString

Channel that serializes each value with JSON.stringify, joins a chunk with newlines, and appends a trailing newline. The inverse of decodeString.

import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

export const out = Stream.make({ a: 1 }, { a: 2 }).pipe(
  Stream.pipeThroughChannel(Ndjson.encodeString()),
  Stream.runCollect
)
// => ['{"a":1}\n', '{"a":2}\n']

Ndjson.decode

Like decodeString but consumes UTF-8 Uint8Array chunks (it pipes through Channel.decodeText first). Use it for raw byte transports such as sockets or file descriptors.

import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

const bytes = Stream.make(new TextEncoder().encode('{"a":1}\n{"a":2}\n'))

export const out = bytes.pipe(
  Stream.pipeThroughChannel(Ndjson.decode()),
  Stream.runCollect
)
// => [{ a: 1 }, { a: 2 }]

Ndjson.encode

Like encodeString but produces UTF-8 Uint8Array chunks, ready to write to a byte sink.

import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

export const out = Stream.make({ a: 1 }).pipe(
  Stream.pipeThroughChannel(Ndjson.encode()),
  Stream.runCollect
)
// => [Uint8Array(...) /* '{"a":1}\n' as UTF-8 bytes */]

Ndjson.decodeSchemaString / Ndjson.decodeSchema

Parse each NDJSON line, then decode the parsed value with a Schema, yielding a validated, typed value. decodeSchemaString consumes strings; decodeSchema consumes Uint8Array bytes. Validation failures surface as Schema.SchemaError; malformed JSON as NdjsonError.

import { Schema, Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

const User = Schema.Struct({ id: Schema.Number, name: Schema.String })

export const out = Stream.make('{"id":1,"name":"Ada"}\n').pipe(
  Stream.pipeThroughChannel(Ndjson.decodeSchemaString(User)()),
  Stream.runCollect
)
// => [{ id: 1, name: "Ada" }]  (typed as { id: number; name: string })

Ndjson.encodeSchemaString / Ndjson.encodeSchema

Run each value through the schema (applying encoding transformations) and then serialize to NDJSON. encodeSchemaString emits strings; encodeSchema emits UTF-8 bytes.

import { Schema, Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

const User = Schema.Struct({ id: Schema.Number, name: Schema.String })

export const out = Stream.make({ id: 1, name: "Ada" }).pipe(
  Stream.pipeThroughChannel(Ndjson.encodeSchemaString(User)()),
  Stream.runCollect
)
// => ['{"id":1,"name":"Ada"}\n']

Ndjson.duplex / Ndjson.duplexString

Wrap a bidirectional byte (or string) channel — such as a socket — so callers send and receive decoded values while the wrapped channel still moves NDJSON bytes/strings. Outgoing values are encoded; incoming bytes/strings are decoded.

import { Channel } from "effect"
import { Ndjson } from "effect/unstable/encoding"

// `socket` is some byte-level duplex Channel<Uint8Array[], …, Uint8Array[], …>.
declare const socket: Channel.Channel<
  ReadonlyArray<Uint8Array> & { 0: Uint8Array },
  never,
  unknown,
  ReadonlyArray<Uint8Array> & { 0: Uint8Array },
  Ndjson.NdjsonError,
  unknown
>

// Now the channel emits decoded `unknown` values and accepts `unknown` to send.
export const jsonSocket = Ndjson.duplex(socket, { ignoreEmptyLines: true })

Ndjson.duplexSchema / Ndjson.duplexSchemaString

The schema-aware duplex: values you send are encoded with inputSchema, values you receive are decoded with outputSchema. Ideal for typed request/response framing over a single connection.

import { Channel, Schema } from "effect"
import { Ndjson } from "effect/unstable/encoding"

const Request = Schema.Struct({ method: Schema.String })
const Response = Schema.Struct({ ok: Schema.Boolean })

declare const socket: Channel.Channel<
  ReadonlyArray<Uint8Array> & { 0: Uint8Array },
  never,
  unknown,
  ReadonlyArray<Uint8Array> & { 0: Uint8Array },
  Ndjson.NdjsonError | Schema.SchemaError,
  unknown
>

export const typed = Ndjson.duplexSchema(socket, {
  inputSchema: Request,
  outputSchema: Response
})
// send { method } values, receive { ok } values

Ndjson.NdjsonError

Tagged error raised on codec failure. kind is "Unpack" for decode failures and "Pack" for encode failures; cause holds the underlying exception. Catch it with Stream.catchTag (see Error handling).

import { Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

export const recovered = Stream.make("not-valid-json\n").pipe(
  Stream.pipeThroughChannel(Ndjson.decodeString()),
  Stream.catchTag("NdjsonError", (err) =>
    Stream.succeed({ recovered: true, kind: err.kind })
  ),
  Stream.runCollect
)
// => [{ recovered: true, kind: "Unpack" }]

---

MessagePack — effect/unstable/encoding/Msgpack

A compact binary serialization format for protocols and storage that expect bytes rather than JSON text (RPC transports, socket streams, caches, DB columns). The shape mirrors Ndjson, but everything is Uint8Array and there is no string variant.

Msgpack.encode

Channel that packs each value into MessagePack bytes (one byte array per value). Packing failures fail with MsgPackError.

import { Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

export const out = Stream.make({ a: 1 }, { a: 2 }).pipe(
  Stream.pipeThroughChannel(Msgpack.encode()),
  Stream.runCollect
)
// => [Uint8Array(...), Uint8Array(...)]  // two packed frames

Msgpack.decode

Channel that unpacks MessagePack byte chunks into values. Incomplete frames are buffered across chunks until enough bytes arrive, so it tolerates arbitrary chunk boundaries. Invalid data fails with MsgPackError.

import { Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

// Round-trip: encode then decode.
export const out = Stream.make({ a: 1 }, { a: 2 }).pipe(
  Stream.pipeThroughChannel(Msgpack.encode()),
  Stream.pipeThroughChannel(Msgpack.decode()),
  Stream.runCollect
)
// => [{ a: 1 }, { a: 2 }]

Msgpack.encodeSchema

Encode each value through a Schema, then pack as MessagePack bytes. Can fail with either Schema.SchemaError or MsgPackError.

import { Schema, Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const Point = Schema.Struct({ x: Schema.Number, y: Schema.Number })

export const out = Stream.make({ x: 1, y: 2 }).pipe(
  Stream.pipeThroughChannel(Msgpack.encodeSchema(Point)()),
  Stream.runCollect
)
// => [Uint8Array(...)]  // packed { x: 1, y: 2 }

Msgpack.decodeSchema

Unpack bytes into unknown, then decode each value with the schema, producing typed, validated values.

import { Schema, Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const Point = Schema.Struct({ x: Schema.Number, y: Schema.Number })

export const out = Stream.make({ x: 1, y: 2 }).pipe(
  Stream.pipeThroughChannel(Msgpack.encodeSchema(Point)()),
  Stream.pipeThroughChannel(Msgpack.decodeSchema(Point)()),
  Stream.runCollect
)
// => [{ x: 1, y: 2 }]

Msgpack.duplex / Msgpack.duplexSchema

Wrap a bidirectional byte channel with MessagePack framing. duplex exposes unknown values; duplexSchema exposes typed input/output via inputSchema and outputSchema. Same role as the NDJSON duplex helpers.

import { Channel, Schema } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const Req = Schema.Struct({ q: Schema.String })
const Res = Schema.Struct({ hits: Schema.Number })

declare const socket: Channel.Channel<
  ReadonlyArray<Uint8Array<ArrayBuffer>> & { 0: Uint8Array<ArrayBuffer> },
  never,
  unknown,
  ReadonlyArray<Uint8Array<ArrayBuffer>> & { 0: Uint8Array<ArrayBuffer> },
  Msgpack.MsgPackError | Schema.SchemaError,
  unknown
>

export const typed = Msgpack.duplexSchema(socket, {
  inputSchema: Req,
  outputSchema: Res
})

Msgpack.schema / Msgpack.transformation

Msgpack.schema(inner) builds a Schema (not a channel) that stores values as MessagePack bytes: decoding a Uint8Array unpacks then runs inner, encoding runs inner then packs. Msgpack.transformation is the underlying Transformation it is built from. Useful for persisting a MessagePack blob as a struct field or a DB column.

import { Schema } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const Session = Schema.Struct({ userId: Schema.Number, roles: Schema.Array(Schema.String) })

// A schema whose Encoded side is Uint8Array (MessagePack bytes).
const SessionBlob = Msgpack.schema(Session)

export const session = Schema.decodeSync(SessionBlob)(
  Schema.encodeSync(SessionBlob)({ userId: 1, roles: ["admin"] })
)
// => { userId: 1, roles: ["admin"] }

Msgpack.MsgPackError

Tagged error for codec failures. kind is "Pack" or "Unpack"; cause holds the original error. Catch with Stream.catchTag("MsgPackError", …).

import { Stream } from "effect"
import { Msgpack } from "effect/unstable/encoding"

const garbage = Stream.make(new Uint8Array([0xc1]) as Uint8Array<ArrayBuffer>)

export const recovered = garbage.pipe(
  Stream.pipeThroughChannel(Msgpack.decode()),
  Stream.catchTag("MsgPackError", (err) => Stream.succeed(err.kind)),
  Stream.runCollect
)
// => ["Unpack"]

---

Server-Sent Events — effect/unstable/encoding/Sse

SSE is the EventSource wire format: line-oriented text for unidirectional server-to-client streams (live updates, notifications, progress feeds). A blank line dispatches an event; repeated data: lines join with newlines; retry: directives surface as Retry failures.

Sse.decode

Channel that parses string chunks into Sse.Event values. Chunk boundaries may split fields; the parser buffers until a full event is available. A retry: directive fails the channel with Sse.Retry.

import { Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

const wire = Stream.make("event: ping\ndata: hello\n\n")

export const out = wire.pipe(
  Stream.pipeThroughChannel(Sse.decode()),
  Stream.runCollect
)
// => [{ _tag: "Event", event: "ping", id: undefined, data: "hello" }]

Sse.decodeSchema

Decode each parsed event with a schema that receives the untagged { id?, event, data } shape — handy for renaming/validating fields. Fails with Schema.SchemaError, Sse.Retry, or the input error.

import { Schema, Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

// Keep only the event name and data, validating both are strings.
const Named = Schema.Struct({ event: Schema.String, data: Schema.String })

export const out = Stream.make("event: tick\ndata: 1\n\n").pipe(
  Stream.pipeThroughChannel(Sse.decodeSchema(Named)),
  Stream.runCollect
)
// => [{ event: "tick", data: "1" }]

Sse.decodeDataSchema

The most common SSE need: JSON-decode each event’s data field with a schema, keeping the event name and optional id. The output data is the decoded domain value.

import { Schema, Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

const Progress = Schema.Struct({ pct: Schema.Number })

export const out = Stream.make(
  'event: progress\ndata: {"pct":42}\n\n'
).pipe(
  Stream.pipeThroughChannel(Sse.decodeDataSchema(Progress)),
  Stream.runCollect
)
// => [{ event: "progress", id: undefined, data: { pct: 42 } }]

Sse.encode

Channel that renders Sse.Event values as SSE text. An upstream Sse.Retry failure is written as a retry: directive and completes the encoder.

import { Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

const event: Sse.Event = {
  _tag: "Event",
  event: "ping",
  id: "1",
  data: "hello"
}

export const out = Stream.make(event).pipe(
  Stream.pipeThroughChannel(Sse.encode()),
  Stream.runCollect
)
// => ["id: 1\nevent: ping\ndata: hello\n\n"]

Sse.encodeSchema

Schema-encode domain values to the untagged SSE shape, convert to Event, then render as SSE text. The inverse pairing for decodeSchema.

import { Schema, Stream } from "effect"
import { Sse } from "effect/unstable/encoding"

const Named = Schema.Struct({ event: Schema.String, data: Schema.String })

export const out = Stream.make({ event: "tick", data: "1" }).pipe(
  Stream.pipeThroughChannel(Sse.encodeSchema(Named)),
  Stream.runCollect
)
// => ["event: tick\ndata: 1\n\n"]

Sse.makeParser / Sse.Parser

The stateful parser underneath decode, for manual feeding outside a stream. Call feed(chunk) to push text; your callback receives Sse.Event or Sse.Retry values. reset() clears buffered state. Parser is the returned interface.

import { Sse } from "effect/unstable/encoding"

const events: Array<Sse.AnyEvent> = []
const parser: Sse.Parser = Sse.makeParser((event) => events.push(event))

parser.feed("data: a\n\n")
parser.feed("data: b\n\n")
// events => [{ _tag: "Event", data: "a", ... }, { _tag: "Event", data: "b", ... }]

Sse.encoder / Sse.Encoder

The default Encoder: its write(event) turns a single Sse.Event or Sse.Retry into SSE text. Sse.encode() uses it internally; reach for encoder when you render one event at a time. The event: line is omitted for the default "message" name.

import { Duration } from "effect"
import { Sse } from "effect/unstable/encoding"

export const line = Sse.encoder.write({
  _tag: "Event",
  event: "message",
  id: undefined,
  data: "hi"
})
// => "data: hi\n\n"   (no `event:` line for the default name)

export const retry = Sse.encoder.write(
  new Sse.Retry({ duration: Duration.seconds(3), lastEventId: undefined })
)
// => "retry: 3000\n\n"

Sse.Event / Sse.EventEncoded / Sse.AnyEvent

Sse.Event is both the interface for a parsed event ({ _tag: "Event", event, id, data }) and a Schema for that tagged shape. Sse.EventEncoded is the untagged { id?, event, data } shape (interface + schema). Sse.AnyEvent is the union Event | Retry that an Encoder can render.

import { Schema } from "effect"
import { Sse } from "effect/unstable/encoding"

// `Sse.EventEncoded` is a real Schema you can decode/encode with.
export const decoded = Schema.decodeUnknownSync(Sse.EventEncoded)({
  id: "7",
  event: "ping",
  data: "x"
})
// => { id: "7", event: "ping", data: "x" }

Sse.Retry

Tagged class representing a retry: directive. Decoders surface it as a failure (reconnect after duration); encoders serialize an upstream Retry failure as a retry: line. Sse.Retry.is is a guard and Sse.Retry.filter separates retries from events.

import { Duration } from "effect"
import { Sse } from "effect/unstable/encoding"

const r = new Sse.Retry({
  duration: Duration.seconds(5),
  lastEventId: "42"
})

export const isRetry = Sse.Retry.is(r) // => true
export const ms = Duration.toMillis(r.duration) // => 5000

Sse.transformEvent

The Transformation that maps the untagged SSE payload to the tagged Event model. encodeSchema uses it internally; you rarely call it directly, but it is exported for building custom schema pipelines.

import { Sse } from "effect/unstable/encoding"

// `Sse.transformEvent` is a SchemaTransformation used inside `Sse.encodeSchema`;
// reference it when composing your own Event schema.
export const t = Sse.transformEvent

---

Schema-driven channel codecs — ChannelSchema

ChannelSchema (in core effect) is the generic machinery the NDJSON/Msgpack schema helpers are built on: it applies a Schema at a Channel boundary so one side works with typed values while the other works with the encoded representation. Schemas are applied to non-empty chunks, and schema services become part of the channel’s requirements. Use it directly when you have a custom transport whose encoded form is already structured data.

ChannelSchema.encode

Channel that encodes typed input chunks (S["Type"]) into the schema’s encoded representation (S["Encoded"]). Failures surface as Schema.SchemaError.

import { ChannelSchema, Schema, Stream } from "effect"

// Type side is `number`; Encoded side is `string`.
const Count = Schema.NumberFromString

export const out = Stream.make(1, 2, 3).pipe(
  Stream.pipeThroughChannel(ChannelSchema.encode(Count)()),
  Stream.runCollect
)
// => ["1", "2", "3"]  (encoded to the schema's Encoded representation)

ChannelSchema.encodeUnknown

Identical to encode but the encoded output chunks are typed as unknown — for boundaries where the encoded representation is intentionally erased.

import { ChannelSchema, Schema, Stream } from "effect"

const User = Schema.Struct({ id: Schema.Number })

export const out = Stream.make({ id: 1 }).pipe(
  Stream.pipeThroughChannel(ChannelSchema.encodeUnknown(User)()),
  Stream.runCollect
)
// => [{ id: 1 }]  (typed as unknown[])

ChannelSchema.decode

Channel that decodes encoded input chunks (S["Encoded"]) into typed schema values (S["Type"]), validating along the way.

import { ChannelSchema, Schema, Stream } from "effect"

const User = Schema.Struct({ id: Schema.Number, name: Schema.String })

export const out = Stream.make({ id: 1, name: "Ada" }).pipe(
  Stream.pipeThroughChannel(ChannelSchema.decode(User)()),
  Stream.runCollect
)
// => [{ id: 1, name: "Ada" }]  (validated)

ChannelSchema.decodeUnknown

decode for boundaries where the encoded input is unknown (e.g. freshly JSON.parsed data). Only the decoded output is statically typed.

import { ChannelSchema, Schema, Stream } from "effect"

const User = Schema.Struct({ id: Schema.Number })

// Simulate untyped input from a parser.
const parsed: ReadonlyArray<unknown> = [{ id: 1 }, { id: 2 }]

export const out = Stream.fromArray(parsed).pipe(
  Stream.pipeThroughChannel(ChannelSchema.decodeUnknown(User)()),
  Stream.runCollect
)
// => [{ id: 1 }, { id: 2 }]

ChannelSchema.duplex

Wrap a bidirectional encoded channel so callers send/receive typed chunks: input is encoded with inputSchema, output is decoded with outputSchema. The wrapped channel keeps operating on the schemas’ encoded chunks.

import { ChannelSchema, Channel, Schema } from "effect"

const Req = Schema.Struct({ q: Schema.String })
const Res = Schema.Struct({ n: Schema.Number })

// Wrapped channel speaks the encoded representations.
declare const transport: Channel.Channel<
  ReadonlyArray<typeof Res.Encoded> & { 0: typeof Res.Encoded },
  never,
  unknown,
  ReadonlyArray<typeof Req.Encoded> & { 0: typeof Req.Encoded },
  Schema.SchemaError,
  unknown
>

export const typed = ChannelSchema.duplex(transport, {
  inputSchema: Req,
  outputSchema: Res
})
// send { q } values, receive { n } values

ChannelSchema.duplexUnknown

duplex for a wrapped channel whose encoded chunks are typed as unknown on both sides — the variant the Ndjson/Msgpack duplexSchema helpers build on.

import { ChannelSchema, Channel, Schema } from "effect"

const Req = Schema.Struct({ q: Schema.String })
const Res = Schema.Struct({ n: Schema.Number })

declare const transport: Channel.Channel<
  ReadonlyArray<unknown> & { 0: unknown },
  never,
  unknown,
  ReadonlyArray<unknown> & { 0: unknown },
  Schema.SchemaError,
  unknown
>

export const typed = ChannelSchema.duplexUnknown(transport, {
  inputSchema: Req,
  outputSchema: Res
})