network

package
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Feb 7, 2026 License: MIT Imports: 23 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	MessageTypeUnknown MessageType = iota

	// Core network message types
	MessageTypeRequest
	MessageTypeResponse
	MessageTypeOneWay
	MessageTypeHeartbeat
	MessageTypeError

	// Cluster-specific message types (start from 10 to avoid conflicts)
	MessageTypePing           = 10
	MessageTypeConnect        = 11
	MessageTypeLeave          = 12
	MessageTypeGossipPush     = 20
	MessageTypeGossipPull     = 21
	MessageTypeGossipResponse = 22
	MessageTypeReadRequest    = 30 // Unified with ClusterMessageTypes.ReadRequest
	MessageTypeReadResponse   = 31 // Unified with ClusterMessageTypes.ReadResponse
	MessageTypeSyncOperation  = 40
)

Message type constants - unified definition for all layers

Variables

View Source 
var (
	// ErrConnClosed indicates connection is closed
	ErrConnClosed = errors.New("connection closed")

	// ErrConnTimeout indicates connection timeout
	ErrConnTimeout = errors.New("connection timeout")

	// ErrPoolExhausted indicates connection pool exhausted
	ErrPoolExhausted = errors.New("connection pool exhausted")

	// ErrPoolClosed indicates connection pool is closed
	ErrPoolClosed = errors.New("connection pool closed")

	// ErrMessageTooLarge indicates message is too large
	ErrMessageTooLarge = errors.New("message too large")

	// ErrBackpressure indicates backpressure is active
	ErrBackpressure = errors.New("backpressure active")

	// ErrTransportNotSupported indicates transport not supported
	ErrTransportNotSupported = errors.New("transport not supported")

	// ErrHandlerNotFound indicates message handler not found
	ErrHandlerNotFound = errors.New("handler not found")

	// ErrInvalidMessage indicates invalid message format
	ErrInvalidMessage = errors.New("invalid message")
)
View Source 
var ClusterMessageTypes = struct {
	Ping           MessageType
	Connect        MessageType
	Leave          MessageType
	GossipPush     MessageType
	GossipPull     MessageType
	GossipResponse MessageType
	ReadRequest    MessageType
	ReadResponse   MessageType
	SyncOperation  MessageType
}{
	Ping:           10,
	Connect:        11,
	Leave:          12,
	GossipPush:     20,
	GossipPull:     21,
	GossipResponse: 22,
	ReadRequest:    30,
	ReadResponse:   31,
	SyncOperation:  40,
}

ClusterMessageTypes defines message types used by cluster

Functions

func DecodeAny added in v0.3.0 

func DecodeAny(data []byte, target interface{}) error

DecodeAny decodes bytes to target type (simple binary format)

func EncodeAny added in v0.3.0 

func EncodeAny(data interface{}) ([]byte, error)

EncodeAny encodes any data to bytes (simple binary format for cluster messages)

func EncodeMessage added in v0.3.0 

func EncodeMessage(msg *Message) ([]byte, error)

EncodeMessage encodes message to bytes (public for cluster usage)

func NewBackpressure added in v0.3.0 

func NewBackpressure(cfg BackpressureConfig) *simpleBackpressure

func NewRouter added in v0.3.0 

func NewRouter() *simpleRouter

func ResetStats added in v0.3.4 

func ResetStats()

ResetStats resets all stats (for testing)

Types

type AdaptiveCfg added in v0.3.4 

type AdaptiveCfg struct {
	MinSize        int
	MaxSize        int
	InitialSize    int
	TargetWaitTime time.Duration
	HighThreshold  time.Duration
	LowThreshold   time.Duration
	IncreaseStep   int
	DecreaseStep   int
	CooldownPeriod time.Duration
	EMAAlpha       float64
}

func DefaultAdaptive added in v0.3.4 

func DefaultAdaptive() AdaptiveCfg

type BackpressureConfig added in v0.3.0 

type BackpressureConfig struct {
	// Threshold is backpressure threshold
	Threshold int

	// MaxCapacity is maximum capacity
	MaxCapacity int

	// Strategy is backpressure strategy
	Strategy BackpressureStrategy
}

BackpressureConfig configures backpressure

func DefaultBackpressureConfig added in v0.3.0 

func DefaultBackpressureConfig() BackpressureConfig

DefaultBackpressureConfig returns default backpressure config

type BackpressureStats added in v0.3.4 

type BackpressureStats struct {
	// Queued is number of queued operations
	Queued int

	// Capacity is maximum capacity
	Capacity int

	// Blocked indicates if operations are blocked
	Blocked bool

	// Rejected is number of rejected operations
	Rejected uint64
}

BackpressureStats represents backpressure statistics

type BackpressureStatus added in v0.3.0 

type BackpressureStatus = BackpressureStats

BackpressureStatus is deprecated, use BackpressureStats instead.

type BackpressureStrategy added in v0.3.0 

type BackpressureStrategy string

BackpressureStrategy is backpressure strategy 

const (
	// StrategyReject rejects new operations when threshold exceeded
	StrategyReject BackpressureStrategy = "reject"

	// StrategyBlock blocks new operations when threshold exceeded
	StrategyBlock BackpressureStrategy = "block"

	// StrategyDrop drops oldest operations when threshold exceeded
	StrategyDrop BackpressureStrategy = "drop"
)

type Client added in v0.3.0 

type Client interface {
	// Send sends message to address
	Send(ctx context.Context, address string, data []byte) error

	// SendWithTimeout sends message with timeout
	SendWithTimeout(ctx context.Context, address string, data []byte, timeout time.Duration) error

	// Request sends request and waits for response
	Request(ctx context.Context, address string, request []byte, timeout time.Duration) ([]byte, error)

	// Broadcast sends message to multiple addresses
	Broadcast(ctx context.Context, addresses []string, data []byte) error

	// Close closes client
	Close() error
}

Client provides high-level network client interface

func NewClient added in v0.3.0 

func NewClient(cfg ClientConfig) Client

type ClientConfig added in v0.3.0 

type ClientConfig struct {
	// Pool is connection pool
	Pool ConnPool

	// DefaultTimeout is default operation timeout (increased for high-load scenarios)
	DefaultTimeout time.Duration

	// RetryCount is retry count on failure
	RetryCount int

	// RetryBackoff is retry backoff duration
	RetryBackoff time.Duration

	// EnableCompression enables message compression
	EnableCompression bool

	// MaxRetries is maximum retry attempts
	MaxRetries int
}

ClientConfig configures client

func DefaultClientConfig added in v0.3.0 

func DefaultClientConfig(pool ConnPool) ClientConfig

DefaultClientConfig returns default client config for cluster communication

type Conn added in v0.3.0 

type Conn interface {
	// Send sends data to remote endpoint
	Send(ctx context.Context, data []byte) error

	// Receive receives data from remote endpoint
	Receive(ctx context.Context) ([]byte, error)

	// RemoteAddr returns remote address
	RemoteAddr() string

	// SetReadDeadline sets read deadline for connection
	SetReadDeadline(t time.Time) error

	// Close closes connection
	Close() error
}

Conn represents a network connection

type ConnPool added in v0.3.0 

type ConnPool interface {
	Get(ctx context.Context, address string) (Conn, error)
	Put(conn Conn)
	Remove(conn Conn)
	Close() error
	Stats() PoolStats
	DebugStats() PoolDebugStatsSnapshot
}

func NewConnPool added in v0.3.0 

func NewConnPool(cfg PoolConfig) ConnPool

type Counter added in v0.3.4 

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

Counter is a thread-safe counter using atomic operations

func (*Counter) Add added in v0.3.4 

func (c *Counter) Add(delta uint64)

Add adds delta to the counter

func (*Counter) Inc added in v0.3.4 

func (c *Counter) Inc()

Inc increments the counter by 1

func (*Counter) Load added in v0.3.4 

func (c *Counter) Load() uint64

Load returns the current counter value

func (*Counter) Store added in v0.3.4 

func (c *Counter) Store(value uint64)

Store sets the counter value

type Gauge added in v0.3.4 

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

Gauge is a thread-safe gauge using atomic operations

func (*Gauge) Add added in v0.3.4 

func (g *Gauge) Add(delta int64)

Add adds delta to the gauge

func (*Gauge) Dec added in v0.3.4 

func (g *Gauge) Dec()

Dec decrements the gauge by 1

func (*Gauge) Inc added in v0.3.4 

func (g *Gauge) Inc()

Inc increments the gauge by 1

func (*Gauge) Load added in v0.3.4 

func (g *Gauge) Load() int64

Load returns the current gauge value

func (*Gauge) Store added in v0.3.4 

func (g *Gauge) Store(value int64)

Store sets the gauge value

type Handler added in v0.3.0 

type Handler func(ctx context.Context, remoteAddr string, data []byte) ([]byte, error)

Handler handles incoming messages

type Listener added in v0.3.0 

type Listener interface {
	// Accept accepts incoming connection
	Accept(ctx context.Context) (Conn, error)

	// Address returns listening address
	Address() string

	// Close closes listener
	Close() error
}

Listener listens for incoming connections

type Message added in v0.3.0 

type Message struct {
	// Type is message type
	Type MessageType

	// ID is message ID (for request-response correlation)
	ID uint64

	// Data is message payload
	Data []byte

	// Timestamp is message timestamp
	Timestamp int64

	// Compressed indicates if data is compressed
	Compressed bool
}

Message represents network message

func DecodeMessage added in v0.3.0 

func DecodeMessage(data []byte) (*Message, error)

DecodeMessage decodes bytes to message (public for cluster usage)

type MessageType added in v0.3.0 

type MessageType uint8

MessageType is message type

type Metrics deprecated added in v0.3.4 

type Metrics = NetworkStats

Deprecated: Use NetworkStats instead

type Network added in v0.3.0 

type Network interface {
	// Client returns network client
	Client() Client

	// Server returns network server
	Server() Server

	// Start starts network layer
	Start(ctx context.Context) error

	// Stop stops network layer
	Stop(ctx context.Context) error

	// Name returns component name for lifecycle management
	Name() string

	// Close stops network layer (lifecycle.Component interface)
	Close(ctx context.Context) error

	// Send sends message to address
	Send(ctx context.Context, address string, data []byte) error

	// SendMessage sends typed message
	SendMessage(ctx context.Context, address string, msg *Message) error

	// Request sends request and waits for response
	Request(ctx context.Context, address string, request []byte, timeout time.Duration) ([]byte, error)

	// RequestMessage sends typed message request and waits for typed response
	RequestMessage(ctx context.Context, address string, request *Message, timeout time.Duration) (*Message, error)

	// RegisterHandler registers message handler
	RegisterHandler(msgType MessageType, handler Handler) error

	// Cluster adapter methods
	// SendFunc returns send function for cluster (MemberMgr)
	SendFunc() func(address string, msg interface{}) error
	// SendBytesFunc returns send bytes function for gossip
	SendBytesFunc() func(address string, data []byte) error
	// GetFunc returns get function for remote reads (Reader)
	GetFunc() func(nodeID string, key string) (interface{}, error)
	// ReceiveFunc returns receive function for gossip
	ReceiveFunc() func() ([]byte, error)
	// RegisterMessageHandler registers handler for specific message type
	RegisterMessageHandler(msgType MessageType, handler Handler) error

	// GetPool returns connection pool for metrics
	GetPool() ConnPool
}

Network provides unified network interface for cluster operations Implements lifecycle.Component for unified resource management

func NewNetwork added in v0.3.0 

func NewNetwork(cfg NetworkConfig) (Network, error)

NewNetwork builds a full stack using provided config.

type NetworkConfig added in v0.3.0 

type NetworkConfig struct {
	// LocalAddress is local listening address
	LocalAddress string

	// TransportType is transport protocol type
	TransportType TransportType

	// TransportConfig is transport configuration
	TransportConfig TransportConfig

	// PoolConfig is connection pool configuration
	PoolConfig PoolConfig

	// ClientConfig is client configuration
	ClientConfig ClientConfig

	// ServerConfig is server configuration
	ServerConfig ServerConfig

	// BackpressureConfig is backpressure configuration
	BackpressureConfig BackpressureConfig

	// EnableMetrics enables metrics collection
	EnableMetrics bool
}

NetworkConfig configures network layer

func DefaultNetworkConfig added in v0.3.0 

func DefaultNetworkConfig(localAddress string) NetworkConfig

DefaultNetworkConfig returns default network config

type NetworkSnapshot added in v0.3.4 

type NetworkSnapshot struct {
	ServerConnections uint64
	ServerMessages    uint64
	ServerBytes       uint64
	ServerErrors      uint64
	ServerActiveConns int64
	PoolTotal         int64
	PoolActive        int64
	PoolIdle          int64
	PoolWaiters       int64
	PoolCreated       uint64
	PoolClosed        uint64
	PoolErrors        uint64
	ClientRequests    uint64
	ClientResponses   uint64
	ClientErrors      uint64
	ClientBytes       uint64
}

NetworkSnapshot represents aggregated network stats snapshot

type NetworkStats added in v0.3.4 

type NetworkStats struct {
	Server Stats
	Pool   Stats
	Client Stats
}

NetworkStats aggregates stats from Server, Pool, and Client components

func GetMetrics deprecated added in v0.3.4 

func GetMetrics() *NetworkStats

Deprecated: Use GetStats instead

func GetStats added in v0.3.4 

func GetStats() *NetworkStats

GetStats returns global network stats

func (*NetworkStats) Snapshot added in v0.3.4 

func (ns *NetworkStats) Snapshot() NetworkSnapshot

Snapshot returns a snapshot of current network stats

type NetworkStatsSnapshot deprecated added in v0.3.4 

type NetworkStatsSnapshot = NetworkSnapshot

Deprecated: Use NetworkSnapshot instead

type PoolConfig added in v0.3.0 

type PoolConfig struct {
	MaxIdle         int
	MaxActive       int
	IdleTimeout     time.Duration
	MaxLifetime     time.Duration
	WaitTimeout     time.Duration
	CleanupInterval time.Duration
	Transport       Transport
}

func DefaultPoolConfig added in v0.3.0 

func DefaultPoolConfig(transport Transport) PoolConfig

type PoolDebugStats added in v0.3.4 

type PoolDebugStats struct {
	GetAttempts      atomic.Uint64
	GetSuccess       atomic.Uint64
	GetExhausted     atomic.Uint64
	GetTimeout       atomic.Uint64
	GetContextCancel atomic.Uint64
	GetDialError     atomic.Uint64

	PutAttempts atomic.Uint64
	PutSuccess  atomic.Uint64
	PutClosed   atomic.Uint64

	WaitQueueLength atomic.Uint64
	MaxWaitQueueLen atomic.Uint64
	TotalWaitTime   atomic.Uint64
	WaitSamples     atomic.Uint64

	ActiveConnPeak atomic.Int64
	IdleConnPeak   atomic.Int64
}

type PoolDebugStatsSnapshot added in v0.4.0 

type PoolDebugStatsSnapshot struct {
	GetAttempts      uint64
	GetSuccess       uint64
	GetExhausted     uint64
	GetTimeout       uint64
	GetContextCancel uint64
	GetDialError     uint64

	PutAttempts uint64
	PutSuccess  uint64
	PutClosed   uint64

	WaitQueueLength uint64
	MaxWaitQueueLen uint64
	TotalWaitTime   uint64
	WaitSamples     uint64

	ActiveConnPeak int64
	IdleConnPeak   int64
}

type PoolMetrics added in v0.3.4 

type PoolMetrics = PoolStats

PoolMetrics is deprecated, use PoolStats instead.

type PoolStats added in v0.3.0 

type PoolStats struct {
	// Connection stats
	Total   int64
	Active  int64
	Idle    int64
	Waiters int64
	Created uint64
	Closed  uint64
	Errors  uint64

	// Performance stats
	AvgWaitTime time.Duration
	MaxWaitTime time.Duration
	AvgHoldTime time.Duration
	RequestRate float64
	WaitSamples uint64
	HoldSamples uint64
}

type RequestHandler added in v0.3.0 

type RequestHandler func(ctx context.Context, remoteAddr string, request []byte) ([]byte, error)

RequestHandler handles request-response pattern

type Server added in v0.3.0 

type Server interface {
	// Start starts server
	Start(ctx context.Context, address string, handler Handler) error

	// StartRequestResponse starts server with request-response handler
	StartRequestResponse(ctx context.Context, address string, handler RequestHandler) error

	// Stop stops server
	Stop(ctx context.Context) error

	// Address returns server address
	Address() string

	// Stats returns server statistics
	Stats() ServerStats
}

Server provides network server interface

func NewServer added in v0.3.0 

func NewServer(cfg ServerConfig) Server

type ServerConfig added in v0.3.0 

type ServerConfig struct {
	// Transport is underlying transport
	Transport Transport

	// MaxConns is maximum concurrent connections
	MaxConns int

	// ReadBufferSize is read buffer size
	ReadBufferSize int

	// WriteBufferSize is write buffer size
	WriteBufferSize int

	// EnableRequestResponse enables request-response pattern
	EnableRequestResponse bool

	// WorkerPoolSize is worker pool size for message handling
	WorkerPoolSize int

	// EnableBackpressure enables backpressure control
	EnableBackpressure bool

	// BackpressureThreshold is backpressure threshold (queued messages)
	BackpressureThreshold int
}

ServerConfig configures server

func DefaultServerConfig added in v0.3.0 

func DefaultServerConfig(transport Transport) ServerConfig

DefaultServerConfig returns default server config

type ServerStats added in v0.3.0 

type ServerStats struct {
	Connections uint64 // Total connections (atomic)
	Messages    uint64 // Total messages (received + sent) (atomic)
	Bytes       uint64 // Total bytes (received + sent) (atomic)
	Errors      uint64 // Errors (atomic)
	ActiveConns int64  // Active connections (atomic)
}

ServerStats represents server statistics (simplified, key metrics only)

type Stats added in v0.3.4 

type Stats struct {
	// Counters
	Requests Counter
	Success  Counter
	Errors   Counter
	Bytes    Counter
	Messages Counter

	// Gauges
	Active    Gauge
	Idle      Gauge
	Waiters   Gauge
	QueueSize Gauge
}

Stats provides a collection of common stats

func (*Stats) Reset added in v0.3.4 

func (s *Stats) Reset()

Reset resets all stats to zero (for testing)

func (*Stats) Snapshot added in v0.3.4 

func (s *Stats) Snapshot() StatsSnapshot

Snapshot returns a snapshot of current stats

type StatsSnapshot added in v0.3.4 

type StatsSnapshot struct {
	Requests  uint64
	Success   uint64
	Errors    uint64
	Bytes     uint64
	Messages  uint64
	Active    int64
	Idle      int64
	Waiters   int64
	QueueSize int64
}

StatsSnapshot represents a point-in-time snapshot of stats

type Transport added in v0.3.0 

type Transport interface {
	// Dial creates connection to remote address
	Dial(ctx context.Context, address string) (Conn, error)

	// Listen starts listening on address
	Listen(ctx context.Context, address string) (Listener, error)

	// Close closes transport
	Close() error
}

Transport provides network transport abstraction

func NewQUICTransport added in v0.3.0 

func NewQUICTransport(cfg TransportConfig) Transport

func NewTCPTransport added in v0.3.0 

func NewTCPTransport(cfg TransportConfig) Transport

func NewTransport added in v0.3.0 

func NewTransport(cfg TransportConfig) (Transport, error)

Transport constructors

type TransportConfig added in v0.3.0 

type TransportConfig struct {
	// Type is transport protocol type
	Type TransportType

	// Timeout is connection timeout
	Timeout time.Duration

	// ReadTimeout is read operation timeout
	ReadTimeout time.Duration

	// WriteTimeout is write operation timeout
	WriteTimeout time.Duration

	// KeepAlive enables keep-alive
	KeepAlive bool

	// KeepAliveInterval is keep-alive interval
	KeepAliveInterval time.Duration

	// MaxMessageSize is maximum message size (bytes)
	MaxMessageSize int

	// EnableZeroCopy enables zero-copy
	EnableZeroCopy bool
}

TransportConfig configures transport

func DefaultTransportConfig added in v0.3.0 

func DefaultTransportConfig() TransportConfig

DefaultTransportConfig returns default transport config

type TransportType added in v0.3.0 

type TransportType string

TransportType specifies transport protocol 

const (
	TransportTCP  TransportType = "tcp"
	TransportQUIC TransportType = "quic"
)

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.