pipe

package module
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 30, 2026 License: MIT Imports: 24 Imported by: 0

README

pipe

Reliable WebRTC DataChannels behind ordinary Go networking APIs.

pipe gives you net.Conn and net.Listener over WebRTC. SDP, ICE, DTLS, SCTP, and Pion's callback API stay inside the library; your code dials a peer by name and reads and writes bytes.

hub := memory.New() // any pipe.Signaler works; memory keeps this to one process

server, err := pipe.New(ctx, pipe.Config{ID: "bob", Signaler: hub})
if err != nil {
	return err
}
defer server.Close()

ln, err := server.Listen()
if err != nil {
	return err
}
defer ln.Close()

go func() {
	for {
		conn, err := ln.Accept()
		if err != nil {
			return
		}
		go io.Copy(conn, conn) // an echo server, unchanged from TCP
	}
}()

client, err := pipe.New(ctx, pipe.Config{ID: "alice", Signaler: hub})
if err != nil {
	return err
}
defer client.Close()

conn, err := client.Dial(ctx, "bob")
if err != nil {
	return err
}
defer conn.Close()

A runnable version is in examples/echo:

go run ./examples/echo

examples/ also has a STUN + TURN server with a configurable throughput budget and a client that proves it went through the relay:

go run ./examples/turnserver -rate 512KiB -stats 2s        # admin/admin on :3478
go run ./examples/turnclient -rate 512KiB -bytes 1MiB      # self-contained

What you get

A pipe.Conn is a real net.Conn: deadlines work, Close is idempotent, one reader and one writer may run concurrently, and errors are *net.OpError with network name webrtc. That is verified rather than asserted — test/compat drives io.Copy, bufio, encoding/json, encoding/gob, tls.Conn (TLS 1.3 handshake and transfer), and net/http (with keep-alive) over live connections.

Signaling

Two peers cannot find each other without a third party to carry offers, answers, and candidates. pipe does not ship a mandatory signaling service; you provide a Signaler:

type Signaler interface {
	Open(ctx context.Context, local PeerID) (SignalConn, error)
}

type SignalConn interface {
	Send(ctx context.Context, msg Signal) error
	Receive(ctx context.Context) (Signal, error)
	Close() error
}

The rules a transport must honor: one Receive caller at a time, Send safe concurrently with Receive, both honoring their context, Close idempotent and unblocking both. Delivery is at-least-once — duplicates and reordering are expected and tolerated.

Writing a transport? Run the conformance suite against it:

func TestConformance(t *testing.T) {
	signalertest.Run(t, signalertest.Config{
		NewSignaler:            func(t *testing.T) pipe.Signaler { return myTransport(t) },
		RejectsDuplicatePeers:  true,
		ReportsUnavailablePeer: true,
	})
}

signaling/memory is an in-process hub for tests and same-process examples. It also injects duplicate, drop, reorder, and disconnect faults deterministically.

Signaling is not STUN and not TURN. STUN and TURN servers are configured through Config.ICEServers and are used by ICE to find a network path; signaling is how the two peers exchange the descriptions in the first place.

Configuration

Field Default Meaning
ID required This endpoint's peer ID (a routing name).
Signaler required Transport for signaling envelopes.
ICEServers none STUN/TURN servers passed to ICE.
ICETransportPolicy all Set to relay-only to force TURN.
DialTimeout 30s Bound on a whole dial.
ICETimeout 20s Bound on connectivity establishment.
KeepAlive off Protocol ping/pong probes; set Interval to enable.
Reconnect 3 attempts ICE-restart recovery budget and backoff.
AcceptBacklog 64 Pending inbound connections before rejection.
FramePayload 16 KiB Bytes per DataChannel message.
ReadBuffer 1 MiB Unread-byte budget per connection.
Logger, Metrics nop *slog.Logger and a metrics sink.
Pion none Escape hatch for the underlying Pion configuration.

TURN credentials come from your configuration or environment. They are never logged, and neither are SDP, ICE credentials, or application payloads.

Limitations

Read these before deploying.

  • A peer ID is a routing identity, not an authenticated one. Authentication belongs to your signaling transport. DTLS fingerprint verification guarantees that only the negotiated party can send you bytes; it does not tell you who that party is. For cryptographic peer identity, run mutual TLS over the pipe.
  • No half-close. There is no CloseWrite. Protocols that use FIN as an end-of-message marker need their own framing.
  • One reliable, ordered stream per connection in protocol version 1.
  • Recovery is bounded. Signaling reconnects and ICE restarts are transparent and keep the same Conn. Anything that would require replacing the PeerConnection closes the connection with ErrDisconnected instead of silently losing, duplicating, or reordering bytes.
  • No published scale numbers yet. Capacity work is deliberately not claimed until it is measured.

Testing

go test ./...              # no network, no Docker, no root required
go test -race ./...
go test ./... -short       # skips the large-transfer cases

Documentation

Overview

package pipe exposes reliable WebRTC DataChannels through APIs that feel like standard Go networking.

The library hides SDP, ICE, DTLS, SCTP, and Pion's callback-driven API behind an Endpoint that dials and accepts connections satisfying net.Conn.

An endpoint owns one signaling subscription and may serve many concurrent inbound and outbound sessions: 

ep, err := pipe.New(ctx, pipe.Config{
	ID:       "alice",
	Signaler: signaler,
	ICEServers: []pipe.ICEServer{
		{URLs: []string{"stun:stun.example.net:3478"}},
	},
})
if err != nil {
	return err
}
defer ep.Close()

conn, err := ep.Dial(ctx, "bob")
if err != nil {
	return err
}
defer conn.Close()

Accepting connections mirrors net.Listener: 

ln, err := ep.Listen()
if err != nil {
	return err
}
for {
	conn, err := ln.Accept()
	if err != nil {
		return err
	}
	go handle(conn)
}

Signaling

Pipe does not ship a mandatory signaling service. Applications provide a Signaler that transports the versioned Signal envelope over WebSocket, SSE, NATS, Redis, MQTT, or anything else. Pipe owns negotiation semantics; the transport only moves envelopes.

The signaling/memory package provides an in-process hub for tests and same-process examples.

Limitations

  • A peer ID identifies a routing destination, not an authenticated party. Authentication belongs to the signaling trust model.
  • Protocol version 1 carries one reliable, ordered DataChannel per connection and has no half-close.
  • Recovery covers signaling reconnects and ICE restarts. If recovery would require a brand-new PeerConnection, the existing connection closes with ErrDisconnected rather than silently losing or reordering bytes.

Index

Constants

View Source 
const (
	// DefaultDialTimeout bounds a single outbound negotiation.
	DefaultDialTimeout = 30 * time.Second

	// DefaultICETimeout bounds connectivity establishment once descriptions
	// have been exchanged.
	DefaultICETimeout = 20 * time.Second

	// DefaultAcceptBacklog bounds inbound sessions that are negotiating or
	// waiting to be accepted.
	DefaultAcceptBacklog = 64

	// DefaultFramePayload is the largest application payload placed in one
	// stream frame.
	DefaultFramePayload = 16 << 10

	// DefaultReadBuffer is the largest amount of received but unread
	// application data held per connection.
	DefaultReadBuffer = 1 << 20

	// MaxFramePayload is the hard ceiling for [Config.FramePayload].
	MaxFramePayload = 256 << 10
)

Documented defaults applied by New to zero-valued configuration.

View Source 
const (
	// MaxEnvelopeSize is the largest encoded signaling envelope a transport
	// should accept, in bytes.
	MaxEnvelopeSize = 256 << 10

	// MaxSDPSize is the largest SDP body accepted in an offer, answer, or
	// restart payload, in bytes.
	MaxSDPSize = 128 << 10

	// MaxCandidateSize is the largest ICE candidate string accepted in a
	// candidate payload, in bytes.
	MaxCandidateSize = 8 << 10

	// MaxPeerIDLength is the largest accepted peer ID, in bytes.
	MaxPeerIDLength = 128

	// MaxReasonLength is the largest accepted diagnostic reason string in a
	// reject or close payload, in bytes.
	MaxReasonLength = 512
)

Protocol limits. They are hard bounds, not tunables: every field is checked against them before Pipe allocates or routes anything.

View Source 
const ProtocolVersion uint16 = 1

ProtocolVersion is the signaling envelope version implemented by this package. Peers that do not share a version reject the session.

Variables

View Source 
var (
	// ErrClosed reports use of an endpoint, listener, or connection that has
	// been closed. It is [net.ErrClosed] so that generic networking code keeps
	// working.
	ErrClosed = net.ErrClosed

	// ErrTimeout reports that an operation exceeded a deadline or a configured
	// timeout. Errors in this category also satisfy [net.Error] with
	// Timeout reporting true.
	ErrTimeout = errors.New("pipe: timeout")

	// ErrSignaling reports a failure of the signaling transport, such as a
	// closed signaling connection or a rejected send.
	ErrSignaling = errors.New("pipe: signaling failure")

	// ErrNegotiation reports a failure while establishing the session:
	// offer/answer exchange, DataChannel setup, or detach.
	ErrNegotiation = errors.New("pipe: negotiation failure")

	// ErrICE reports that connectivity establishment failed or was lost.
	ErrICE = errors.New("pipe: ICE failure")

	// ErrProtocol reports that a peer violated the signaling or stream
	// protocol, including invalid envelopes and malformed frames.
	ErrProtocol = errors.New("pipe: protocol violation")

	// ErrPeerRejected reports that the remote peer explicitly refused the
	// session.
	ErrPeerRejected = errors.New("pipe: peer rejected the session")

	// ErrPeerUnavailable reports that signaling could not reach the peer.
	ErrPeerUnavailable = errors.New("pipe: peer unavailable")

	// ErrDuplicatePeer reports that the peer ID is already registered with the
	// signaling transport.
	ErrDuplicatePeer = errors.New("pipe: duplicate peer registration")

	// ErrDisconnected reports that an established connection was lost and
	// could not be recovered within the configured policy.
	ErrDisconnected = errors.New("pipe: connection lost")

	// ErrConfig reports invalid configuration passed to [New].
	ErrConfig = errors.New("pipe: invalid configuration")

	// ErrAlreadyListening is returned by the second and later calls to
	// [Endpoint.Listen] on the same endpoint.
	ErrAlreadyListening = errors.New("pipe: endpoint is already listening")

	// ErrNotImplemented reports functionality that this build does not
	// provide yet. It never appears on a supported code path.
	ErrNotImplemented = errors.New("pipe: not implemented")
)

Sentinel error categories. Every error returned by this package matches at least one of them through errors.Is, so callers never need to compare strings.

Functions

func Dial added in v0.3.0 

func Dial(ctx context.Context, cfg Config, peer PeerID) (net.Conn, error)

Dial creates a single-use endpoint, connects to peer, and returns the connection. Closing the returned connection also closes the hidden endpoint.

Use New and Endpoint.Dial when an application makes more than one connection: one endpoint can serve many sessions over a single signaling subscription.

func Listen added in v0.3.0 

func Listen(ctx context.Context, cfg Config) (net.Listener, error)

Listen creates a single-use endpoint and returns its listener. Closing the returned listener also closes the hidden endpoint, which closes every connection that came from it.

func NewID added in v0.3.0 

func NewID() string

NewID returns a fresh random 128-bit identifier in lowercase hexadecimal, suitable for Signal.ID and Signal.SessionID. It panics only if the system random source fails, which Go treats as unrecoverable.

Types

type Addr added in v0.3.0 

type Addr struct {
	// Peer is the peer ID of the addressed side.
	Peer PeerID

	// Session is the session identifier, or the empty string for an address
	// that is not bound to a session (for example a listener address).
	Session string
}

Addr is the logical address of one side of a pipe connection. Pipe has no IP-level identity of its own: a peer is named by its PeerID and a session is named by its identifier.

func (Addr) Network added in v0.3.0 

func (a Addr) Network() string

Network returns "webrtc".

func (Addr) String added in v0.3.0 

func (a Addr) String() string

String returns the peer ID, suffixed with "#" and the session ID when the address belongs to a session.

type Backoff added in v0.3.0 

type Backoff struct {
	// Initial is the first delay.
	Initial time.Duration

	// Maximum caps the delay.
	Maximum time.Duration

	// Factor multiplies the delay after each attempt. Values below 1 are
	// rejected.
	Factor float64

	// Jitter randomizes each delay by up to this fraction, in [0, 1].
	Jitter float64
}

Backoff describes exponential backoff with jitter.

type CandidateType added in v0.3.0 

type CandidateType string

CandidateType names the kind of ICE candidate selected for a connection. 

const (
	// CandidateUnknown means no pair has been selected or Pion did not report
	// one.
	CandidateUnknown CandidateType = ""
	// CandidateHost is a local interface address.
	CandidateHost CandidateType = "host"
	// CandidateServerReflexive is an address observed through STUN.
	CandidateServerReflexive CandidateType = "srflx"
	// CandidatePeerReflexive is an address learned during connectivity checks.
	CandidatePeerReflexive CandidateType = "prflx"
	// CandidateRelay is a TURN relay address.
	CandidateRelay CandidateType = "relay"
)

Candidate types reported in ConnStats.

type CloseCode added in v0.3.0 

type CloseCode string

CloseCode is the closed set of reasons for terminating a session through signaling. 

const (
	// CloseNormal reports an ordinary close initiated by the application.
	CloseNormal CloseCode = "normal"
	// CloseGoingAway reports that the endpoint is shutting down.
	CloseGoingAway CloseCode = "going_away"
	// CloseProtocolError reports that the peer violated the protocol.
	CloseProtocolError CloseCode = "protocol_error"
	// CloseTimeout reports that an operation exceeded its budget.
	CloseTimeout CloseCode = "timeout"
	// CloseInternal reports a failure on the closing side.
	CloseInternal CloseCode = "internal"
)

Close codes defined by protocol version 1.

type Config added in v0.3.0 

type Config struct {
	// ID is the local peer ID used for signaling.
	ID PeerID

	// Signaler opens the signaling connection.
	Signaler Signaler

	// ICEServers lists STUN and TURN servers. An empty list restricts
	// connectivity to host candidates.
	ICEServers []ICEServer

	// ICETransportPolicy restricts candidate types.
	ICETransportPolicy ICETransportPolicy

	// DialTimeout bounds one outbound negotiation. It defaults to
	// [DefaultDialTimeout].
	DialTimeout time.Duration

	// ICETimeout bounds connectivity establishment. It defaults to
	// [DefaultICETimeout].
	ICETimeout time.Duration

	// KeepAlive configures ping and pong probes. Probes are disabled by
	// default.
	KeepAlive KeepAliveConfig

	// Reconnect bounds ICE-restart recovery. It defaults to enabled with a
	// small number of attempts.
	Reconnect ReconnectPolicy

	// AcceptBacklog bounds inbound sessions that are negotiating or waiting to
	// be accepted. It defaults to [DefaultAcceptBacklog].
	AcceptBacklog int

	// FramePayload is the largest application payload per stream frame. It
	// defaults to [DefaultFramePayload] and may not exceed [MaxFramePayload].
	FramePayload int

	// ReadBuffer bounds received but unread application data per connection.
	// It defaults to [DefaultReadBuffer].
	ReadBuffer int

	// Logger receives structured logs. Logging is discarded when nil; Pipe
	// never falls back to the global logger.
	Logger *slog.Logger

	// Metrics receives counters, gauges, and durations. Measurement is
	// discarded when nil.
	Metrics Metrics

	// Pion is the advanced escape hatch for tuning the WebRTC engine. The
	// defaults are correct for ordinary use.
	Pion PionOptions
	// contains filtered or unexported fields
}

Config configures an Endpoint. Only ID and Signaler are required; every other field has a documented default. Config is copied by New and is not consulted again afterwards.

type Conn 

type Conn struct {
	// contains filtered or unexported fields
}

Conn is an established pipe connection. It satisfies net.Conn over a reliable, ordered WebRTC DataChannel.

Read and Write present a byte stream even though the underlying DataChannel is message-oriented, so a Read may return fewer bytes than a matching Write sent. Multiple goroutines may use a Conn concurrently: one reader and one writer run in parallel, and concurrent writers are serialized so that the frames of a single Write stay contiguous.

Protocol version 1 has no half-close. Closing either side ends both directions.

func (*Conn) Close 

func (c *Conn) Close() error

Close implements net.Conn. It is idempotent, unblocks pending I/O, and releases the session, PeerConnection, and DataChannel that back the connection.

func (*Conn) LocalAddr added in v0.3.0 

func (c *Conn) LocalAddr() net.Addr

LocalAddr implements net.Conn. The address is logical: pipe has no IP-level identity of its own.

func (*Conn) PeerID added in v0.3.0 

func (c *Conn) PeerID() PeerID

PeerID returns the remote peer ID.

func (*Conn) Read 

func (c *Conn) Read(p []byte) (int, error)

Read implements net.Conn.

func (*Conn) RemoteAddr added in v0.3.0 

func (c *Conn) RemoteAddr() net.Addr

RemoteAddr implements net.Conn.

func (*Conn) SessionID added in v0.3.0 

func (c *Conn) SessionID() string

SessionID returns the session identifier that correlates negotiation for this connection.

func (*Conn) SetDeadline added in v0.3.0 

func (c *Conn) SetDeadline(t time.Time) error

SetDeadline implements net.Conn.

func (*Conn) SetReadDeadline added in v0.3.0 

func (c *Conn) SetReadDeadline(t time.Time) error

SetReadDeadline implements net.Conn.

func (*Conn) SetWriteDeadline added in v0.3.0 

func (c *Conn) SetWriteDeadline(t time.Time) error

SetWriteDeadline implements net.Conn.

func (*Conn) State added in v0.3.0 

func (c *Conn) State() ConnectionState

State returns the coarse connection state.

func (*Conn) Stats added in v0.3.0 

func (c *Conn) Stats() ConnStats

Stats returns a snapshot of the connection's counters. Counters may advance between fields; a snapshot is never torn.

func (*Conn) Write 

func (c *Conn) Write(p []byte) (int, error)

Write implements net.Conn. A successful call reports len(p). If some frames reached the peer before a failure, the returned count is the number of application bytes committed.

type ConnStats added in v0.3.0 

type ConnStats struct {
	// State is the connection state at snapshot time.
	State ConnectionState

	// BytesRead counts application bytes returned by Read.
	BytesRead uint64

	// BytesWritten counts application bytes accepted by Write.
	BytesWritten uint64

	// FramesRead counts decoded stream frames, including control frames.
	FramesRead uint64

	// FramesWritten counts encoded stream frames, including control frames.
	FramesWritten uint64

	// EstablishedAt is when the connection became usable.
	EstablishedAt time.Time

	// ConnectDuration is how long negotiation took.
	ConnectDuration time.Duration

	// ICERestarts counts completed ICE restarts on this connection.
	ICERestarts int

	// LocalCandidate is the local candidate type of the selected pair.
	LocalCandidate CandidateType

	// RemoteCandidate is the remote candidate type of the selected pair.
	RemoteCandidate CandidateType

	// KeepAliveRTT is the round-trip time of the most recent successful
	// keepalive probe, or zero when keepalive is disabled or has not completed
	// a probe.
	KeepAliveRTT time.Duration
}

ConnStats is a snapshot of one connection's counters. Snapshots are taken without stopping I/O, so counters may advance between fields.

type ConnectionState added in v0.3.0 

type ConnectionState int

ConnectionState is the coarse, stable lifecycle state of a connection. Pion's own transport states stay internal. 

const (
	// StateNew is the state of a session that has not begun negotiating.
	StateNew ConnectionState = iota

	// StateSignaling means the offer/answer exchange is in progress.
	StateSignaling

	// StateConnecting means descriptions are exchanged and connectivity is
	// being established.
	StateConnecting

	// StateConnected means the byte stream is usable.
	StateConnected

	// StateRecovering means connectivity was lost and bounded recovery is in
	// progress.
	StateRecovering

	// StateClosing means teardown has begun.
	StateClosing

	// StateClosed is terminal.
	StateClosed
)

Connection states. The lifecycle is: 

new -> signaling -> connecting -> connected -> recovering -> connected
                                       \                        /
                                        ---> closing -> closed <-

The transition into StateClosed happens exactly once.

func (ConnectionState) String added in v0.3.0 

func (s ConnectionState) String() string

String implements fmt.Stringer.

type Endpoint added in v0.3.0 

type Endpoint struct {
	// contains filtered or unexported fields
}

Endpoint is the primary API. It owns one signaling connection and may serve many concurrent inbound and outbound sessions.

An Endpoint is safe for concurrent use. Its lifetime is controlled by Endpoint.Close, not by the context passed to New.

func New added in v0.3.0 

func New(ctx context.Context, cfg Config) (*Endpoint, error)

New validates the configuration, opens the signaling connection, and starts the endpoint's receive loop.

ctx governs construction only. Once New returns successfully, cancelling ctx has no effect on the endpoint; call Endpoint.Close to release it.

func (*Endpoint) Addr added in v0.3.0 

func (e *Endpoint) Addr() net.Addr

Addr returns the endpoint's logical address.

func (*Endpoint) Close added in v0.3.0 

func (e *Endpoint) Close() error

Close releases the endpoint: every session, its PeerConnections, the listener, and the signaling connection. It is idempotent and safe from concurrent callers.

func (*Endpoint) Dial added in v0.3.0 

func (e *Endpoint) Dial(ctx context.Context, peer PeerID) (*Conn, error)

Dial establishes a connection to peer. It returns only after the connection is fully negotiated, detached, and ready for I/O.

Cancelling ctx removes the pending session and releases its resources. The call is also bounded by Config.DialTimeout.

func (*Endpoint) Listen added in v0.3.0 

func (e *Endpoint) Listen() (net.Listener, error)

Listen returns a listener that accepts inbound sessions. An endpoint has at most one listener; later calls return ErrAlreadyListening.

func (*Endpoint) LocalID added in v0.3.0 

func (e *Endpoint) LocalID() PeerID

LocalID returns the endpoint's peer ID.

type ICECredentialType added in v0.3.0 

type ICECredentialType int

ICECredentialType selects how ICEServer.Credential is interpreted. 

const (
	// ICECredentialPassword is a long-term credential password. It is the
	// default.
	ICECredentialPassword ICECredentialType = iota

	// ICECredentialOAuth is reserved for OAuth credentials and is not
	// supported yet.
	ICECredentialOAuth
)

Supported ICE credential types.

func (ICECredentialType) String added in v0.3.0 

func (t ICECredentialType) String() string

String implements fmt.Stringer.

type ICEServer added in v0.3.0 

type ICEServer struct {
	// URLs lists stun:, stuns:, turn:, or turns: URLs for one logical server.
	URLs []string

	// Username is the TURN username. It must be empty for STUN-only servers.
	Username string

	// Credential is the TURN credential.
	Credential string

	// CredentialType selects how Credential is interpreted.
	CredentialType ICECredentialType
}

ICEServer describes one STUN or TURN server. Pipe validates the URLs and credential combination and hands the result to Pion, which owns gathering, STUN transactions, and TURN allocations.

Credentials are never logged, exported in metrics labels, or placed in signaling messages.

type ICETransportPolicy added in v0.3.0 

type ICETransportPolicy int

ICETransportPolicy restricts which candidate types Pipe gathers. 

const (
	// ICETransportPolicyAll gathers host, server-reflexive, and relay
	// candidates. It is the default.
	ICETransportPolicyAll ICETransportPolicy = iota

	// ICETransportPolicyRelay gathers relay candidates only, which forces
	// traffic through TURN.
	ICETransportPolicyRelay
)

Supported ICE transport policies.

func (ICETransportPolicy) String added in v0.3.0 

func (p ICETransportPolicy) String() string

String implements fmt.Stringer.

type KeepAliveConfig added in v0.3.0 

type KeepAliveConfig struct {
	// Interval is the delay between probes.
	Interval time.Duration

	// Timeout bounds the wait for a matching pong. It defaults to Interval
	// when zero and must not exceed Interval.
	Timeout time.Duration
}

KeepAliveConfig configures protocol-level ping and pong probes on an established connection. Keepalive is disabled unless Interval is positive.

type Label added in v0.3.0 

type Label struct {
	Key   string
	Value string
}

Label is one bounded-cardinality metric dimension. Peer IDs and session IDs are never used as label values.

type Metrics added in v0.3.0 

type Metrics interface {
	// Count adds delta to a counter.
	Count(name string, delta int64, labels ...Label)

	// Gauge records the current value of a gauge.
	Gauge(name string, value int64, labels ...Label)

	// Duration records an observed duration.
	Duration(name string, d time.Duration, labels ...Label)
}

Metrics receives measurements without imposing an observability vendor. Implementations must be safe for concurrent use and must not block.

The metric names and labels emitted by this package are: 

pipe.dial.attempts        counter   -
pipe.dial.results         counter   result
pipe.accept.attempts      counter   -
pipe.accept.results       counter   result
pipe.sessions.active      gauge     -
pipe.sessions.pending     gauge     -
pipe.signal.sent          counter   kind, result
pipe.signal.received      counter   kind, result
pipe.connect.duration     duration  role
pipe.restart.attempts     counter   -
pipe.restart.results      counter   result
pipe.restart.duration     duration  -
pipe.stream.bytes.read    counter   -
pipe.stream.bytes.written counter   -
pipe.keepalive.rtt        duration  -
pipe.keepalive.failures   counter   reason
pipe.protocol.failures    counter   scope

type PeerID added in v0.3.0 

type PeerID string

PeerID names a signaling destination. It is a routing identity, not an authenticated one: proving that a peer owns its ID is the responsibility of the signaling transport and its authentication hooks.

type PionOptions added in v0.3.0 

type PionOptions struct {
	// ConfigureSettingEngine adjusts the setting engine once, before the shared
	// API is built. Use it for options such as network types, interface
	// filters, or a custom ICE UDP multiplexer.
	ConfigureSettingEngine func(*webrtc.SettingEngine)

	// ConfigureConfiguration adjusts the configuration applied to every
	// PeerConnection the endpoint creates. ICE servers and the transport policy
	// come from [Config] and are already applied.
	ConfigureConfiguration func(*webrtc.Configuration)
}

PionOptions is the explicitly named escape hatch for tuning the WebRTC engine. It is the only place where Pion types appear in the public API, and the defaults are correct for ordinary use.

Pipe always enables detached data channels and blocking data-channel writes before building the API, because the stream adapter owns reads, writes, and backpressure. Do not undo those settings.

type ReconnectPolicy added in v0.3.0 

type ReconnectPolicy struct {
	// Enabled turns ICE-restart recovery on.
	Enabled bool

	// MaxAttempts bounds consecutive recovery attempts.
	MaxAttempts int

	// AttemptTimeout bounds one recovery attempt.
	AttemptTimeout time.Duration

	// Backoff spaces successive attempts.
	Backoff Backoff
}

ReconnectPolicy bounds recovery of an established connection. Pipe recovers by restarting ICE on the existing PeerConnection. It never replaces the PeerConnection underneath a live connection; see ErrDisconnected.

type RejectCode added in v0.3.0 

type RejectCode string

RejectCode is the closed set of reasons for refusing a session. 

const (
	// RejectNotListening reports that the peer has no active listener.
	RejectNotListening RejectCode = "not_listening"
	// RejectBusy reports that the peer's accept backlog is full.
	RejectBusy RejectCode = "busy"
	// RejectUnauthorized reports that the peer refused the caller.
	RejectUnauthorized RejectCode = "unauthorized"
	// RejectUnsupportedVersion reports that the peers share no protocol
	// version.
	RejectUnsupportedVersion RejectCode = "unsupported_version"
	// RejectInvalidOffer reports that the offer was malformed or unusable.
	RejectInvalidOffer RejectCode = "invalid_offer"
	// RejectInternal reports a failure on the rejecting side.
	RejectInternal RejectCode = "internal"
)

Reject codes defined by protocol version 1.

type RejectedError added in v0.3.0 

type RejectedError struct {
	// Code is the machine-readable reason.
	Code RejectCode

	// Reason is a human-readable diagnostic. It never carries secrets.
	Reason string

	// Peer is the peer that refused the session.
	Peer PeerID
}

RejectedError reports that a peer refused a session. It matches ErrPeerRejected through errors.Is.

func (*RejectedError) Error added in v0.3.0 

func (e *RejectedError) Error() string

Error implements error.

func (*RejectedError) Is added in v0.3.0 

func (e *RejectedError) Is(target error) bool

Is reports that the error belongs to the ErrPeerRejected category.

type Signal added in v0.3.0 

type Signal struct {
	// Version must equal [ProtocolVersion].
	Version uint16 `json:"v"`
	// ID is a fresh random 128-bit value in lowercase hex. It exists so that
	// receivers can discard duplicates.
	ID string `json:"id"`
	// SessionID is a random 128-bit value in lowercase hex chosen by the
	// offerer. Negotiation is correlated by session, never by peer.
	SessionID string `json:"session_id"`
	// Kind identifies the payload.
	Kind SignalKind `json:"kind"`
	// From is the sending peer.
	From PeerID `json:"from"`
	// To is the receiving peer.
	To PeerID `json:"to"`
	// Payload is the kind-specific body, absent for kinds that carry none.
	Payload json.RawMessage `json:"payload,omitempty"`
}

Signal is the versioned signaling envelope exchanged by two endpoints. Signaling transports move envelopes without interpreting their payloads.

Additive fields are permitted within version 1: decoders ignore unknown object members. Any change to the meaning of an existing field requires a new kind or a new version.

func (Signal) Validate added in v0.3.0 

func (s Signal) Validate() error

Validate reports whether the envelope is structurally valid for protocol version 1. It checks the version, identifiers, kind, peer names, and the kind-specific payload against the documented limits.

Validate does not know which peer is local; endpoints additionally require that To names the local peer.

type SignalConn added in v0.3.0 

type SignalConn interface {
	// Send transmits msg. A nil error means the transport accepted the
	// message, not that the peer processed it.
	Send(ctx context.Context, msg Signal) error

	// Receive returns the next signal addressed to the local peer.
	Receive(ctx context.Context) (Signal, error)

	// Close releases the connection.
	Close() error
}

SignalConn is a bidirectional signaling connection.

Implementations may reconnect internally, but must preserve these rules:

  • Receive has exactly one caller at a time; the endpoint receive loop is that caller.
  • Send may be called concurrently with Receive.
  • Send and Receive honor the supplied context.
  • Close is idempotent and unblocks Send and Receive.

Delivery is at-least-once and ordered when practical. Pipe tolerates duplicates and reordering, so implementations must not claim exactly-once delivery.

type SignalKind added in v0.3.0 

type SignalKind string

SignalKind identifies the meaning of a Signal. Unknown kinds are rejected rather than guessed. 

const (
	// KindOffer carries an SDP offer and starts a session.
	KindOffer SignalKind = "offer"
	// KindAnswer carries the SDP answer for an offer or a restart.
	KindAnswer SignalKind = "answer"
	// KindCandidate carries one trickled ICE candidate.
	KindCandidate SignalKind = "candidate"
	// KindICEComplete reports end-of-candidates for the sender.
	KindICEComplete SignalKind = "ice-complete"
	// KindRestart carries a new offer for an ICE restart on an existing
	// session.
	KindRestart SignalKind = "restart"
	// KindReject refuses a session that has not been established.
	KindReject SignalKind = "reject"
	// KindClose terminates a session.
	KindClose SignalKind = "close"
)

Signal kinds defined by protocol version 1.

type Signaler added in v0.3.0 

type Signaler interface {
	// Open establishes a signaling connection for local. It must honor ctx
	// while connecting. The returned connection is owned by the caller.
	Open(ctx context.Context, local PeerID) (SignalConn, error)
}

Signaler opens a signaling connection for a local peer. A Signaler may be shared by several endpoints as long as each endpoint uses a distinct peer ID.

Source Files

View all Source files

Directories

Path Synopsis
examples
echo command
Command echo runs a two-peer echo service over a pipe.
 Command echo runs a two-peer echo service over a pipe.
internal/turnx
Package turnx runs a self-contained STUN and TURN server for the examples, with a configurable throughput budget on the relayed path.
 Package turnx runs a self-contained STUN and TURN server for the examples, with a configurable throughput budget on the relayed path.
turnclient command
Command turnclient moves data through a pipe connection that is forced onto a TURN relay, and reports what it measured.
 Command turnclient moves data through a pipe connection that is forced onto a TURN relay, and reports what it measured.
turnserver command
Command turnserver runs a STUN and TURN server with a configurable throughput budget, for developing and testing against pipe.
 Command turnserver runs a STUN and TURN server with a configurable throughput budget, for developing and testing against pipe.
internal
clock
Package clock provides the time source used by pipe timers so that recovery, keepalive, and timeout behavior can be tested without sleeping.
 Package clock provides the time source used by pipe timers so that recovery, keepalive, and timeout behavior can be tested without sleeping.
frame
Package frame implements the pipe stream framing defined in design/PROTOCOL.md section 4 and turns a message-oriented DataChannel into a byte stream.
 Package frame implements the pipe stream framing defined in design/PROTOCOL.md section 4 and turns a message-oriented DataChannel into a byte stream.
pionx
Package pionx contains every direct use of pion/webrtc.
 Package pionx contains every direct use of pion/webrtc.
testutil
Package testutil provides helpers shared by pipe's tests: goroutine leak detection and a fake detached DataChannel.
 Package testutil provides helpers shared by pipe's tests: goroutine leak detection and a fake detached DataChannel.
signaling
memory
Package memory provides an in-process signaling hub for tests and same-process examples.
 Package memory provides an in-process signaling hub for tests and same-process examples.
signalertest
Package signalertest is a black-box conformance suite for pipe.Signaler implementations.
 Package signalertest is a black-box conformance suite for pipe.Signaler implementations.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.