Documentation
¶
Overview ¶
Package transport pkg/cxo/node/transport/connection.go c2-net-cxo
Package transport pkg/cxo/node/transport/dmsg.go c2-net-cxo dmsg.go implements a CXO transport factory over DMSG connections, allowing CXO nodes to communicate via skywire's DMSG network instead of raw TCP.
Package transport pkg/cxo/node/transport/noise_tcp.go c2-net-cxo
Package transport pkg/cxo/node/transport/tcp.go c2-net-cxo
Package transport pkg/cxo/node/transport/udp.go c2-net-cxo
Index ¶
Constants ¶
const DefaultCXOPort = skyenv.DmsgCXOPort
DefaultCXOPort is the default DMSG port for CXO connections. Sourced from skyenv so the port allocation is visible in the same table as DmsgHypervisorPort (46), DmsgTransportSetupPort (47), etc.
Was previously the literal 46, which clashed with DmsgHypervisorPort: whichever module called dmsgC.Listen first won, and on a hypervisor- enabled visor the CXO publisher's Listen typically fired before the hypervisor's, so hypervisor-RPC-over-DMSG silently failed with "port already occupied" and remote visors stopped showing up.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Connection ¶
type Connection struct {
// contains filtered or unexported fields
}
Connection wraps a net.Conn with channel-based message I/O. Messages are length-prefixed (4 bytes, big-endian).
func (*Connection) Close ¶
func (c *Connection) Close()
Close closes the connection and its channels.
func (*Connection) GetChanIn ¶
func (c *Connection) GetChanIn() <-chan []byte
GetChanIn returns the receive channel.
func (*Connection) GetChanOut ¶
func (c *Connection) GetChanOut() chan<- Frame
GetChanOut returns the send channel.
func (*Connection) GetRemoteAddr ¶
func (c *Connection) GetRemoteAddr() net.Addr
GetRemoteAddr returns the remote address of the connection.
func (*Connection) IsClosed ¶
func (c *Connection) IsClosed() bool
IsClosed returns true if the connection has been closed.
func (*Connection) IsTCP ¶
func (c *Connection) IsTCP() bool
IsTCP returns true if this is a TCP connection.
type DMSGFactory ¶
type DMSGFactory struct {
AcceptedCallback func(conn *Connection)
// MaxPendingAccepts caps in-flight AcceptedCallback invocations.
// Zero (default) leaves accepts unbounded for backwards compatibility;
// set it from the Node's MaxPendingConnections to apply backpressure
// to the dmsg listener queue when handshake processing falls behind.
MaxPendingAccepts int
// contains filtered or unexported fields
}
DMSGFactory creates and manages CXO connections over DMSG. It implements the same interface as TCPFactory but uses DMSG streams as the underlying transport.
func NewDMSGFactory ¶
func NewDMSGFactory(dmsgC *dmsg.Client, port uint16) *DMSGFactory
NewDMSGFactory creates a new DMSGFactory using the given DMSG client. The port parameter specifies the DMSG port to listen on and dial to.
func (*DMSGFactory) Address ¶
func (f *DMSGFactory) Address() string
Address returns a string representation of the listening address (PK:port).
func (*DMSGFactory) Close ¶
func (f *DMSGFactory) Close()
Close stops the factory and closes the listener.
func (*DMSGFactory) Connect ¶
func (f *DMSGFactory) Connect(ctx context.Context, remotePK cipher.PubKey) (*Connection, error)
Connect dials a remote CXO node over DMSG by public key. ctx is passed straight to dmsg.Client.Dial — a canceled or expired ctx aborts the dial and returns ctx.Err() (wrapped). Callers that don't need cancellation can pass context.Background(); the (*DMSG).ConnectPK wrapper threads its caller's ctx through here.
func (*DMSGFactory) Listen ¶
func (f *DMSGFactory) Listen() error
Listen starts listening on the DMSG port for incoming CXO connections.
type Frame ¶
Frame is one outbound message split into a small per-message Head (the 8-byte seq/rseq the CXO Conn prepends) and a Body (the encoded message). Splitting them lets a broadcast encode the (potentially large) Body ONCE and share that immutable []byte across every subscriber connection — each conn supplies only its own tiny Head. On the wire the frame is identical to the old single-buffer form: [4-byte length][Head][Body]. Sharing the Body is what turns a fan-out of an N-MB feed to M subscribers from O(N*M) memory + encode CPU (one encoded+buffered copy per conn) into O(N + M). See docs — the transport-discovery CXO node ballooned to multi-GB RSS re-encoding the all-transports feed once per subscriber.
type TCPFactory ¶
type TCPFactory struct {
AcceptedCallback func(conn *Connection)
// MaxPendingAccepts caps in-flight AcceptedCallback invocations.
// Zero (default) leaves accepts unbounded for backwards compatibility;
// set it from the Node's MaxPendingConnections to apply backpressure
// to the listener queue when handshake processing falls behind.
MaxPendingAccepts int
// contains filtered or unexported fields
}
TCPFactory creates and manages TCP connections. Every connection (dial + accept) is Noise_XX-encrypted via wrapNoiseXX using the factory's identity, so CXO-TCP is no longer a plaintext transport.
func NewTCPFactory ¶
func NewTCPFactory(localPK cipher.PubKey, localSK cipher.SecKey) *TCPFactory
NewTCPFactory creates a new TCPFactory bound to the given identity, used for the Noise XX handshake on every TCP connection.
func (*TCPFactory) Close ¶
func (f *TCPFactory) Close()
Close stops the factory and closes the listener.
func (*TCPFactory) Connect ¶
func (f *TCPFactory) Connect(address string) (*Connection, error)
Connect dials a TCP address and returns a Connection.
func (*TCPFactory) Listen ¶
func (f *TCPFactory) Listen(address string) error
Listen starts listening on the given address for incoming TCP connections.
func (*TCPFactory) ListenerAddress ¶
func (f *TCPFactory) ListenerAddress() string
ListenerAddress returns the actual address the listener is bound to. Useful when listening on ":0" to get the OS-assigned port.
type UDPFactory ¶
type UDPFactory struct {
AcceptedCallback func(conn *Connection)
// MaxPendingAccepts caps in-flight AcceptedCallback invocations.
// Zero (default) leaves accepts unbounded for backwards compatibility;
// set it from the Node's MaxPendingConnections to apply backpressure
// to the listener queue when handshake processing falls behind.
MaxPendingAccepts int
// contains filtered or unexported fields
}
UDPFactory creates and manages UDP connections. Currently implemented over TCP for reliable delivery. A future version may use a true reliable UDP protocol.
func (*UDPFactory) Close ¶
func (f *UDPFactory) Close()
Close stops the factory and closes the listener.
func (*UDPFactory) Connect ¶
func (f *UDPFactory) Connect(address string) (*Connection, error)
Connect dials the given address and returns a Connection.
func (*UDPFactory) Listen ¶
func (f *UDPFactory) Listen(address string) error
Listen starts listening on the given address.
Source Files