Help us shape the future of Go! Take the 2026 Go Developer Survey

link

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Oct 10, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

README

github.com/nfltr/link is the nfltr endpoint-to-endpoint link layer: many sessions, each a byte stream with its own flow control, multiplexed over one connection the caller dials and secures (TLS, a proxy, a pipe). A session whose reader is slow stalls only itself; one write to the network carries every frame queued at the time.

  • Standard library only (TestModuleBoundary).
  • Apache-2.0 (LICENSE, NOTICE; every source file carries an SPDX header).
  • Wire formats version 1 and 2 are specified in doc.go; testdata/v1_session.golden and testdata/v2_session.golden hold recorded sessions every release is tested against, byte for byte.

Packages

package what
link Client/Server (the hello of either version, an Authenticator's credential, version negotiation, Config, Observer), Conn (frames, write coalescing; version 2: OPEN/CLOSE, ping liveness, GOAWAY), Credit (per-session flow control, also over a caller's own transport), Mux and Session (sessions as net.Conns), Pool (several connections per peer, growth, backoff, make-before-break drain), Liveness and its presets; Hello/ReadHello/Reply (the version-1 handshake by hand)
link/bytequeue a FIFO of byte slices bounded by total bytes; PushWithin for a pusher that serves other queues too

Use

c, err := link.Client(ctx, nc, link.Config{Window: 4 << 20, Liveness: link.LivenessFast, Auth: auth}) // dialer
m := link.NewMux(c, nil)
go m.Serve(nil)
s, err := m.Open(func(string) ([]byte, error) { return header, nil })

c, id, err := link.Server(ctx, nc, link.Config{Auth: auth}) // listener
go link.NewMux(c, nil).Serve(func(header []byte, s *link.Session) { /* ... */ })

p := link.NewPool(link.PoolConfig{Link: cfg, Dial: dial, Max: 8, Backoff: backoff})
s, err := p.Open(ctx, header)

Checks

make check: build, vet, tests with -race, fuzz-short (every decoder, FUZZTIME each, default 10s), gosec, semgrep (docker), govulncheck.

Status

v0.1.0: the version-1 wire, extracted unchanged from nfltr's TCP data plane. v0.2.0: wire version 2 (negotiated in the hello; version 1 served byte for byte), Client/Server with an Authenticator, ping liveness driven by the Liveness presets, GOAWAY, Pool, Observer, Credit for callers with a transport of their own. Planned: session deadlines, the replica link.

Documentation

Overview

Package link is the nfltr endpoint-to-endpoint link layer: many sessions, each a byte stream with its own flow control, multiplexed over one connection the caller dials and secures (TLS, a proxy, a pipe). A session whose reader is slow stalls only itself; one write to the network carries every frame queued at the time.

It imports only the Go standard library.

Client and Server bind a connection (the hello, an Authenticator's credential, version negotiation); Mux carries sessions over it as net.Conns; Pool keeps several connections to one peer. Credit is the per-session flow control, also for a caller that carries its sessions on a transport of its own.

Wire format, version 1 

hello  dialer -> listener: "NFDP" version(1) token(32 bytes) window(4)
reply  listener -> dialer: status(1), 0 = bound; window(4) when bound
frame  both ways:          length(4, of the body) type(1) body
  type 0, data:    sid_len(1) sid payload
  type 1, message: opaque bytes of the caller's (never empty)
  type 2, window:  sid_len(1) sid increment(4)

Integers are big endian. A session id (sid) is at most 255 bytes, chosen by the side that opens the session. A window is how many bytes of one session the side that announces it buffers (see Credit.Acquire). In version 1 the frames that open and close a session are message frames in the caller's encoding (Codec); session ids that begin with '_' name control frames, which Mux leaves to the caller.

testdata/v1_session.golden holds a recorded version-1 session; the tests hold every release to those bytes.

Wire format, version 2

Same magic, the version byte 2; a listener serves both versions. 

hello  "NFDP" version(1)=2 max_version(1) features(4) window(4)
       max_sessions(4) interval_ms(4) timeout_ms(4)
       auth_method(1) auth_len(2) auth(auth_len, at most 4096)
reply  status(1):
         0 bound:   version(1)=2 features(4) window(4) max_sessions(4)
         1 refused (the credential)
         2 version: max_version(1), the highest the listener speaks
frame  as version 1, plus:
  type 3, open:    sid_len(1) sid header (the opener's bytes, at most 64 KiB)
  type 4, close:   sid_len(1) sid code(1): 0 the sender neither sends
                   nor takes more; 1 it sends no more (half-close)
  type 5, ping:    8 opaque bytes
  type 6, pong:    the ping's 8 bytes
  type 7, goaway:  drain_ms(4, 0 = none) reason (UTF-8, at most 1024)

The dialer says which version it speaks (out of band, the listener advertised which it speaks); a listener answers a version it does not speak with status 2 and the dialer redials. A version-1 listener reads 37 bytes of any hello before it refuses one of version 2.

max_sessions is how many sessions the peer may have open toward the side that announces it at once; an open past it ends the connection. Feature bits are reserved: a side sends 0 and ignores the bits it does not know. Liveness is the dialer's (interval 1s to 15s, timeout at least one interval); a listener takes it unless it has its own: each side pings once it received nothing for the interval and ends the connection when nothing arrived for the timeout after that; every ping is answered. An open starts the session's credit on both sides with the peer's window; a close of code 0 ends it. A Mux picks session ids as uvarints, odd on the dialer and even on the listener. After a goaway the receiver opens no more sessions on the connection; opens that cross it are refused with a close.

testdata/v2_session.golden holds a recorded version-2 session.

Index

Constants

View Source 
const (
	MethodNone  byte = 0 // no credential (NoAuth)
	MethodToken byte = 1 // a bearer token the listener issued; version 1 carries only this, TokenLen bytes
)

Auth methods with a meaning in this package; the others are the Authenticator's to define (128 and up are never assigned here).

View Source 
const (
	StatusBound   = 0
	StatusRefused = 1
)

Status bytes of the listener's reply to a hello.

View Source 
const (
	MinWindow = 2 * MaxChunk
	MaxWindow = 1 << 30
)

Window bounds: a side announces a window in [MinWindow, MaxWindow]. Credit returns in steps of half a window, which must hold a MaxChunk.

View Source 
const (
	MinLivenessInterval = time.Second
	MaxLivenessInterval = 15 * time.Second
)

Bounds Validate enforces. A heartbeat sender keeps an interval in [MinLivenessInterval, MaxLivenessInterval]: below it the frames cost more than they tell, above it a 45 s limit for a silent peer (one that answers each frame) would drop the connection. Timeout is at least one more interval, since a peer skips a frame right after sending data.

View Source 
const DefaultWindow = 1 << 20

DefaultWindow is a Config's window when it names none.

View Source 
const Magic = "NFDP"

Magic opens a link connection, where an HTTP/2 connection starts "PRI ": a listener that shares its port can tell the two apart.

View Source 
const MaxAuth = 4096

MaxAuth bounds a hello's auth payload.

View Source 
const MaxChunk = 64 << 10

MaxChunk is the most data a sender acquires credit for at once: larger writes are split.

View Source 
const MaxSID = 255

MaxSID is the longest session id.

View Source 
const MaxSessions = 1024

MaxSessions is the most sessions one Mux carries at once by default: Open refuses more (ErrFull), Serve closes a session the peer opens past it. On version 2 the peer's hello sets the limit toward it.

View Source 
const MaxVersion = 2

MaxVersion is the highest wire version this package speaks.

View Source 
const StatusVersion = 2

StatusVersion is a version-2 listener's reply to a hello whose version it does not speak; its highest version follows.

View Source 
const TokenLen = 32

TokenLen is the length of a hello's token: the hex of 16 random bytes.

View Source 
const Version = 1

Version is the hello's protocol version.

Variables

View Source 
var (
	// ErrDeadPeer ends a connection on which nothing arrived for its
	// liveness's interval plus timeout, a ping included.
	ErrDeadPeer = errors.New("link: peer silent past the liveness timeout")
	// ErrGoAway is Open's refusal on a connection that is going away
	// (either side sent GOAWAY).
	ErrGoAway = errors.New("link: connection going away")
	// ErrAuth is a listener's refusal of a dialer its Authenticator did
	// not verify.
	ErrAuth = errors.New("link: peer not authenticated")
	// ErrVersion is a peer that does not speak the version asked for
	// (VersionError says which it speaks).
	ErrVersion = errors.New("link: wire version not supported")
)
View Source 
var (
	// LivenessDefault: dead within 15 s, a frame each way every 5 s.
	LivenessDefault = Liveness{Interval: 5 * time.Second, Timeout: 10 * time.Second}
	// LivenessFast: quick failover and interactive use; dead within 4.5 s,
	// a frame each way every 1.5 s.
	LivenessFast = Liveness{Interval: 1500 * time.Millisecond, Timeout: 3 * time.Second}
	// LivenessNAT: long idle behind NATs, conntrack and load balancers that
	// drop a mapping idle for 30 s or more; dead within 30 s, a frame each
	// way every 10 s.
	LivenessNAT = Liveness{Interval: 10 * time.Second, Timeout: 20 * time.Second}
	// LivenessDatacenter: stable networks; dead within 45 s, a frame each
	// way every 15 s.
	LivenessDatacenter = Liveness{Interval: 15 * time.Second, Timeout: 30 * time.Second}
	// LivenessMobile: laptops and phones on flaky Wi-Fi or metered links:
	// the fewest wakeups, and a radio that sleeps or hands over is not
	// taken for a dead peer; dead within 60 s, a frame each way every 15 s.
	LivenessMobile = Liveness{Interval: 15 * time.Second, Timeout: 45 * time.Second}
)

The presets. Idle traffic is one small frame each way per Interval per connection; the bound is Interval + Timeout from the peer's last frame.

View Source 
var ErrCredit = errors.New("link: peer exceeded the session window")

ErrCredit is the receive error of a peer that broke the credit rules: it sent a session more data than this side's window, or granted more credit than its own window.

View Source 
var ErrFrame = errors.New("link: malformed frame")

ErrFrame reports bytes that break the frame format.

View Source 
var ErrFull = errors.New("link: connection carries its most sessions")

ErrFull is Open's refusal of a session past the most sessions allowed.

View Source 
var ErrRefused = errors.New("link: peer refused the connection")

ErrRefused is the listener's refusal of a hello: an unknown token, or one that binds no more connections.

View Source 
var LivenessPresets = []string{"default", "fast", "nat", "datacenter", "mobile"}

LivenessPresets names the presets, in the order help and docs list them.

Functions

func AppendWindow added in v0.2.0 

func AppendWindow(b []byte, sid string, n int) []byte

AppendWindow appends a window update's body, sid_len(1) sid increment(4), to b: the body of a window frame, and what a caller with a transport of its own carries for a Credit's grant. sid is at most MaxSID bytes and n at most a window.

func CheckWindow 

func CheckWindow(w uint32) (int, error)

CheckWindow returns w as a window, or an error when it is out of range.

func Hello 

func Hello(nc net.Conn, token string, window int) (int, error)

Hello binds a new connection with token, announcing window, and returns the listener's window from its reply.

func IsHello added in v0.2.0 

func IsHello(prefix []byte) bool

IsHello reports whether a connection's first bytes are a link hello's: a listener that shares its port (with TLS or HTTP/2) tells them apart by its first len(Magic) bytes.

func ParseWindow added in v0.2.0 

func ParseWindow(b []byte) (sid string, n int, err error)

ParseWindow reads a window update's body (AppendWindow).

func ReadHello 

func ReadHello(r io.Reader) (string, int, error)

ReadHello reads the rest of a hello, after its Magic, and returns the token and the dialer's window.

func Reply 

func Reply(w io.Writer, status byte, window int) error

Reply answers a hello; the reply of a bound connection announces window.

func Server added in v0.2.0 

func Server(ctx context.Context, nc net.Conn, cfg Config) (*Conn, Identity, error)

Server binds nc, whose security (TLS) is the caller's, as the listener: it reads the dialer's hello (Magic first: a caller that sniffed it replays it), verifies its credential with cfg.Auth, replies, and returns the connection and who the dialer is. It speaks every version up to cfg.Version: a version-1 hello is a token (MethodToken) and gets the version-1 reply, byte for byte. A refused dialer gets the refusal and Server ErrAuth. On error the caller closes nc.

Types

type Authenticator added in v0.2.0 

type Authenticator interface {
	// Hello returns the dialer's auth method and payload (at most MaxAuth
	// bytes).
	Hello(ctx context.Context, cs tls.ConnectionState) (method byte, payload []byte, err error)
	// Verify checks a dialer's method and payload and returns who it is.
	// An error refuses the connection.
	Verify(ctx context.Context, cs tls.ConnectionState, method byte, payload []byte) (Identity, error)
}

Authenticator proves a dialer to a listener. The connection's TLS state (zero on a plaintext one) is there for channel binding. 

var NoAuth Authenticator = noAuth{}

NoAuth sends no credential and accepts only peers that send none: for connections whose security is elsewhere (an mTLS peer port, a pipe).

type Codec 

type Codec interface {
	// Close returns the message that closes session sid.
	Close(sid string) ([]byte, error)
	// Parse reads a received message: the session it names, whether it
	// opens or closes it, and data it carries for the session.
	Parse(msg []byte) (Message, error)
}

Codec reads and writes the messages that open and close sessions: in wire version 1 their encoding is the caller's. Version 2 needs none.

type Config added in v0.2.0 

type Config struct {
	// Version is the highest wire version to speak: 0 is MaxVersion. A
	// dialer speaks it; give it what the listener advertised (out of
	// band): a version-1 listener refuses a version-2 hello only after it
	// read 37 bytes of it.
	Version int
	// Window is how many bytes of one session this side buffers: 0 is
	// DefaultWindow.
	Window int
	// MaxSessions is how many sessions the peer may have open toward this
	// side at once (version 2): 0 is MaxSessions.
	MaxSessions int
	// Liveness (version 2): this side declares the connection dead when it
	// received nothing for Interval, sent a ping, and received nothing for
	// Timeout more (ErrDeadPeer). A dialer's zero is LivenessDefault; a
	// listener's zero is the dialer's, so both ends detect a dead path
	// within the same bound.
	Liveness Liveness
	// Auth builds the dialer's credential (Hello) and verifies it on the
	// listener (Verify). A dialer's nil is NoAuth; a listener needs one.
	Auth Authenticator
	// Observer receives the connection's events; nil is none.
	Observer Observer
}

Config is one side's settings for a connection (Client, Server).

type Conn 

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

Conn sends and receives frames on a bound connection. One goroutine Receives; the sends, Flush and the credit calls (its Credit's) are safe from any.

Reads go straight to the connection: over TLS, tls.Conn's record buffer serves the small header reads, and an idle connection pins no buffer. ponytail: a plaintext connection (loopback, dev) reads the socket for each header; give it a bufio.Reader if one carries real traffic.

func Client added in v0.2.0 

func Client(ctx context.Context, nc net.Conn, cfg Config) (*Conn, error)

Client binds nc, whose security (TLS) is the caller's, as the dialer: it sends the hello of cfg.Version with cfg.Auth's credential and reads the listener's reply. A listener that does not speak the version answers with a VersionError (errors.Is ErrVersion) naming the versions it speaks, or, at version 1, by closing the connection; redial with one it speaks. A refusal is ErrRefused. On error the caller closes nc.

func NewConn 

func NewConn(nc net.Conn, window, peerWindow int) *Conn

NewConn returns a version-1 Conn over nc, whose hello is done: window is this side's window, peerWindow the other side's.

func (*Conn) Close 

func (c *Conn) Close() error

Close closes the connection; senders waiting for credit give up.

func (*Conn) Done added in v0.2.0 

func (c *Conn) Done() <-chan struct{}

Done is closed once the connection is closed. allow:wrapper

func (*Conn) Err added in v0.2.0 

func (c *Conn) Err() error

Err is why the connection ended when the link ended it (ErrDeadPeer), else nil.

func (*Conn) Flush 

func (c *Conn) Flush() error

Flush writes the buffered frames: inline when no write is in flight, else the write in flight takes them next. The buffer goes back to the pool once written, so an idle connection holds none.

func (*Conn) GoAway added in v0.2.0 

func (c *Conn) GoAway(drain time.Duration, reason string) error

GoAway tells the peer (version 2) to open no more sessions on this connection: the open ones may run for drain, then this side closes it. It also stops this side opening more (ErrGoAway). It writes at once.

func (*Conn) GoingAway added in v0.2.0 

func (c *Conn) GoingAway() bool

GoingAway reports whether either side sent GOAWAY. allow:wrapper

func (*Conn) Identity added in v0.2.0 

func (c *Conn) Identity() Identity

Identity is who the peer is, as the listener's Authenticator said (zero on the dialer). allow:wrapper

func (*Conn) PeerGoAway added in v0.2.0 

func (c *Conn) PeerGoAway() <-chan struct{}

PeerGoAway is closed once the peer sent GOAWAY (version 2; never on version 1).

func (*Conn) PeerMaxSessions added in v0.2.0 

func (c *Conn) PeerMaxSessions() int

PeerMaxSessions is how many sessions this side may have open toward the peer at once (version 2). allow:wrapper

func (*Conn) Recv 

func (c *Conn) Recv() (Frame, error)

Recv returns the next data, message, open or close frame; it applies the others itself and charges data frames against their session's window. A peer that breaks the credit rules gets ErrCredit (credit.go); the caller closes the connection on any error. A connection the link ended returns why (ErrDeadPeer).

func (*Conn) SendClose added in v0.2.0 

func (c *Conn) SendClose(sid string, half bool) error

SendClose buffers a CLOSE frame (version 2) for sid: half, the side sends no more of the session and still takes the peer's; else it neither sends nor takes more, and the session's credit ends.

func (*Conn) SendData 

func (c *Conn) SendData(sid string, data []byte) error

SendData buffers sid's data as data frames, as SendMsg does. The caller acquired credit for it (Acquire).

func (*Conn) SendMsg 

func (c *Conn) SendMsg(b []byte) error

SendMsg buffers a message frame; Flush, or a full buffer, writes it. The caller does not modify b afterwards.

func (*Conn) SendOpen added in v0.2.0 

func (c *Conn) SendOpen(sid string, header []byte) error

SendOpen buffers an OPEN frame (version 2) for session sid, with header (the opener's bytes, at most 64 KiB), and starts the session's credit. It refuses a session past the peer's max sessions (ErrFull) and any on a connection that is going away (ErrGoAway).

func (*Conn) Version added in v0.2.0 

func (c *Conn) Version() int

Version is the connection's wire version. allow:wrapper

type Credit added in v0.2.0 

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

Credit is the per-session flow control of one connection. A Conn has one and carries its window updates in window frames; a caller whose sessions travel on a transport of its own (a gRPC stream) makes one with NewCredit and carries the updates itself (AppendWindow, ParseWindow). Its methods are safe from any goroutine.

func NewCredit added in v0.2.0 

func NewCredit(window, peerWindow int, grant func(sid string, n int)) *Credit

NewCredit returns the credit of a connection whose side buffers window bytes per session and whose peer buffers peerWindow. grant sends the peer a window update for n bytes of sid (AppendWindow); it is called from Consumed, never with a lock held.

func (*Credit) Acquire added in v0.2.0 

func (c *Credit) Acquire(sid string, n int) bool

Acquire waits until sid may send n more bytes and takes them. n must not exceed the peer's window (MaxChunk does not). It reports false once the session is forgotten or the credit closed.

Before it waits it flushes: the peer returns credit only for bytes it received, so a sender waiting with its own frames still buffered (a Session.Write of several chunks) would wait for good.

func (*Credit) Close added in v0.2.0 

func (c *Credit) Close()

Close makes every sender waiting in Acquire, and every later Acquire, give up.

func (*Credit) Consumed added in v0.2.0 

func (c *Credit) Consumed(sid string, n int)

Consumed acknowledges n bytes of sid that left the receive buffer; a window update goes out once half a window has.

func (*Credit) Forget added in v0.2.0 

func (c *Credit) Forget(sid string)

Forget ends a session's credit: a sender waiting in Acquire gives up.

func (*Credit) Grant added in v0.2.0 

func (c *Credit) Grant(sid string, n int) error

Grant adds a window update's bytes to sid's credit, which never exceeds the peer's window.

func (*Credit) Open added in v0.2.0 

func (c *Credit) Open(sid string)

Open starts the credit of a session on both sides: the opener before it sends the open frame, the other side when it receives it. Opening an id that is open starts it again: its sender waiting in Acquire gives up.

func (*Credit) PeerWindow added in v0.2.0 

func (c *Credit) PeerWindow() int

PeerWindow is the other side's window. allow:wrapper

func (*Credit) Received added in v0.2.0 

func (c *Credit) Received(sid string, n int) error

Received charges n bytes of data for sid against this side's window: a Conn does for data frames, a caller for data it carries otherwise. Data of a session this side does not know is not buffered by anyone and is not charged.

func (*Credit) Window added in v0.2.0 

func (c *Credit) Window() int

Window is this side's window. allow:wrapper

type Frame 

type Frame struct {
	SID    string
	Data   []byte
	Msg    []byte
	Open   bool
	Header []byte
	Close  bool
	Half   bool
}

Frame is a received frame: a session's data (SID, Data), a message (Msg, the caller's bytes), or in version 2 a session's open (SID, Open, Header: the opener's bytes) or close (SID, Close; Half when the peer only stopped sending). Window, ping, pong and goaway frames never surface: Recv applies them.

type Identity added in v0.2.0 

type Identity struct{ Method, Subject string }

Identity is who the peer is, as the listener's Authenticator says.

type Liveness 

type Liveness struct {
	Interval time.Duration
	Timeout  time.Duration
}

Liveness is how a connection notices its peer is gone without a FIN (it lost power, a NAT dropped the path): the peer sends a frame at least every Interval while idle, and the connection is dead once Interval + Timeout pass without any frame. Where one side sends the frames (its heartbeat), the other asks for Interval, the sender states the interval it keeps, and the asker waits DeadAfter(stated).

func ParseLiveness 

func ParseLiveness(spec string) (Liveness, error)

ParseLiveness reads a preset name, raw values ("interval=2s,timeout=6s"), or a preset with raw values over it ("mobile,timeout=90s"); "" is LivenessDefault. The result is validated.

func (Liveness) DeadAfter 

func (l Liveness) DeadAfter(stated time.Duration) time.Duration

DeadAfter is how long to wait for any frame once the peer stated the interval it keeps: stated + Timeout, and never under three stated intervals (a peer that keeps another interval than the one asked for, or skips a frame after data). 0 when the peer stated none: liveness is then the transport keepalive's.

func (Liveness) String 

func (l Liveness) String() string

String is the spec ParseLiveness reads back.

func (Liveness) Validate 

func (l Liveness) Validate() error

Validate refuses values a heartbeat sender would not keep or that would take a live connection for a dead one.

type Message 

type Message struct {
	SID         string
	Open, Close bool
	Data        []byte
}

Message is a parsed message frame (Codec.Parse).

type Mux 

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

Mux carries sessions over a Conn, each a net.Conn. Either side opens sessions (Open), the other takes them (Serve's accept). A session's data goes on credit, so a session whose reader is slow stalls only itself; a Write goes to the network inline, or with the frames of other sessions when a write is in flight (Conn).

On version 1 opens and closes are the caller's messages (Codec), and the Mux picks session ids from 1 up. On version 2 they are OPEN and CLOSE frames, the open carrying the opener's header, and the Mux picks ids as uvarints, odd on the dialer and even on the listener, so the two sides never pick the same.

func NewMux 

func NewMux(c *Conn, codec Codec) *Mux

NewMux returns a Mux over c whose open and close messages codec reads and writes (version 1; nil on version 2).

func (*Mux) Close 

func (m *Mux) Close() error

Close closes the connection: Serve returns and every session ends. allow:wrapper

func (*Mux) Conn added in v0.2.0 

func (m *Mux) Conn() *Conn

Conn is the Mux's connection. allow:wrapper

func (*Mux) GoAway added in v0.2.0 

func (m *Mux) GoAway(deadline time.Time) error

GoAway (version 2) tells the peer to open no more sessions here and stops this side opening more; the connection closes once its last session ends, and at deadline at the latest (zero: no deadline). OPENs that arrive afterwards are refused: their opener retries elsewhere.

func (*Mux) Open 

func (m *Mux) Open(open func(sid string) ([]byte, error)) (*Session, error)

Open starts a session: open returns the bytes that open it for the session id the Mux chose (version 1: the caller's open message; version 2: the OPEN frame's header). Open sends it at once, so a peer that speaks first is reached without the caller's bytes.

func (*Mux) Serve 

func (m *Mux) Serve(accept func(open []byte, s *Session)) error

Serve receives frames until the connection fails, then closes it and ends every session. accept, nil on a side that takes no sessions, runs in a goroutine of its own for each session the peer opens, with the bytes that opened it (version 1: the open message; version 2: the header); it owns the session and closes it.

func (*Mux) Sessions added in v0.2.0 

func (m *Mux) Sessions() int

Sessions is how many sessions are open.

type NopObserver added in v0.2.0 

type NopObserver struct{}

NopObserver ignores every event.

func (NopObserver) ConnDown added in v0.2.0 

func (NopObserver) ConnDown(Identity, error)

func (NopObserver) ConnUp added in v0.2.0 

func (NopObserver) ConnUp(Identity)

func (NopObserver) CreditWait added in v0.2.0 

func (NopObserver) CreditWait(time.Duration)

func (NopObserver) GoAway added in v0.2.0 

func (NopObserver) GoAway(Identity)

func (NopObserver) PingRTT added in v0.2.0 

func (NopObserver) PingRTT(time.Duration)

func (NopObserver) SessionClose added in v0.2.0 

func (NopObserver) SessionClose(error)

func (NopObserver) SessionOpen added in v0.2.0 

func (NopObserver) SessionOpen()

type Observer added in v0.2.0 

type Observer interface {
	ConnUp(Identity)
	ConnDown(Identity, error)
	SessionOpen()             // a session opened by an OPEN frame (version 2)
	SessionClose(error)       // one of those ended
	CreditWait(time.Duration) // a sender waited this long for credit
	PingRTT(time.Duration)    // a ping's round trip (version 2)
	GoAway(Identity)          // the peer sent GOAWAY (version 2)
}

Observer receives a connection's events, to map onto the caller's metrics. Calls come from the connection's goroutines and must not block. Embed NopObserver to take only some.

type OwnedWriter 

type OwnedWriter interface {
	WriteOwned(b []byte) error
}

OwnedWriter is a writer that may keep the slices it is given: WriteTo hands it the received slices themselves, no copy.

type Pool added in v0.2.0 

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

Pool keeps connections to one peer and opens sessions on them: on the one with the fewest sessions, dialling more as Grow says. A connection the peer sends GOAWAY on takes no more sessions; its open ones run on. Drain replaces every connection make-before-break. Sessions the peer opens toward a Pool are refused.

func NewPool added in v0.2.0 

func NewPool(cfg PoolConfig) *Pool

NewPool returns a Pool; it dials its Min connections in the background.

func (*Pool) Close added in v0.2.0 

func (p *Pool) Close() error

Close closes every connection; Open fails from then on.

func (*Pool) Drain added in v0.2.0 

func (p *Pool) Drain(ctx context.Context) error

Drain replaces the Pool's connections make-before-break: it dials a replacement (Min of them, at least one) first, then sends GOAWAY on every old connection, which then takes no new session and closes once its open sessions ended, at ctx's deadline at the latest. It returns once they all closed. A replacement that cannot be dialled leaves the old connections as they were.

func (*Pool) Open added in v0.2.0 

func (p *Pool) Open(ctx context.Context, header []byte) (*Session, error)

Open opens a session with header on one of the Pool's connections.

func (*Pool) Stats added in v0.2.0 

func (p *Pool) Stats() PoolStats

Stats is the Pool's load.

type PoolConfig added in v0.2.0 

type PoolConfig struct {
	// Link is each connection's Config; a Pool speaks version 2 (Version
	// 0 or 2).
	Link Config
	// Dial opens a connection to the peer, its security done (TLS, a
	// proxy): the caller's.
	Dial func(ctx context.Context) (net.Conn, error)
	// Min connections are kept open, redialled with Backoff when one ends;
	// Max (0 is 1) is the most open at once.
	Min, Max int
	// Grow, when set, says whether Open dials one more connection (below
	// Max) although one could take the session; nil grows only when none
	// can.
	Grow func(PoolStats) bool
	// Backoff is how long to wait before redialling after attempt failed
	// dials (from 1). nil: Open fails with the dial's error, and Min is
	// redialled once a second.
	Backoff func(attempt int) time.Duration
}

PoolConfig is a Pool's settings. Every policy is the caller's.

type PoolStats added in v0.2.0 

type PoolStats struct{ Conns, Sessions int }

PoolStats is a Pool's load: open connections that take sessions, and their sessions.

type Session 

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

Session is one session of a Mux. Read and WriteTo take the peer's bytes in order, then io.EOF once the peer closed the session (or its write side); Write sends on credit. Deadlines are not supported: Close ends a session.

func (*Session) Close 

func (s *Session) Close() error

Close ends the session: the peer gets a close unless it closed it; a pending Read returns.

func (*Session) CloseWrite added in v0.2.0 

func (s *Session) CloseWrite() error

CloseWrite (version 2) ends the session's write side: the peer reads io.EOF after what was written; this side still reads the peer's bytes. The session ends once both sides closed their write side.

func (*Session) Header added in v0.2.0 

func (s *Session) Header() []byte

Header is the bytes that opened the session: the open message on version 1, the OPEN frame's header on version 2. allow:wrapper

func (*Session) LocalAddr 

func (s *Session) LocalAddr() net.Addr

LocalAddr is the connection's. allow:wrapper

func (*Session) Read 

func (s *Session) Read(b []byte) (int, error)

func (*Session) RemoteAddr 

func (s *Session) RemoteAddr() net.Addr

RemoteAddr is the connection's. allow:wrapper

func (*Session) SetDeadline 

func (s *Session) SetDeadline(time.Time) error

SetDeadline and its siblings are no-ops: Close ends a session.

func (*Session) SetReadDeadline 

func (s *Session) SetReadDeadline(time.Time) error

func (*Session) SetWriteDeadline 

func (s *Session) SetWriteDeadline(time.Time) error

func (*Session) Write 

func (s *Session) Write(b []byte) (int, error)

Write sends b to the peer in chunks it has credit for, waiting for credit, then writes them to the network (Flush).

func (*Session) WriteTo 

func (s *Session) WriteTo(w io.Writer) (int64, error)

WriteTo writes the peer's bytes to w until the session ends; a backlog goes in one net.Buffers write (writev on a TCP conn), or slice by slice to an OwnedWriter.

type VersionError added in v0.2.0 

type VersionError struct{ Max int }

VersionError is a listener's refusal of a hello's version: it speaks versions up to Max.

func (*VersionError) Error added in v0.2.0 

func (e *VersionError) Error() string

func (*VersionError) Is added in v0.2.0 

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

Is makes a VersionError an ErrVersion. allow:wrapper

Source Files

View all Source files

Directories

Path Synopsis
Package bytequeue is a FIFO of byte slices bounded by total bytes.
 Package bytequeue is a FIFO of byte slices bounded by total bytes.
linkvet module
peerlink
grpclink module
quiclink module

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.