server

Documentation

Overview

Package server implements the rmtt protocol server library.

A server accepts device connections over one or more transports (tcp, kcp, tls, quic, ws, wss) through Listener implementations, authenticates devices with an Authenticator, routes uplink PUSH messages to a MessageHandler and pushes downlink messages with Server.Push. Keepalive is negotiated per connection through KeepalivePolicy, and connection lifecycle events are reported through ConnectionListener.

Usage: build a ServerOptions, register listeners with AddListener and install callbacks, then create the server with NewServer and serve with ListenAndServe. When no listener is added, a single TCP listener on options.Port is used.

The package is silent by default; install loggers with SetLogger or the per-level setters (SetErrorLogger/SetInfoLogger/SetWarnLogger/SetDebugLogger).

Index

• func SetDebugLogger(logger Logger)
• func SetErrorLogger(logger Logger)
• func SetInfoLogger(logger Logger)
• func SetLogger(logger Logger)
• func SetWarnLogger(logger Logger)
• type Authenticator
• type ConnectionListener
• type ConnectionStore
  • func NewConnectionStore() *ConnectionStore
  • func (s *ConnectionStore) All() []DeviceConnection
  • func (s *ConnectionStore) Get(deviceID string) (DeviceConnection, bool)
  • func (s *ConnectionStore) Register(deviceID string, conn DeviceConnection) (prev DeviceConnection)
  • func (s *ConnectionStore) Remove(deviceID string, conn DeviceConnection) bool
• type DeviceConnection
• type KeepalivePolicy
  • func DefaultKeepalivePolicy() *KeepalivePolicy
  • func (p *KeepalivePolicy) Decide(clientKp int64) int64
• type Listener
  • func NewKCPListener(addr string) Listener
  • func NewQUICListener(addr string, config *tls.Config) Listener
  • func NewQUICListenerWithConfig(addr string, config *tls.Config, quicCfg *quic.Config) Listener
  • func NewTCPListener(addr string) Listener
  • func NewTLSListener(addr string, config *tls.Config) Listener
  • func NewWSListener(addr, path string) Listener
  • func NewWSSListener(addr, path string, config *tls.Config) Listener
• type Logger
• type MessageHandler
• type NOOPLogger
  • func (NOOPLogger) Printf(format string, v ...interface{})
  • func (NOOPLogger) Println(v ...interface{})
• type Server
  • func NewServer(opts *ServerOptions) Server
• type ServerOptions
  • func NewServerOptions() *ServerOptions
  • func (o *ServerOptions) AddListener(l Listener) *ServerOptions
  • func (o *ServerOptions) SetAuthenticator(auth Authenticator) *ServerOptions
  • func (o *ServerOptions) SetConnectionListener(listener ConnectionListener) *ServerOptions
  • func (o *ServerOptions) SetKeepalivePolicy(policy *KeepalivePolicy) *ServerOptions
  • func (o *ServerOptions) SetMessageHandler(handler MessageHandler) *ServerOptions
  • func (o *ServerOptions) SetPort(port int) *ServerOptions

Functions

func SetDebugLogger

func SetDebugLogger(logger Logger)

SetDebugLogger sets the Logger used for debug-level messages only.

func SetErrorLogger

func SetErrorLogger(logger Logger)

SetErrorLogger sets the Logger used for error-level messages only.

func SetInfoLogger

func SetInfoLogger(logger Logger)

SetInfoLogger sets the Logger used for info-level messages only.

func SetLogger

func SetLogger(logger Logger)

SetLogger sets a single Logger implementation for all levels (error/info/warn/debug).

func SetWarnLogger

func SetWarnLogger(logger Logger)

SetWarnLogger sets the Logger used for warn-level messages only.

Types

type Authenticator

type Authenticator interface {
	Authenticate(credential string) (deviceID string, allowed bool)
}

Authenticator authenticates a CONNECT credential and maps it to a device ID; allowed=false rejects the connection.

type ConnectionListener

type ConnectionListener interface {
	OnConnectionEstablished(deviceID string)
	OnConnectionClosed(deviceID string, reason string)
}

ConnectionListener receives device connection lifecycle events.

type ConnectionStore

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

ConnectionStore tracks the active device connections by device ID. It is safe for concurrent use.

func NewConnectionStore

func NewConnectionStore() *ConnectionStore

NewConnectionStore returns an empty connection store.

func (*ConnectionStore) All

func (s *ConnectionStore) All() []DeviceConnection

All returns every registered connection.

func (*ConnectionStore) Get

func (s *ConnectionStore) Get(deviceID string) (DeviceConnection, bool)

Get returns the connection registered for deviceID.

func (*ConnectionStore) Register

func (s *ConnectionStore) Register(deviceID string, conn DeviceConnection) (prev DeviceConnection)

Register maps deviceID to conn, returning the previous connection for that ID (nil when none).

func (*ConnectionStore) Remove

func (s *ConnectionStore) Remove(deviceID string, conn DeviceConnection) bool

Remove deletes the mapping for deviceID if and only if the currently registered connection is conn.

type DeviceConnection

type DeviceConnection interface {
	DeviceID() string
	IsActive() bool
	Write(payload []byte) error
	SendDisconnect(reason byte)
	Close()
}

DeviceConnection is a single authenticated device connection.

type KeepalivePolicy

type KeepalivePolicy struct {
	MinSeconds     int64
	MaxSeconds     int64
	DefaultSeconds int64
	AllowDisable   bool
}

KeepalivePolicy clamps the client's Keepalive proposal into a server-side range. A non-positive proposal maps to DefaultSeconds unless AllowDisable is set (which returns 0 and disables server-side keepalive enforcement).

func DefaultKeepalivePolicy

func DefaultKeepalivePolicy() *KeepalivePolicy

DefaultKeepalivePolicy returns the default policy: client proposals are clamped into [30, 600] seconds, 60s fallback, keepalive cannot be disabled.

func (*KeepalivePolicy) Decide

func (p *KeepalivePolicy) Decide(clientKp int64) int64

Decide maps the client's Keepalive proposal (seconds) to the server-side keepalive echoed in CONNACK.

type Listener

type Listener interface {
	Serve(ctx context.Context, handler func(net.Conn)) error
	Close() error
}

Listener abstracts a transport acceptor. Serve blocks until ctx is cancelled or an irrecoverable error occurs, invoking handler for each accepted connection.

func NewKCPListener

func NewKCPListener(addr string) Listener

NewKCPListener returns a KCP (reliable UDP) listener bound to addr (e.g. ":18883").

func NewQUICListener

func NewQUICListener(addr string, config *tls.Config) Listener

NewQUICListener returns a QUIC (UDP + TLS 1.3) listener bound to addr (e.g. ":18885"), using the library's hardened transport defaults.

func NewQUICListenerWithConfig

func NewQUICListenerWithConfig(addr string, config *tls.Config, quicCfg *quic.Config) Listener

NewQUICListenerWithConfig returns a QUIC listener with an explicit quic-go transport config. Passing nil keeps the hardened defaults. A supplied Config with KeepAlivePeriod <= 0 is rejected (the hardened default is used instead) because disabling transport-level keepalive would let otherwise live idle connections be torn down — see defaultQuicConf for the full warning.

func NewTCPListener

func NewTCPListener(addr string) Listener

NewTCPListener returns a plain TCP listener bound to addr (e.g. ":18883").

func NewTLSListener

func NewTLSListener(addr string, config *tls.Config) Listener

NewTLSListener returns a TLS-over-TCP listener bound to addr (e.g. ":18884").

func NewWSListener

func NewWSListener(addr, path string) Listener

NewWSListener returns a WebSocket listener bound to addr (e.g. ":18886"). path is the upgrade endpoint (defaults to "/rmtt" when empty).

func NewWSSListener

func NewWSSListener(addr, path string, config *tls.Config) Listener

NewWSSListener returns a secure WebSocket listener bound to addr.

type Logger

type Logger interface {
	Println(v ...interface{})
	Printf(format string, v ...interface{})
}

Logger is the pluggable logging interface used by the server library. It mirrors the standard library's log.Logger contract, so *log.Logger satisfies it directly. Users can install their own implementation per level via SetLogger/SetDebugLogger/... (see the client package for the paho-style equivalent API).

var (
	ERROR Logger = NOOPLogger{}
	INFO  Logger = NOOPLogger{}
	WARN  Logger = NOOPLogger{}
	DEBUG Logger = NOOPLogger{}
)

Package-level loggers, one per severity. All default to NOOPLogger; replace them with SetLogger or the per-level setters.

type MessageHandler

type MessageHandler func(deviceID string, payload []byte)

MessageHandler receives the uplink PUSH payloads of connected devices.

type NOOPLogger

type NOOPLogger struct{}

NOOPLogger discards all log output; it is the default logger.

func (NOOPLogger) Printf

func (NOOPLogger) Printf(format string, v ...interface{})

func (NOOPLogger) Println

func (NOOPLogger) Println(v ...interface{})

type Server

type Server interface {
	ListenAndServe() error
	ListenAndServeContext(ctx context.Context) error
	Push(deviceID string, payload []byte) error
	Kick(deviceID string, reason byte) error
	Close() error
}

Server is an rmtt server that accepts device connections over its registered listeners, authenticates devices and routes messages.

func NewServer

func NewServer(opts *ServerOptions) Server

NewServer creates a Server from the supplied options. A nil options value uses NewServerOptions defaults; when no listener was added via AddListener, a single TCP listener on options.Port is used.

type ServerOptions

type ServerOptions struct {
	Port               int
	Authenticator      Authenticator
	MessageHandler     MessageHandler
	ConnectionListener ConnectionListener
	KeepalivePolicy    *KeepalivePolicy
	// contains filtered or unexported fields
}

ServerOptions holds the configuration for a Server. Create with NewServerOptions and adjust with the Set* helpers or direct field assignment.

func NewServerOptions

func NewServerOptions() *ServerOptions

NewServerOptions returns ServerOptions with library defaults: TCP port 18883 and DefaultKeepalivePolicy.

func (*ServerOptions) AddListener

func (o *ServerOptions) AddListener(l Listener) *ServerOptions

AddListener registers an extra transport listener (KCP, TLS, QUIC, WS, WSS...). If no listener is added, a single TCP listener on Port is used.

func (*ServerOptions) SetAuthenticator

func (o *ServerOptions) SetAuthenticator(auth Authenticator) *ServerOptions

SetAuthenticator installs the authentication callback. When nil, the CONNECT credential is used directly as the device ID.

func (*ServerOptions) SetConnectionListener

func (o *ServerOptions) SetConnectionListener(listener ConnectionListener) *ServerOptions

SetConnectionListener installs the connection lifecycle event callbacks.

func (*ServerOptions) SetKeepalivePolicy

func (o *ServerOptions) SetKeepalivePolicy(policy *KeepalivePolicy) *ServerOptions

SetKeepalivePolicy replaces the keepalive negotiation policy.

func (*ServerOptions) SetMessageHandler

func (o *ServerOptions) SetMessageHandler(handler MessageHandler) *ServerOptions

SetMessageHandler installs the callback invoked for each uplink PUSH.

func (*ServerOptions) SetPort

func (o *ServerOptions) SetPort(port int) *ServerOptions

SetPort sets the port for the default single TCP listener, used only when no listener was added via AddListener.