transport

@kyneta/transport

Transport infrastructure for @kyneta/exchange — base class, channel types, message vocabulary, identity types, client utilities, wire pipeline, and bridge transport.

What's in this package

Module	Contents
Identity types	PeerId, DocId, ChannelId, TransportType, PeerIdentityDetails
Message vocabulary	ChannelMsg, PresentMsg, InterestMsg, OfferMsg, DismissMsg, EstablishmentMsg, AddressedEnvelope, ReturnEnvelope
Channel types	Channel, ConnectedChannel, EstablishedChannel, GeneratedChannel, ChannelDirectory
Transport base class	Transport<G>, TransportFactory, TransportContext
Wire pipeline	Pipeline<S, R>, FrameStreamParser, Encoding, PayloadOf, WireOpts
Client state machine	StateTransition<S>, TransitionListener<S> (re-exported from @kyneta/machine)
Reconnection	computeBackoffDelay, shouldReconnect, ReconnectDecision, ReconnectOptions, DEFAULT_RECONNECT, JITTER_FRACTION
Re-exports from wire	Result, Ok, Err, ok, err, WireError
Bridge transport	Moved to @kyneta/bridge-transport — codec-faithful + alias-aware in-process testing

Wire pipeline

The Pipeline<S, R> class is the single wire pipeline for all transports. It composes alias resolution, wire-message encoding, fragmentation, and validation into a send/receive pair that transforms ChannelMsg ↔ wire pieces.

Four shapes cover every transport:

import { Pipeline } from "@kyneta/transport"

// Most transports (WebSocket, WebRTC, Unix socket)
const symmetric = new Pipeline({ send: "binary" })

// Symmetric text (if needed)
const text = new Pipeline({ send: "text" })

// SSE server — sends text downstream, receives binary CBOR uploads
const sseServer = new Pipeline({ send: "text", receive: "binary" })

// SSE client — sends binary CBOR uploads, receives text downstream
const sseClient = new Pipeline({ send: "binary", receive: "text" })

The type parameter S is the send encoding and R is the receive encoding (defaults to S for symmetric pipelines). PayloadOf<E> maps "binary" → Uint8Array and "text" → string.

Who depends on this

@kyneta/transport  (defines Transport, Channel, ChannelMsg, Pipeline, ...)
    ↑           ↑           ↑            ↑
@kyneta/exchange  @kyneta/bridge-transport  @kyneta/websocket-transport  ...
(Synchronizer)    (in-process testing)       (extends Transport)

• @kyneta/wire — @kyneta/transport depends on @kyneta/wire (workspace). Wire is a leaf — it provides format primitives. Transport builds the pipeline on top.
• @kyneta/exchange — depends on @kyneta/transport and re-exports its infrastructure (identity types, message vocabulary, channel types, Transport base class, reconnection utilities). The exchange adds the sync runtime (Synchronizer, Exchange, TransportManager) on top.
• Transport implementations (@kyneta/bridge-transport, @kyneta/websocket-transport, @kyneta/sse-transport, @kyneta/unix-socket-transport, @kyneta/webrtc-transport) — peer-depend on @kyneta/transport for the Transport<G> base class, channel types, message vocabulary, and Pipeline.

Creating a transport

Extend the Transport<G> base class:

import { Transport, type GeneratedChannel } from "@kyneta/transport"

class MyTransport extends Transport<void> {
  constructor() {
    super({ transportType: "my-transport" })
  }

  generate(): GeneratedChannel {
    return {
      transportType: this.transportType,
      send: (msg) => { /* send over your wire */ },
      stop: () => { /* cleanup */ },
    }
  }

  async onStart() {
    const channel = this.addChannel(undefined)
    this.establishChannel(channel.channelId)
  }

  async onStop() {
    // cleanup
  }
}

Dependencies

{
  "dependencies": {
    "@kyneta/wire": "workspace:^",
    "@kyneta/random": "workspace:^"
  }
}

@kyneta/schema and @kyneta/machine are peer dependencies (needed for SyncMode, ReplicaType, SubstratePayload, and the state-machine types used in message definitions).

License

MIT