proxy

Documentation

Overview

internal/gateway/server.go

Index

• func CreateTransportConfig() *quic.Config
• func LoadTLSConfig(certPath, keyPath, caCertPath string) (*tls.Config, error)
• type AddTopicRequest
• type ConnectionPair
  • func (p *ConnectionPair) Close()
  • func (p *ConnectionPair) SetInbound(ctx context.Context, conn *quic.Conn, toClusterCh chan<- model.Command, ...)
  • func (p *ConnectionPair) SetOutbound(ctx context.Context, conn *quic.Conn)
• type ConnectionType
• type Gateway
  • func NewGateway(selfNodeID string, certPath string, keyPath string, caCertPath string, ...) (*Gateway, error)
  • func (g *Gateway) Addr() net.Addr
  • func (g *Gateway) Close() error
  • func (a *Gateway) GetLeaderID() string
  • func (g *Gateway) Run(ctx context.Context) error
• type Handshake
• type HandshakeResponse
• type ProxyRegistry
  • func NewProxyRegistry(ctx context.Context, logger *zap.Logger, metrics *internal.GatewayMetrics, ...) *ProxyRegistry
  • func (r *ProxyRegistry) Close()
  • func (r *ProxyRegistry) Get(proxyID string) (*ConnectionPair, bool)
  • func (r *ProxyRegistry) GetOrCreate(proxyID string, status *atomic.Uint32, leaderID *atomic.Value, ...) *ConnectionPair
  • func (r *ProxyRegistry) Remove(proxyID string)
• type ProxyTopologyUpdate
• type ProxyTopologyUpdateType

Functions

func CreateTransportConfig

func CreateTransportConfig() *quic.Config

CreateTransportConfig creates the QUIC transport configuration

func LoadTLSConfig

func LoadTLSConfig(certPath, keyPath, caCertPath string) (*tls.Config, error)

LoadTLSConfig loads and configures TLS with client CA verification

Types

type AddTopicRequest

type AddTopicRequest struct {
	RequestID string `msgpack:"request_id"`
	Topic     string `msgpack:"topic"`
}

type ConnectionPair

type ConnectionPair struct {
	ProxyID string
	// contains filtered or unexported fields
}

ConnectionPair holds both connections for a single proxy

func (*ConnectionPair) Close

func (p *ConnectionPair) Close()

Close shuts down both connections and cancels handlers

func (*ConnectionPair) SetInbound

func (p *ConnectionPair) SetInbound(ctx context.Context,
	conn *quic.Conn,
	toClusterCh chan<- model.Command,
	topicCmdCh chan<- state.TopicCommand,
	hbRouter *state.HeartbeatRouter)

SetInbound registers an inbound connection and starts its handler

func (*ConnectionPair) SetOutbound

func (p *ConnectionPair) SetOutbound(ctx context.Context, conn *quic.Conn)

SetOutbound registers an outbound connection and starts its handler

type ConnectionType

type ConnectionType string

const (
	ConnectionTypeInbound  ConnectionType = "inbound"
	ConnectionTypeOutbound ConnectionType = "outbound"
)

type Gateway

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

Gateway manages QUIC connections from proxies

func NewGateway

func NewGateway(
	selfNodeID string,
	certPath string,
	keyPath string,
	caCertPath string,
	listenAddr string,
	logger *zap.Logger,
	topicCmdCh chan state.TopicCommand,
	hbRouter *state.HeartbeatRouter,
	toClusterCh chan<- model.Command,
	status *atomic.Uint32,
	currentTerm *atomic.Uint64,
	members *model.Members,
	leaderID *atomic.Value,
) (*Gateway, error)

NewGateway creates a new Gateway instance

func (*Gateway) Addr

func (g *Gateway) Addr() net.Addr

func (*Gateway) Close

func (g *Gateway) Close() error

Close gracefully shuts down the gateway

func (*Gateway) GetLeaderID

func (a *Gateway) GetLeaderID() string

func (*Gateway) Run

func (g *Gateway) Run(ctx context.Context) error

Run handles incoming QUIC connections from proxies

type Handshake

type Handshake struct {
	ProxyID         string         `msgpack:"proxy_id"`
	ConnectionType  ConnectionType `msgpack:"connection_type"`
	ProtocolVersion int            `msgpack:"protocol_version"`
}

Handshake represents the initial handshake message from proxy

type HandshakeResponse

type HandshakeResponse struct {
	Status  string `msgpack:"status"`
	Message string `msgpack:"message,omitempty"`
	NodeID  string `msgpack:"node_id"`
}

HandshakeResponse sent back to proxy

type ProxyRegistry

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

ProxyRegistry manages all connected proxies

func NewProxyRegistry

func NewProxyRegistry(ctx context.Context, logger *zap.Logger, metrics *internal.GatewayMetrics, cmdCh chan<- state.TopicCommand) *ProxyRegistry

NewProxyRegistry creates a new registry

func (*ProxyRegistry) Close

func (r *ProxyRegistry) Close()

Close shuts down all proxy connections

func (*ProxyRegistry) Get

func (r *ProxyRegistry) Get(proxyID string) (*ConnectionPair, bool)

func (*ProxyRegistry) GetOrCreate

func (r *ProxyRegistry) GetOrCreate(proxyID string, status *atomic.Uint32, leaderID *atomic.Value, currentTerm *atomic.Uint64, members *model.Members) *ConnectionPair

GetOrCreate returns existing pair or creates a new one

func (*ProxyRegistry) Remove

func (r *ProxyRegistry) Remove(proxyID string)

Remove deletes a proxy pair from registry

type ProxyTopologyUpdate

type ProxyTopologyUpdate struct {
	Type    ProxyTopologyUpdateType      `msgpack:"type"` // "add" or "remove"
	ProxyID string                       `msgpack:"proxy_id"`
	PushCh  chan model.ToConsumerMessage `msgpack:"-"` // not sent over wire
}

ProxyTopologyUpdate is sent when proxy connects or disconnects

type ProxyTopologyUpdateType

type ProxyTopologyUpdateType string

const (
	ProxyAdd    ProxyTopologyUpdateType = "add"
	ProxyRemove ProxyTopologyUpdateType = "remove"
)