memory

Documentation

Overview

Package memory provides an in-process signaling hub for tests and same-process examples. It routes envelopes between peers registered on the same Hub without sockets, and it can inject faults deterministically so that duplicate, dropped, reordered, and disconnected signaling can be tested without timing luck.

Index

• Constants
• type Fault
  • func DropFirst(kind pipe.SignalKind, n int) Fault
  • func DropKind(kinds ...pipe.SignalKind) Fault
  • func DuplicateAll() Fault
  • func SwapAdjacent(kind pipe.SignalKind) Fault
• type FaultFunc
  • func (f FaultFunc) Apply(sig pipe.Signal) []pipe.Signal
• type Hub
  • func New(opts ...Option) *Hub
  • func (h *Hub) Disconnect(id pipe.PeerID) bool
  • func (h *Hub) Open(ctx context.Context, local pipe.PeerID) (pipe.SignalConn, error)
  • func (h *Hub) Registered() []pipe.PeerID
• type Option
  • func WithFault(f Fault) Option
  • func WithQueueSize(n int) Option

Constants

const DefaultQueueSize = 64

DefaultQueueSize is the per-peer inbound queue depth.

Types

type Fault

type Fault interface {
	Apply(sig pipe.Signal) []pipe.Signal
}

Fault transforms one outbound signal into the signals actually delivered. Returning an empty slice drops the signal; returning several copies duplicates it. Implementations must be safe for concurrent use.

func DropFirst

func DropFirst(kind pipe.SignalKind, n int) Fault

DropFirst drops the first n signals of the given kind and delivers the rest.

func DropKind

func DropKind(kinds ...pipe.SignalKind) Fault

DropKind drops every signal of the given kinds.

func DuplicateAll

func DuplicateAll() Fault

DuplicateAll delivers every signal twice, which exercises deduplication.

func SwapAdjacent

func SwapAdjacent(kind pipe.SignalKind) Fault

SwapAdjacent holds back every other signal of the given kind and releases it after the next one, which reorders delivery deterministically. An odd final signal stays held, so this fault also drops at most one signal per kind.

type FaultFunc

type FaultFunc func(sig pipe.Signal) []pipe.Signal

FaultFunc adapts a function to Fault.

func (FaultFunc) Apply

func (f FaultFunc) Apply(sig pipe.Signal) []pipe.Signal

Apply implements Fault.

type Hub

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

Hub is an in-process signaling relay. It implements pipe.Signaler and may be shared by any number of endpoints with distinct peer IDs.

func New

func New(opts ...Option) *Hub

New returns an empty hub.

func (*Hub) Disconnect

func (h *Hub) Disconnect(id pipe.PeerID) bool

Disconnect closes the signaling connection of one peer as if its transport had failed. It reports whether the peer was registered.

func (*Hub) Open

func (h *Hub) Open(ctx context.Context, local pipe.PeerID) (pipe.SignalConn, error)

Open registers local and returns its signaling connection. Registering a peer ID that is already live fails with pipe.ErrDuplicatePeer.

func (*Hub) Registered

func (h *Hub) Registered() []pipe.PeerID

Registered returns the peer IDs with a live connection.

type Option

type Option func(*Hub)

Option configures a Hub.

func WithFault

func WithFault(f Fault) Option

WithFault installs a fault injector applied to every delivered signal.

func WithQueueSize

func WithQueueSize(n int) Option

WithQueueSize sets the per-peer inbound queue depth. A send to a full queue blocks until the peer reads or the context ends.