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

link

package module
v0.8.0 Latest Latest
Warning

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

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

README

github.com/nfltr/link multiplexes many byte-stream sessions, each a net.Conn with its own flow control, over connections you dial and secure (plain TCP, TLS, mutual TLS). A slow reader stalls only its own session, a dead peer is noticed within a bound you pick, and the main module imports only the Go standard library (Apache-2.0).

Quick start

Two peers in one program: a dials b over plaintext TCP, then each opens a session to the other on that one connection. Run it with go run github.com/nfltr/link/examples/quickstart@latest, or save it as main.go in a module of your own (go get github.com/nfltr/link@latest):

// Command quickstart runs two peers, a and b, in one process: a dials b
// over plaintext TCP, then each opens a session to the other on that one
// connection.
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"time"

	"github.com/nfltr/link"
	"github.com/nfltr/link/peerlink"
)

func main() {
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		log.Fatal(err)
	}

	// b listens. NameAuth: each peer sends its name and is known by it
	// (a trusted network only; see docs/recipes.md for mutual TLS).
	b, err := peerlink.New(peerlink.Config{
		Listener: ln,
		Link:     link.Config{Auth: link.NameAuth("b")},
	})
	if err != nil {
		log.Fatal(err)
	}
	go b.Serve(reply("b"))

	// a dials b, and keeps a connection to it (Connect) so that b can
	// open sessions to a as well.
	a, err := peerlink.New(peerlink.Config{
		Link:    link.Config{Auth: link.NameAuth("a")},
		Connect: []string{"b"},
		Dial: func(ctx context.Context, peer string) (net.Conn, error) {
			var d net.Dialer
			return d.DialContext(ctx, "tcp", ln.Addr().String()) // peer is "b"
		},
	})
	if err != nil {
		log.Fatal(err)
	}
	go a.Serve(reply("a"))

	fmt.Println(ask(a, "b")) // a opens a session to b
	fmt.Println(ask(b, "a")) // b opens one to a, on the connection a dialled

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_ = a.Drain(ctx)
	_ = b.Drain(ctx)
}

// reply serves each session a peer opens: it reads the request and
// answers it.
func reply(me string) func(peer string, header []byte, s net.Conn) {
	return func(peer string, header []byte, s net.Conn) {
		defer s.Close()
		req, _ := io.ReadAll(s) // until the opener's CloseWrite
		fmt.Fprintf(s, "%s got %q (%s) from %s", me, req, header, peer)
	}
}

// ask opens a session to peer, sends a request and reads the answer.
func ask(l peerlink.Link, peer string) string {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	s, err := l.Open(ctx, peer, []byte("greeting"))
	if err != nil {
		log.Fatal(err)
	}
	defer s.Close()
	if _, err := io.WriteString(s, "hello"); err != nil {
		log.Fatal(err)
	}
	// The peer reads io.EOF after "hello"; this side can still read.
	if err := s.(peerlink.Session).CloseWrite(); err != nil {
		log.Fatal(err)
	}
	answer, err := io.ReadAll(s)
	if err != nil {
		log.Fatal(err)
	}
	return string(answer)
}

It prints:

b got "hello" (greeting) from a
a got "hello" (greeting) from b

In two processes, one runs b's half (net.Listen, a Node with Listener, Serve) and the other a's (a Node whose Dial reaches b's address, with Connect: []string{"b"} so that b can open sessions back, and Serve); either then opens sessions to the other. For mutual TLS, gRPC or HTTP over a link, liveness, draining, reconnects and metrics, see the recipes.

Which API

you want use
sessions between named peers over TCP or TLS: start here peerlink.Node (peerlink)
the same over QUIC (UDP), surviving address changes quiclink.Node (peerlink/quiclink, a module of its own)
the same over a network that passes only gRPC grpclink.Node (peerlink/grpclink, a module of its own)
gRPC or HTTP served over a link peerlink.Listener (recipes)
code that works with any of these, and tests that check them the peerlink.Link interface; peerlinktest.Run (the conformance suite)
advanced: your own connection management link.Pool (connections to one peer), link.Mux (sessions on one connection), link.Client/link.Server (the handshake)
advanced: your own receive loop or transport link.Conn (frames), link.Credit (per-session flow control)

Constructors check their configuration and name the field to fix (peerlink.New, link.NewPool, link.Config.Validate); a zero link.Config is valid. Each package's documentation (go doc, or pkg.go.dev) has an Example for each common use.

Errors

Test errors with errors.Is (or errors.As for *link.VersionError); the messages name what to change.

error where you see it what it means what to change
link.ErrHandshake (wraps the network's error) Open (dialing), Client, Server, peerlink.Config.HandshakeError the hello or its reply did not arrive whole: the other end closed or timed out, is not link, or speaks TLS where this end does not TLS on both ends or neither; the address of a link listener
link.ErrRefused Open (dialing), Client the listener's Authenticator refused the dialer's credential the dialer's Config.Auth: what the listener accepts (CertAuth's certificate, the token)
link.ErrAuth (wraps Verify's error) Server, peerlink.Config.HandshakeError this listener refused a dialer the reason it wraps; the dialer's credential or this Config.Auth
link.ErrVersion, *link.VersionError, link.ErrV1 Client, Server the peer speaks another wire version upgrade the older end (every release since v0.5.0 speaks version 2)
link.ErrDeadPeer session reads, Mux.Serve, Observer.ConnDown nothing arrived for the liveness bound: the peer or the path died nothing, if it did; else a preset with a longer bound for a path that goes silent (LivenessNAT, LivenessMobile)
link.ErrGoAway Mux.Open, Conn.SendOpen the connection is going away (a drain) open on another connection (Pool and Node do)
link.ErrFull Open the peer's most sessions at once are open close sessions sooner, raise the peer's Config.MaxSessions, or allow more connections (Pool.Max)
peerlink.ErrDraining Node.Open this Node is draining nothing: it is shutting down
peerlink.ErrPeerUnreachable (and the context's error) Node.Open without Dial no connection from that peer before the context ended the peer's name (the error lists the connected ones); that peer's Connect and Dial; a longer context
link.ErrReset session reads the peer reset the session the peer's reason
link.ErrConnClosed session reads the connection closed under the session open a new session (Pool and Node redial)
link.ErrFrame, link.ErrCredit Conn.Recv, Mux.Serve the peer broke the wire protocol a peer that is not link, or a bug: please report it
a message naming a field peerlink.New, link.NewPool, Validate, Client, Server a configuration that cannot work the field it names, as it says
io.EOF session reads the peer ended its writes on purpose (CloseWrite, Close) nothing: the normal end
os.ErrDeadlineExceeded session reads the read deadline passed nothing, or a later deadline

quiclink adds its own codes (quiclink.Code*, in the QUIC errors its sessions read); grpclink's are gRPC statuses.

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, Node (an implementation over Pool and Mux) and Listener (peerlink/README.md)
link/peerlink/peerlinktest the contract's conformance suite (Run) and the fault-injecting Proxy (TCP) and UDPProxy, for any implementation
link/examples/quickstart the quick start above

peerlink/grpclink (a module of its own, make check-grpclink) implements peerlink.Link over gRPC streams and passes the same suite (one direction: the dialer opens). peerlink/quiclink (a module of its own, make check-quiclink) implements it over QUIC streams and passes the suite in both directions; its connections survive the dialler's address changing (peerlink/quiclink/README.md).

Wire format version 2 is specified in WIRE.md, with test vectors for every message (testdata/wire_vectors.json); testdata/v2_session.golden holds a recorded session. Every release is tested against both, byte for byte.

Checks

make check: build, vet, tests with -race (the Examples, and TestDocCodeBlocksAreCompiled: every Go block in the Markdown here is a verbatim excerpt of a Go file, so a doc cannot rot), fuzz-short (every decoder, FUZZTIME each, default 10s), gosec, semgrep (docker), govulncheck. make check-api: the exported API against the latest release, with apidiff; the changes must be the ones listed in api/ (an API change is never an accident). make check-grpclink, make check-quiclink: the gRPC and QUIC implementations' modules. GitHub Actions (.github/workflows/check.yml) runs them on every push and pull request, and every fuzz target for 5 minutes weekly. make fuzz-long FUZZTIME=30m is the long fuzz run (FUZZING.md).

Status and compatibility

The latest release is v0.8.0; CHANGELOG.md lists every release. link is pre-1.0 and follows semantic versioning as Go modules do for v0:

  • The wire is the stronger promise: version 2 (WIRE.md) is held byte for byte by its test vectors and golden recording. A change to it is a new wire version, negotiated in the hello, and announced in the changelog; feature bits add optional frames without one.
  • The Go API may change in a minor release (v0.x.0), with the change and its migration in the changelog; a patch release (v0.x.y) never removes or changes an exported name. Removals are announced a minor release ahead where possible.
  • Behaviour that the docs state (bounds, errors to test with errors.Is, the peerlink contract) changes only in a minor release, except to fix a bug, which the changelog flags.
  • peerlink/grpclink and peerlink/quiclink are modules of their own, versioned on their own (peerlink/<name>/vX.Y.Z).

v1.0.0 will freeze the API under Go's compatibility rules once the maturity criteria below hold.

Maturity

Evidence that link works, provable from this repository alone:

criterion evidence status
wire specified, with test vectors WIRE.md, testdata/wire_vectors.json (TestWireVectors), testdata/v2_session.golden met
every decoder fuzzed, corpus checked in, 24 h cumulative without a crasher FUZZING.md (hours per target, crashers), testdata/fuzz/ 4.5 h, no crasher in the library (target: 24 h)
security review of the hello and the Authenticator contract SECURITY.md (threat model, review record) met (2026-10-14)
standard library only TestModuleBoundary met
CI green on every change, with history GitHub Actions (check) since 2026-10-14
cumulative soak hours under faults, with leak and integrity checks SOAK.md (make soak), summaries in soak/results/ 2.35 h (macOS, Linux CI), clean (target: 24 h)
benchmarks published in the repository compare/README.md (link vs yamux), peerlink/quiclink/README.md met
API reviewed for v1.0 the checklist below; make check-api (apidiff against the latest release, in CI); TestExportedNamesDocumented open (v0.8.0: API trimmed, constructors checked, every name documented)

v1.0 API checklist: every exported name documented with an example where it is not obvious; errors to test with errors.Is listed per call; no exported name that exists only for one caller; option structs instead of positional growth; the peerlink contract unchanged for two minor releases; deprecated names removed. Use in production by its first user (an HTTP and TCP relay) is supporting evidence, not a criterion.

Documentation

Overview

Package link multiplexes many byte-stream sessions, each a net.Conn with its own flow control, over one connection that the caller dials and secures (plain TCP, TLS, a pipe). A session whose reader is slow stalls only itself, dead peers are noticed within a bound the caller picks, and the package imports only the Go standard library.

Where to start

Most programs want github.com/nfltr/link/peerlink: a Node opens sessions to named peers and serves the sessions they open, keeping the connections (dialling, redialling, draining) for it. The README's quick start and docs/recipes.md (plaintext and mutual TLS, liveness presets, gRPC and HTTP over a link, graceful drain, reconnects, metrics) build on it.

This package is the layer below, for callers that manage connections themselves:

  • Client and Server run the handshake on a connection (the hello, an Authenticator's credential) and return a Conn.
  • Mux carries sessions over a Conn as [Session]s (net.Conns).
  • Pool keeps several connections to one peer, with growth, dial backoff and make-before-break Pool.Drain.
  • Conn and Credit are frames and per-session flow control, for a caller with its own receive loop or its own transport.

Authenticators: NameAuth on a trusted network, CertAuth over mutual TLS, NoAuth where peers need no names, or the caller's own (SECURITY.md). Config.Validate and PoolConfig.Validate name the field to fix.

Liveness

A connection is dead once nothing arrived for Interval + Timeout (ErrDeadPeer): each side pings once it received nothing for Interval. Pick a preset by how fast a dead peer must be noticed against how many idle frames you can afford (one small frame each way per Interval per connection):

  • LivenessFast (4.5 s): failover and interactive use on a good network.
  • LivenessDefault (15 s): most services; what a zero Config gets.
  • LivenessNAT (30 s): paths through NATs, conntrack or load balancers that drop mappings idle for 30 s or more.
  • LivenessDatacenter (45 s): stable networks where fewer wakeups matter more than fast failover.
  • LivenessMobile (60 s): laptops and phones whose radio sleeps or hands over: the fewest wakeups, and a short silence is not taken for death.

ParseLiveness reads a preset by name, with overrides ("nat,timeout=40s"), for a flag or a config file. The listener takes the dialer's liveness unless its Config sets one.

Errors

Every error to test with errors.Is, by where it comes from (the README has the same as a table, with what to change):

  • Client: ErrRefused (the listener refused the credential), ErrVersion (errors.As a *VersionError), ErrHandshake (the reply did not arrive: not a link listener, or a TLS mismatch; it wraps the network's error), ErrFrame, ctx's error.
  • Server: ErrAuth (the Authenticator refused; it wraps Verify's error), ErrV1 and ErrVersion, ErrHandshake (with ErrFrame when the first bytes are not a hello), ctx's error.
  • Conn.Recv, Mux.Serve: ErrFrame, ErrCredit, ErrFull (the peer broke the protocol), ErrDeadPeer (liveness ended the connection), the network's (io.EOF once the peer closed).
  • Conn.SendOpen, Mux.Open, Pool.Open: ErrFull (the peer's max sessions), ErrGoAway (the connection is going away); Mux.Open after Serve returned: net.ErrClosed.
  • Session reads: io.EOF (the peer ended its writes on purpose), ErrReset (the peer's Reset), ErrDeadPeer or ErrConnClosed (the connection ended under the session), os.ErrDeadlineExceeded.
  • Config.Validate, PoolConfig.Validate, NewPool, Client, Server: a message naming the field and what to set.

Wire format, version 2

WIRE.md in the module is the full specification, with test vectors for every message (testdata/wire_vectors.json); in short:

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; Conn.SendMsg
                   and Conn.Recv carry them, a Mux drops them)
  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, reset (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
}

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() {
				// The handshake's deadline: a dialer that never finishes its
				// hello does not hold the connection (SECURITY.md).
				hctx, cancel := context.WithTimeout(ctx, 10*time.Second)
				c, id, err := link.Server(hctx, nc, link.Config{Auth: auth})
				cancel()
				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, err := 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 },
	})
	if err != nil {
		panic(err)
	}
	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, CertAuth)
	MethodToken byte = 1 // a bearer token the listener issued
	MethodName  byte = 2 // a name the dialer claims (NameAuth)
)

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 (
	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 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.

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")
	// ErrHandshake is a hello or its reply that did not arrive whole:
	// the other end closed the connection, timed out, or is not a link
	// end speaking the same TLS-or-not (the error says which, and wraps
	// the network's).
	ErrHandshake = errors.New("link: handshake failed")
	// 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 wire format (WIRE.md): a frame, or a hello or reply with a field out of its bounds.

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 version 2 (MaxVersion). 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.

ctx bounds the handshake: give it a deadline. Without one, a dialer that sends its hello slowly, or not at all, holds nc and this goroutine for as long as it likes.

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.

Verify runs before the listener has bound anything, on bytes from an unauthenticated peer (SECURITY.md): it compares secrets in constant time (crypto/subtle), bounds its own work (ctx carries the handshake's deadline), and returns an Identity whose Subject names the peer uniquely where callers route by it (peerlink names a peer by Subject). Its error is not sent to the dialer, which reads only a refusal. 

var CertAuth Authenticator = certAuth{}

CertAuth is an Authenticator for mutual TLS: the dialer sends no credential, and the listener accepts a dialer whose client certificate its tls.Config verified (ClientAuth: tls.RequireAndVerifyClientCert), named by the certificate's Subject CommonName, else its first DNS name. Give Server (or peerlink.Config.Listener) a *tls.Conn. 

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). Every peer it accepts has the zero Identity: a peerlink.Node names them all "" (see NameAuth and CertAuth for peers with names).

func NameAuth added in v0.8.0 

func NameAuth(name string) Authenticator

NameAuth is an Authenticator for a trusted network (plaintext inside a cluster, a pipe): the dialer sends name, and the listener accepts any name of 1 to MaxSID bytes as the peer's Identity.Subject. It proves nothing: anyone who reaches the listener can claim any name. Over an untrusted network use CertAuth with mutual TLS, or an Authenticator that checks a secret.

type Config added in v0.2.0 

type Config struct {
	// 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.
	// A listener that must bound how long a vanished dialer holds its
	// connection sets its own: the dialer's Timeout has no upper 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:
	// NameAuth on a trusted network, CertAuth over mutual TLS, NoAuth
	// where peers need no names, or the caller's own.
	Auth Authenticator
	// Observer receives the connection's events; nil is none.
	Observer Observer
}

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

func (Config) Validate added in v0.8.0 

func (cfg Config) Validate() error

Validate reports the first field of cfg out of its range, naming it and what to set; the zero Config is valid. Client and Server (and NewPool, for PoolConfig.Link) refuse a Config it refuses.

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. On a plaintext connection each frame header is a read of the socket: where one carries heavy traffic, give Client or Server a net.Conn whose Read goes through a bufio.Reader.

Example

ExampleConn drives a Conn without a Mux: message frames, the caller's own bytes, one receive loop. 

package main

import (
	"context"
	"fmt"
	"net"
	"time"

	"github.com/nfltr/link"
)

// bind binds both ends of one connection (here an in-memory pipe; in
// practice a TLS connection the caller dialled and accepted).
func bind(auth link.Authenticator) (dialer, listener *link.Conn) {
	a, b := net.Pipe()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	type bound struct {
		c   *link.Conn
		err error
	}
	ch := make(chan bound, 1)
	go func() {
		c, _, err := link.Server(ctx, b, link.Config{Auth: auth})
		ch <- bound{c, err}
	}()
	dialer, err := link.Client(ctx, a, link.Config{Auth: auth})
	l := <-ch
	if err != nil || l.err != nil {
		panic(fmt.Sprint(err, l.err))
	}
	return dialer, l.c
}

func main() {
	dc, lc := bind(link.NoAuth)
	defer dc.Close()
	defer lc.Close()

	go func() {
		for _, msg := range []string{"one", "two"} {
			_ = dc.SendMsg([]byte(msg))
		}
		_ = dc.Flush() // a send buffers; Flush writes
	}()
	for range 2 {
		f, err := lc.Recv()
		if err != nil {
			panic(err)
		}
		fmt.Println(string(f.Msg))
	}
}

Output:
one
two

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 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.

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.

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).

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.

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. The peer reads it from Recv (Frame.Msg); a Mux on the peer drops it, so messages are for callers that drive the Conn themselves.

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.

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.

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.

Example

ExampleParseLiveness: the presets, and a preset with an override (from a flag, say). A connection is dead once nothing arrived for Interval + Timeout; idle, each side sends a frame every Interval. 

package main

import (
	"fmt"

	"github.com/nfltr/link"
)

func main() {
	for _, spec := range []string{"fast", "default", "nat", "datacenter", "mobile", "nat,timeout=40s"} {
		l, err := link.ParseLiveness(spec)
		if err != nil {
			panic(err)
		}
		fmt.Printf("%-16s a frame every %v idle, dead after %v\n", spec, l.Interval, l.Interval+l.Timeout)
	}
}

Output:
fast             a frame every 1.5s idle, dead after 4.5s
default          a frame every 5s idle, dead after 15s
nat              a frame every 10s idle, dead after 30s
datacenter       a frame every 15s idle, dead after 45s
mobile           a frame every 15s idle, dead after 1m0s
nat,timeout=40s  a frame every 10s idle, dead after 50s

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.

Message frames (Conn.SendMsg) are a Conn-level feature for callers that run their own receive loop; a Mux owns the Conn's receive loop and drops them. Per-peer control traffic on a Mux goes in a session whose header names its purpose.

Example

ExampleMux carries sessions over one connection without a Pool: each side wraps its Conn in a Mux, the listener serves the sessions the dialer opens, and each session is a net.Conn. 

dc, lc := bind(tokenAuth("s3cret"))

go link.NewMux(lc).Serve(func(header []byte, s *link.Session) {
	defer s.Close()
	n, _ := io.Copy(io.Discard, s) // until the opener's CloseWrite
	fmt.Fprintf(s, "%s: %d bytes", header, n)
})

m := link.NewMux(dc)
go m.Serve(nil) // the dialer takes no sessions
defer m.Close()
s, err := m.Open(func(sid string) ([]byte, error) { return []byte("count"), nil })
if err != nil {
	panic(err)
}
_, _ = io.WriteString(s, "hello, world")
_ = s.CloseWrite()
reply, _ := io.ReadAll(s)
_ = s.Close()
fmt.Println(string(reply))

Output:
count: 12 bytes

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.

func (*Mux) Conn added in v0.2.0 

func (m *Mux) Conn() *Conn

Conn is the Mux's connection.

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)

ConnDown ignores the event.

func (NopObserver) ConnUp added in v0.2.0 

func (NopObserver) ConnUp(Identity)

ConnUp ignores the event.

func (NopObserver) CreditWait added in v0.2.0 

func (NopObserver) CreditWait(time.Duration)

CreditWait ignores the event.

func (NopObserver) GoAway added in v0.2.0 

func (NopObserver) GoAway(Identity)

GoAway ignores the event.

func (NopObserver) PingRTT added in v0.2.0 

func (NopObserver) PingRTT(time.Duration)

PingRTT ignores the event.

func (NopObserver) SessionClose added in v0.2.0 

func (NopObserver) SessionClose(error)

SessionClose ignores the event.

func (NopObserver) SessionOpen added in v0.2.0 

func (NopObserver) SessionOpen()

SessionOpen ignores the event.

type Observer added in v0.2.0 

type Observer interface {
	ConnUp(Identity)          // the handshake bound the connection
	ConnDown(Identity, error) // it ended: ErrDeadPeer, the peer's or network's error, nil when this side closed it
	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 on the Pool's connections go to Accept (nil: refused).

func NewPool added in v0.2.0 

func NewPool(cfg PoolConfig) (*Pool, error)

NewPool returns a Pool; it dials its Min connections in the background. It refuses a PoolConfig that Validate refuses.

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
	// Accept, when set, takes the sessions the peer opens on the Pool's
	// connections (Mux.Serve's accept: it owns the session and closes
	// it). nil: they are refused.
	Accept func(open []byte, s *Session)
}

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

func (PoolConfig) Validate added in v0.8.0 

func (cfg PoolConfig) Validate() error

Validate reports the first field of cfg that NewPool refuses, naming it and what to set: Dial is required; Min and Max are not negative, and Min is at most a Max that is set; Link is valid (Config.Validate).

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.

func (*Session) LocalAddr 

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

LocalAddr is the connection's.

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)

Read reads the peer's bytes in order: io.EOF once the peer ended its writes, or the error that cut the session (see Session).

func (*Session) RemoteAddr 

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

RemoteAddr is the connection's.

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

Error says which versions the peer speaks.

func (*VersionError) Is added in v0.2.0 

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

Is makes a VersionError an ErrVersion.

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.
examples
quickstart command
Command quickstart runs two peers, a and b, in one process: a dials b over plaintext TCP, then each opens a session to the other on that one connection.
 Command quickstart runs two peers, a and b, in one process: a dials b over plaintext TCP, then each opens a session to the other on that one connection.
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.