transport

package
v1.17.0 Latest Latest
Warning

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

 Go to latest Published: Aug 18, 2026 License: AGPL-3.0 Imports: 14 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrAddrAlreadyInUse = errors.New("address already in use")
View Source 
var ErrConnClosed = errors.New("connection closed")
View Source 
var ErrIdleTimeout = errors.New("transfer stalled: idle timeout")

ErrIdleTimeout reports that a transfer stopped making progress and was aborted locally. It is distinct from a peer-initiated abort so a caller can say which side gave up.

View Source 
var ErrListenerClosed = errors.New("listener closed")
View Source 
var ErrMissingAddr = errors.New("missing address")
View Source 
var ErrNoListener = errors.New("no listener")
View Source 
var ErrTransportClosed = errors.New("transport closed")

Functions

This section is empty.

Types

type Addr 

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

Addr names one endpoint: the network carrying it and the host:port that selects it there. Both transports address their endpoints this way.

func NewAddr 

func NewAddr(network Network, hostPort string) *Addr

func (*Addr) Network 

func (a *Addr) Network() string

func (*Addr) String 

func (a *Addr) String() string

type Conn 

type Conn interface {
	AcceptStream(ctx context.Context) (Stream, error)
	OpenStream(ctx context.Context) (Stream, error)
	LocalAddr() net.Addr
	RemoteAddr() net.Addr
	Context() context.Context
	Close() error // Idempotent
}

type ConnErrorCode 

type ConnErrorCode uint64

ConnErrorCode is reported to the peer when a connection is closed with a reason. ConnCodeShutdown is the only one a transport sends. 

const ConnCodeShutdown ConnErrorCode = 3

ConnCodeShutdown is reported to the peer when a connection is closed because the listener is going away.

type IdleGuard added in v1.17.0 

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

IdleGuard bounds a transfer by progress rather than by total duration. It wraps the reader driving the transfer and aborts the stream when a single read makes no progress within the idle timeout.

Progress, not duration, is the right bound for bulk data: a transfer that keeps moving is never cut off however large it is, while one that stalls is aborted promptly. A total cap belongs on the small fixed exchanges either side of the body, where the size is known in advance.

func NewReadGuard added in v1.17.0 

func NewReadGuard(r io.Reader, stream Stream, idle time.Duration) *IdleGuard

NewReadGuard guards a body being read from stream: a stall aborts the read side. Only the read itself is bounded, so a caller that pauses between reads is never punished for it.

func NewWriteGuard added in v1.17.0 

func NewWriteGuard(src io.Reader, stream Stream, idle time.Duration) *IdleGuard

NewWriteGuard guards a body being written to stream. It wraps the source the copy pulls from: a source read happens only once the previous write has landed, so a write that blocks shows up as a read that never comes. The timer therefore stays armed across the write and is reset by the next read.

func (*IdleGuard) Expired added in v1.17.0 

func (g *IdleGuard) Expired() bool

Expired reports whether the guard aborted the stream.

func (*IdleGuard) Read added in v1.17.0 

func (g *IdleGuard) Read(p []byte) (int, error)

Read passes through to the wrapped reader under the idle bound. A read returns as soon as any bytes arrive, so bounding each read individually is what makes this an idle timeout rather than a total one.

func (*IdleGuard) Stop added in v1.17.0 

func (g *IdleGuard) Stop()

Stop releases the guard. It is safe to call more than once, and must be called once the caller is done with the stream so no timer outlives it.

type Listener 

type Listener interface {
	// Accept returns new connections. It should be called in a loop.
	Accept(ctx context.Context) (Conn, error)
	// Addr returns the local network address that the server is listening on.
	Addr() net.Addr
	// Close closes the listener. Accept will return ErrListenerClosed as soon as
	// all connections in the accept queue have been accepted. Already established
	// (accepted) connections will be unaffected.
	Close() error
}

type Network 

type Network string
const NetworkPipe Network = "pipe"
const NetworkQUIC Network = "quic"

type PipeConn 

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

func (*PipeConn) AcceptStream 

func (pc *PipeConn) AcceptStream(ctx context.Context) (Stream, error)

func (*PipeConn) Close 

func (pc *PipeConn) Close() error

Close closes the connection for both endpoints: pending and future OpenStream/AcceptStream calls fail with ErrConnClosed. Streams already established are unaffected.

func (*PipeConn) Context 

func (pc *PipeConn) Context() context.Context

Context returns a context that is cancelled when either endpoint closes the connection, letting callers detect a dead connection without a stream operation.

func (*PipeConn) LocalAddr 

func (pc *PipeConn) LocalAddr() net.Addr

func (*PipeConn) OpenStream 

func (pc *PipeConn) OpenStream(ctx context.Context) (Stream, error)

func (*PipeConn) RemoteAddr 

func (pc *PipeConn) RemoteAddr() net.Addr

type PipeListener 

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

PipeListener is the transport's single listener, holding its registry entry.

func (*PipeListener) Accept 

func (pl *PipeListener) Accept(ctx context.Context) (Conn, error)

func (*PipeListener) Addr 

func (pl *PipeListener) Addr() net.Addr

func (*PipeListener) Close 

func (pl *PipeListener) Close() error

type PipeStream 

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

PipeStream is one end of a bidirectional in-memory stream: two unidirectional io.Pipes cross-wired between the endpoints. Writes rendezvous with peer reads, so backpressure is immediate.

func (*PipeStream) CancelRead 

func (ps *PipeStream) CancelRead(code StreamErrorCode)

CancelRead aborts the read side: the peer's writes and our own reads fail from now on. The code is surfaced to the peer as a StreamError.

func (*PipeStream) CancelWrite 

func (ps *PipeStream) CancelWrite(code StreamErrorCode)

CancelWrite aborts the write side: the peer's reads fail with a StreamError carrying the code instead of io.EOF.

func (*PipeStream) Close 

func (ps *PipeStream) Close() error

Close closes the write side: the peer observes io.EOF after draining buffered data. The read side stays open, matching QUIC stream semantics.

func (*PipeStream) LocalAddr 

func (ps *PipeStream) LocalAddr() net.Addr

func (*PipeStream) Read 

func (ps *PipeStream) Read(p []byte) (int, error)

func (*PipeStream) ReadFrom 

func (ps *PipeStream) ReadFrom(r io.Reader) (int64, error)

ReadFrom and WriteTo satisfy transport.Stream; io.Pipe has no zero-copy path, so both delegate to io.Copy on the underlying pipe half.

func (*PipeStream) RemoteAddr 

func (ps *PipeStream) RemoteAddr() net.Addr

func (*PipeStream) Write 

func (ps *PipeStream) Write(p []byte) (int, error)

func (*PipeStream) WriteTo 

func (ps *PipeStream) WriteTo(w io.Writer) (int64, error)

type PipeTransport 

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

PipeTransport connects endpoints within a single process: connections are channel-linked endpoint pairs and streams are in-memory pipes. It carries traffic between colocated nodes, where no network or auth is required.

It binds one address in a process-wide namespace, which it both dials from and listens on, so an accepted connection's remote is the dialer's source.

func NewPipeTransport 

func NewPipeTransport(addr string, port int) *PipeTransport

NewPipeTransport binds the in-process name addr:port. Port 0 takes a synthesized unique name, which yields a transport that dials but is never dialed.

func (*PipeTransport) Addr 

func (pt *PipeTransport) Addr() net.Addr

Addr is the name the transport bound, which differs from the requested one when the request named port 0.

func (*PipeTransport) Close 

func (pt *PipeTransport) Close() error

Close closes the transport and its listener. Established connections are unaffected.

func (*PipeTransport) Dial 

func (pt *PipeTransport) Dial(ctx context.Context, remote net.Addr) (Conn, error)

func (*PipeTransport) Listen 

func (pt *PipeTransport) Listen() (Listener, error)

Listen accepts connections on the transport's own name. One name serves one node, so it may be called once.

func (*PipeTransport) Network 

func (pt *PipeTransport) Network() string

type QUICConn 

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

QUICConn reports its endpoints as transport addresses rather than the UDP addresses underneath, so a connection's remote is comparable with the addresses a route names.

func (*QUICConn) AcceptStream 

func (qc *QUICConn) AcceptStream(ctx context.Context) (Stream, error)

func (*QUICConn) Close 

func (qc *QUICConn) Close() error

func (*QUICConn) Context 

func (qc *QUICConn) Context() context.Context

Context returns a context that is cancelled when the connection is closed, letting callers detect a dead connection without a stream operation.

func (*QUICConn) LocalAddr 

func (qc *QUICConn) LocalAddr() net.Addr

func (*QUICConn) OpenStream 

func (qc *QUICConn) OpenStream(ctx context.Context) (Stream, error)

func (*QUICConn) RemoteAddr 

func (qc *QUICConn) RemoteAddr() net.Addr

type QUICOption 

type QUICOption func(*QUICTransport)

func WithRootCAs 

func WithRootCAs(p *x509.CertPool) QUICOption

WithRootCAs verifies peer certificates against p when dialing. Unset means the OS trust store.

type QUICStream 

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

func (*QUICStream) CancelRead 

func (qs *QUICStream) CancelRead(code StreamErrorCode)

func (*QUICStream) CancelWrite 

func (qs *QUICStream) CancelWrite(code StreamErrorCode)

func (*QUICStream) Close 

func (qs *QUICStream) Close() error

Close closes the write side: the peer observes io.EOF after draining. The read side stays open.

func (*QUICStream) LocalAddr 

func (qs *QUICStream) LocalAddr() net.Addr

func (*QUICStream) Read 

func (qs *QUICStream) Read(p []byte) (int, error)

func (*QUICStream) ReadFrom 

func (qs *QUICStream) ReadFrom(r io.Reader) (int64, error)

func (*QUICStream) RemoteAddr 

func (qs *QUICStream) RemoteAddr() net.Addr

func (*QUICStream) Write 

func (qs *QUICStream) Write(p []byte) (int, error)

func (*QUICStream) WriteTo 

func (qs *QUICStream) WriteTo(w io.Writer) (int64, error)

type QUICTransport 

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

QUICTransport carries streams to other hosts over QUIC with TLS 1.3, one UDP socket per node. Intra-cluster auth is the server certificate: dialers verify it against RootCAs, and both sides pin the cluster ALPN.

func NewQUICDialTransport added in v1.17.0 

func NewQUICDialTransport(addr string, port int, opts ...QUICOption) (*QUICTransport, error)

NewQUICDialTransport binds a UDP socket at addr:port for dialing only, with no TLS identity of its own. Dial never presents a client certificate — the wire has nowhere for one, since nothing in this cluster requests one back — so a transport built to dial rather than listen needs none to load, and asking a caller with no stake in this cluster's PKI for one would be asking for something it has no business holding. Listen on it always fails.

func NewQUICTransport 

func NewQUICTransport(addr string, port int, cert, key string, opts ...QUICOption) (*QUICTransport, error)

NewQUICTransport binds a UDP socket at addr:port. Port 0 takes an OS-assigned port, which yields a transport that dials but is never dialed.

func (*QUICTransport) Addr 

func (qt *QUICTransport) Addr() net.Addr

Addr is the endpoint the socket bound, which differs from the requested one when the request named port 0.

func (*QUICTransport) Close 

func (qt *QUICTransport) Close() error

Close tears down the listener and the socket. Established connections are terminated abruptly, so callers drain them first.

func (*QUICTransport) Dial 

func (qt *QUICTransport) Dial(ctx context.Context, remote net.Addr) (Conn, error)

func (*QUICTransport) Listen 

func (qt *QUICTransport) Listen() (Listener, error)

Listen accepts connections on the transport's own socket. One socket serves one node, so it may be called once.

func (*QUICTransport) Network 

func (qt *QUICTransport) Network() string

type Stream 

type Stream interface {
	io.ReadWriteCloser
	io.ReaderFrom
	io.WriterTo
	CancelRead(code StreamErrorCode)
	CancelWrite(code StreamErrorCode)
	LocalAddr() net.Addr
	RemoteAddr() net.Addr
}

type StreamConn 

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

StreamConn adapts a Stream to net.Conn for consumers that require that contract, notably the hashicorp/raft network transport.

Deadlines are destructive: a stream cannot interrupt a blocked read or write and resume later, so an expired deadline aborts the affected direction and the connection is unusable afterwards. Consumers that discard connections on deadline errors — raft does — observe standard net.Conn behavior.

func NewStreamConn 

func NewStreamConn(s Stream) *StreamConn

func (*StreamConn) Close 

func (c *StreamConn) Close() error

Close aborts the read side and closes the write side so both directions terminate promptly, then signals Done.

func (*StreamConn) Done 

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

Done is closed when the connection is closed; it lets the goroutine that owns the underlying rpc stream hold it open for the connection's lifetime.

func (*StreamConn) LocalAddr 

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

func (*StreamConn) Read 

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

func (*StreamConn) RemoteAddr 

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

func (*StreamConn) SetDeadline 

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

func (*StreamConn) SetReadDeadline 

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

func (*StreamConn) SetWriteDeadline 

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

func (*StreamConn) Write 

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

type StreamError 

type StreamError struct {
	Code StreamErrorCode
}

StreamError reports that a stream was aborted with an application error code, either locally via CancelRead/CancelWrite or by the peer. Callers unwrap it from Read/Write errors to recover the code.

func (*StreamError) Error 

func (e *StreamError) Error() string

type StreamErrorCode 

type StreamErrorCode uint64
const StreamCodeIdle StreamErrorCode = 2

StreamCodeIdle is sent to the peer when an idle timeout aborts a stream.

type Transport 

type Transport interface {
	// Network names the network this transport serves, matching the Network()
	// of every address it accepts.
	Network() string
	Dial(ctx context.Context, remote net.Addr) (Conn, error)
	// Listen serves this transport's own endpoint. It may be called once.
	Listen() (Listener, error)
	// Addr is the endpoint the transport bound, which differs from the one
	// requested when the request named port 0.
	Addr() net.Addr
	Close() error
}

Transport carries streams over one network for one node: it owns a single endpoint, bound at construction, that it both dials from and listens on.

type UnknownNetworkError 

type UnknownNetworkError string

func (UnknownNetworkError) Error 

func (e UnknownNetworkError) Error() string

Source Files

View all Source files

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.