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

link

package module
v0.6.0 Latest Latest
Warning

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

 Go to latest Published: Oct 12, 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 format version 2 is specified in doc.go; testdata/v2_session.golden holds a recorded session every release is tested against, byte for byte. Version 1 was removed in v0.5.0.

Packages

package what
link Client/Server (the hello, an Authenticator's credential, Config, Observer), Conn (frames, write coalescing, 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
link/bytequeue a FIFO of byte slices bounded by total bytes; PushWithin for a pusher that serves other queues too
link/peerlink a transport-neutral API for sessions between peers (Link: Open, Serve, Drain), its contract, and Node, an implementation over Pool and Mux (peerlink/README.md)
link/peerlink/peerlinktest the contract's conformance suite (Run) and the fault-injecting Proxy, for any implementation

peerlink/grpclink (a module of its own, make check-grpclink) implements peerlink.Link over gRPC streams and passes the same suite.

Use

example_test.go (Example, run by go test, shown by go doc) is a complete program: a TLS listener that authenticates dialers and serves sessions, and a dialer's Pool opening sessions on it. In short:

c, err := link.Client(ctx, nc, link.Config{Window: 4 << 20, Liveness: link.LivenessFast, Auth: auth}) // dialer
m := link.NewMux(c)
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).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. make check-grpclink: the gRPC implementation's module.

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. v0.3.0: Session.Reset (CLOSE code 2, announced by feature bit FeatureReset; sent as a plain close to v0.2 peers), a session cut with its connection reads the connection's error (never io.EOF), Session.Done, session read deadlines (net/http serves sessions). Write deadlines are not supported. v0.4.0: Session.Pump (a session's bytes go to a writer from a goroutine that runs only while bytes are queued: an idle session holds none), Session.WriteTo returns received buffers to the pool after a writev (it dropped them), a drained bytequeue reuses its array, bytequeue.TryPopBatch and Ready. No wire change.

v0.5.0 (breaking): wire version 1 removed. Version 2 is the only wire version: Hello, ReadHello, Reply, Version, TokenLen, NewConn, Codec and Message are gone, NewMux takes only the connection, and Config.Version is 0 or 2. A version-1 hello (a dialer of v0.4 or older) gets the version reply, which such a dialer reads as a refusal, and Server returns ErrV1 (errors.Is ErrVersion). A v0.5 dialer to a listener that speaks only version 1 is refused after the listener read 37 bytes. Version 2 on the wire is unchanged (testdata/v2_session.golden).

v0.6.0: peerlink (the Link API, its contract, Node over Pool and Mux), peerlink/peerlinktest (the conformance suite) and the peerlink/grpclink module. No wire change.

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 2

Version 2 is the only version since v0.5.0 (version 1, the hello of a bare token and opens and closes in the caller's messages, was removed; a version-1 hello gets the version reply, below, and Server ErrV1). 

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  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)
  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);
                   2 as 0, and the session failed (reset; only to a
                   peer that announced FeatureReset)
  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)

Integers are big endian. A session id (sid) is at most 255 bytes. A window is how many bytes of one session the side that announces it buffers (see Credit.Acquire). Session ids that begin with '_' name control frames, which Mux leaves to the caller.

A listener answers a version it does not speak with status 2. A listener of link v0.4 or older that speaks only version 1 reads 37 bytes of a version-2 hello before it closes the connection.

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: a side announces the ones it reads and ignores the bits it does not know. Bit 0, FeatureReset (since v0.3.0): the side reads CLOSE code 2; v0.2 sides announce none and end the connection on code 2, so a reset to them is sent as code 0. 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 (dead within interval + timeout of the peer's last frame, idle or busy); every ping is answered (the latest, when several wait). 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. The sender follows its goaway with a ping: opens that arrive before that ping's pong crossed the goaway and run like any other session (since v0.4.2; before, they were refused with a close), opens after it are refused; the sender closes the connection once its last session ended and the pong arrived, after writing every frame.

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

Example

Example is a listener and a dialer end to end over TLS on loopback: the listener authenticates the dialer and serves each session it opens (here, upper-casing what it reads); the dialer keeps a Pool of connections to it and opens a session per request. 

package main

import (
	"bytes"
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/subtle"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"io"
	"math/big"
	"net"
	"time"

	"github.com/nfltr/link"
)

// tokenAuth is a shared-secret Authenticator: the dialer sends the token,
// the listener compares it. Method 128 and up are the caller's to define.
type tokenAuth []byte

func (t tokenAuth) Hello(context.Context, tls.ConnectionState) (byte, []byte, error) {
	return 128, t, nil
}

func (t tokenAuth) Verify(_ context.Context, _ tls.ConnectionState, method byte, payload []byte) (link.Identity, error) {
	if method != 128 || subtle.ConstantTimeCompare(payload, t) != 1 {
		return link.Identity{}, errors.New("bad token")
	}
	return link.Identity{Method: "token", Subject: "example-client"}, nil
}

// Example is a listener and a dialer end to end over TLS on loopback: the
// listener authenticates the dialer and serves each session it opens (here,
// upper-casing what it reads); the dialer keeps a Pool of connections to it
// and opens a session per request.
func main() {
	ctx := context.Background()
	auth := tokenAuth("s3cret")
	serverTLS, clientTLS := exampleTLS()

	// Listener: one link per accepted connection, one goroutine per session.
	ln, err := tls.Listen("tcp", "127.0.0.1:0", serverTLS)
	if err != nil {
		panic(err)
	}
	defer ln.Close()
	who := make(chan link.Identity, 1)
	go func() {
		for {
			nc, err := ln.Accept()
			if err != nil {
				return
			}
			go func() {
				c, id, err := link.Server(ctx, nc, link.Config{Auth: auth})
				if err != nil {
					nc.Close()
					return
				}
				who <- id
				_ = link.NewMux(c).Serve(func(header []byte, s *link.Session) {
					defer s.Close()
					body, _ := io.ReadAll(s) // until the dialer's CloseWrite
					fmt.Fprintf(s, "%s: %s", header, bytes.ToUpper(body))
				})
			}()
		}
	}()

	// Dialer: a Pool dials (TLS is the caller's), authenticates, and opens
	// sessions on its connections; the listener adopts the dialer's
	// liveness, so both ends notice a dead path within 1.5 s + 3 s.
	p := link.NewPool(link.PoolConfig{
		Link: link.Config{Auth: auth, Liveness: link.LivenessFast},
		Dial: func(ctx context.Context) (net.Conn, error) {
			var d tls.Dialer
			d.Config = clientTLS
			return d.DialContext(ctx, "tcp", ln.Addr().String())
		},
		Max:     2,
		Backoff: func(attempt int) time.Duration { return time.Duration(attempt) * 100 * time.Millisecond },
	})
	defer p.Close()

	for _, msg := range []string{"hello", "world"} {
		s, err := p.Open(ctx, []byte("upper"))
		if err != nil {
			panic(err)
		}
		_, _ = io.WriteString(s, msg)
		_ = s.CloseWrite() // the listener reads io.EOF after msg
		reply, err := io.ReadAll(s)
		s.Close()
		if err != nil {
			panic(err)
		}
		fmt.Println(string(reply))
	}
	fmt.Println("listener saw:", (<-who).Subject)
}

// exampleTLS is a throwaway self-signed certificate for 127.0.0.1 and the
// client config that trusts it.
func exampleTLS() (server, client *tls.Config) {
	key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	tmpl := &x509.Certificate{
		SerialNumber: big.NewInt(1),
		IPAddresses:  []net.IP{net.ParseIP("127.0.0.1")},
		NotBefore:    time.Now().Add(-time.Minute),
		NotAfter:     time.Now().Add(time.Hour),
	}
	der, _ := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
	cert, _ := x509.ParseCertificate(der)
	roots := x509.NewCertPool()
	roots.AddCert(cert)
	return &tls.Config{Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}, MinVersion: tls.VersionTLS13},
		&tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS13}
}

Output:
upper: HELLO
upper: WORLD
listener saw: example-client

Index

Examples

Constants

View Source 
const (
	MethodNone  byte = 0 // no credential (NoAuth)
	MethodToken byte = 1 // a bearer token the listener issued
)

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 (StatusVersion too).

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

Window bounds: a side announces a window in [MinWindow, MaxWindow]. Credit returns in steps of a quarter window; a window of two MaxChunks always returns a waiting sender the MaxChunk it asks for.

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 (
	// FeatureReset: the side reads CLOSE code 2 (a reset). A side sends
	// code 2 only to a peer that announced it, else code 0.
	FeatureReset uint32 = 1 << 0
)

Feature bits a side announces in its hello or reply.

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. 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 listener's reply to a hello whose version it does not speak; its highest version follows.

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")
	// ErrV1 is a listener's refusal of a version-1 hello: a dialer of
	// link v0.4 or older (version 1 was removed in v0.5.0).
	ErrV1 = fmt.Errorf("link: a version-1 hello: the peer is older than link v0.5.0, which speaks only version 2: %w", ErrVersion)
	// 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 ErrConnClosed = errors.New("link: connection closed")

ErrConnClosed is what a session reads once its connection closed with no error of the link's (the network's, or Close).

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 ErrReset = errors.New("link: session reset by the peer")

ErrReset is what a session reads once the peer reset it (Reset).

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 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 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 (a dialer of link v0.4 or older) gets the version reply (StatusVersion, MaxVersion), which such a dialer reads as a refusal, and Server returns ErrV1. 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 Config added in v0.2.0 

type Config struct {
	// Version is the wire version to speak: 0 is MaxVersion, the only
	// version since v0.5.0.
	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: 0 is MaxSessions.
	MaxSessions int
	// Liveness: 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; a listener of link v0.4 or older that speaks only version 1 closes the connection after it read 37 bytes. A refusal is ErrRefused. On error the caller closes nc.

func (*Conn) Close 

func (c *Conn) Close() error

Close closes the connection; senders waiting for credit give up. ConnDown gets why it ended: ErrDeadPeer, what the peer or the network did (lost), or nil when this side closed it.

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) Drained added in v0.4.2 

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

Drained is closed once the peer answered a ping this side sent after its GOAWAY (Mux.GoAway sends one at once): the peer has read the GOAWAY, so it opens nothing more here, and every OPEN it sent before has arrived (frames keep their order). An OPEN that crossed the GOAWAY is a session like any other; one after Drained breaks the protocol. A peer before v0.4.2 answers only one of two pings that arrive together: then a later ping of the liveness's closes it, or the drain deadline ends the connection.

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 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) PeerFeatures added in v0.3.0 

func (c *Conn) PeerFeatures() uint32

PeerFeatures is the feature bits the peer announced. allow:wrapper

func (*Conn) PeerGoAway added in v0.2.0 

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

PeerGoAway is closed once the peer sent GOAWAY.

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. 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 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 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) SendReset added in v0.3.0 

func (c *Conn) SendReset(sid string) error

SendReset is SendClose of the whole session saying it failed: a peer that announced FeatureReset reads it as a reset, another as a close.

func (*Conn) Version added in v0.2.0 

func (c *Conn) Version() int

Version is the connection's wire version: 2, the only one since v0.5.0.

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
	Reset  bool // a close that says the session failed (code 2)
}

Frame is a received frame: a session's data (SID, Data), a message (Msg, the caller's bytes), 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 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).

Opens and closes 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) *Mux

NewMux returns a Mux over c.

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 tells the peer to open no more sessions here and stops this side opening more; the connection closes once its last session ended and the peer read the GOAWAY (an OPEN it sent before that runs like the others), with every frame written; at deadline at the latest (zero: no deadline).

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 (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 header that opened it; 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
	SessionClose(error)       // one of those ended
	CreditWait(time.Duration) // a sender waited this long for credit
	PingRTT(time.Duration)    // a ping's round trip
	GoAway(Identity)          // the peer sent GOAWAY
}

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.
	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, not counting those going away
	// (GOAWAY), which take no new session.
	Min, Max int
	// Grow, when set, says whether Open dials one more connection (below
	// Max) although one could take the session: in the background, one
	// such dial at a time. nil grows only when none can.
	Grow func(PoolStats) bool
	// GrowWait is how long the Open that started a growth dial waits for
	// it: the session goes on the new connection when it is up by then,
	// else on one that is up (the dial goes on). 0: it does not wait.
	// Sessions opened in a burst spread over the connections only when
	// the Open waits.
	GrowWait time.Duration
	// 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), or the error that cut it: ErrReset (the peer's Reset), the connection's (ErrDeadPeer, ErrConnClosed). Write sends on credit. A read deadline ends a pending Read with os.ErrDeadlineExceeded; write 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 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) Done added in v0.3.0 

func (s *Session) Done() <-chan struct{}

Done is closed once the session ended: both sides closed it, the peer reset it, or its connection closed. Bytes already received stay readable.

func (*Session) Header added in v0.2.0 

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

Header is the bytes that opened the session: the OPEN frame's header. allow:wrapper

func (*Session) LocalAddr 

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

LocalAddr is the connection's. allow:wrapper

func (*Session) Pump added in v0.4.0 

func (s *Session) Pump(w io.Writer, done func(n int64, err error))

Pump writes the peer's bytes to w as WriteTo does, without a goroutine waiting while the session is idle: one runs only while bytes are queued, started as they arrive (the receive loop never waits for it). done gets what WriteTo would return, once, from that goroutine: the session ended, or w failed (the session's further bytes stay unread: close it). Pump takes over the session's reads: no Read, WriteTo or second Pump after it.

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) Reset added in v0.3.0 

func (s *Session) Reset() error

Reset ends the session as failed: the peer reads ErrReset (a peer that announced FeatureReset; else it reads a close). Reads here end as after Close.

func (*Session) SetDeadline 

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

SetDeadline is SetReadDeadline: write deadlines are not supported.

func (*Session) SetReadDeadline 

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

SetReadDeadline makes a pending or later Read return os.ErrDeadlineExceeded once t passed; zero t clears it. WriteTo ignores it.

func (*Session) SetWriteDeadline 

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

SetWriteDeadline is a no-op: Close ends a session whose writes wait.

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) WriteBuffers added in v0.3.0 

func (s *Session) WriteBuffers(bufs [][]byte) (int64, error)

WriteBuffers is Write of each of bufs in turn, with one flush at the end (and before any wait for credit): a backlog goes to the network in one write.

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
Package peerlink is a small, transport-neutral API for byte-stream sessions between peers (Link), an implementation of it over github.com/nfltr/link (Node), and, in peerlinktest, a conformance suite any implementation runs to show it keeps the contract below.
 Package peerlink is a small, transport-neutral API for byte-stream sessions between peers (Link), an implementation of it over github.com/nfltr/link (Node), and, in peerlinktest, a conformance suite any implementation runs to show it keeps the contract below.
peerlinktest
Package peerlinktest is the conformance suite of the peerlink contract: Run checks an implementation of peerlink.Link against every rule of the contract, on pairs of Links the implementation's test makes.
 Package peerlinktest is the conformance suite of the peerlink contract: Run checks an implementation of peerlink.Link against every rule of the contract, on pairs of Links the implementation's test makes.
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.