client

package
v1.0.2 Latest Latest
Warning

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

 Go to latest Published: Aug 12, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

Package client implements the rmtt protocol client library.

A client connects to a server over one of six transports (tcp, kcp, tls, quic, ws, wss), authenticates with a credential, exchanges PUSH messages and keeps the connection alive with a fixed or adaptive heartbeat.

Usage follows the paho.mqtt.golang style: configure a ClientOptions with NewClientOptions and its setters, create the client with NewClient, then call Connect and wait on the returned Token. Asynchronous results are delivered through Token, and inbound PUSH messages are dispatched to handlers registered with AddPayloadHandlerLast. Lost connections are retried with exponential backoff and jitter when AutoReconnect is set.

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

Index

Constants

View Source 
const (
	NET string = "[net]     "
	CLI string = "[client]     "
)

NET and CLI are the log prefixes used by the net and client subsystems.

Variables

View Source 
var (
	RefusedNotAuthorisedErr      = errors.New("The server has rejected our request. Please check your permissions")
	RefusedBadProtocolVersionErr = errors.New("Server does not support protocol version")
	ProtocolViolationErr         = errors.New("The server has rejected our request. Please check your permissions")
	ErrDisconnectReceived        = errors.New("disconnect received from server")
)

Errors surfaced by Connect when the server rejects the CONNECT request (bad protocol version / not authorised) or a protocol violation is detected during the handshake.

View Source 
var ErrNotConnected = errors.New("not Connected")

ErrNotConnected is returned by Push when the client is not connected.

View Source 
var TimedOut = errors.New("context canceled")

TimedOut is returned by WaitTokenTimeout when the token did not complete within the given duration.

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.

func WaitTokenTimeout 

func WaitTokenTimeout(t Token, d time.Duration) error

WaitTokenTimeout waits up to d for the token to complete and returns its error, or TimedOut if it did not complete in time.

Types

type Client 

type Client interface {
	IsConnected() bool
	Connect() Token
	Push(payload interface{}) Token
	Disconnect(quiesce uint)
	AddPayloadHandlerLast(handler MessageHandler)
}

Client is a single connection to an rmtt server. Create one with NewClient; asynchronous operations are completed via Token.

func NewClient 

func NewClient(o *ClientOptions) Client

NewClient creates a Client from the given options. The options value is copied, so later mutations have no effect on the client.

type ClientOptions 

type ClientOptions struct {
	Servers              []*url.URL
	Credential           string
	Heartbeat            int64
	ProtocolVersion      uint
	ConnectRetry         bool
	ConnectRetryInterval time.Duration
	ConnectTimeout       time.Duration
	WriteTimeout         time.Duration
	AutoReconnect        bool
	OnConnectionLost     ConnectionLostHandler
	MaxReconnectInterval time.Duration
	OnReconnecting       ReconnectHandler
	TLSConfig            *tls.Config
	OnConnectAttempt     ConnectionAttemptHandler
	ReconnectBase        time.Duration
	ReconnectJitter      float64

	// Adaptive heartbeat (client-side policy). When enabled, the client probes the
	// maximum sustainable heartbeat interval within [AdaptiveShort, AdaptiveMax] (capped by the
	// negotiated server_kp from CONNACK) and settles at ~90% of the found maximum. The CONNECT
	// Keepalive proposal becomes AdaptiveMax. Mutually exclusive with a fixed Heartbeat.
	AdaptiveHeartbeat bool
	AdaptiveShort     int64         // seconds
	AdaptiveMax       int64         // seconds
	ProbeCount        int           // consecutive successful short heartbeats before probing starts
	ResponseWindow    time.Duration // max wait for PINGRESP before a probe counts as failed
	FineStep          int64         // seconds; nudge step of the fine-tuning probing phase
	// contains filtered or unexported fields
}

ClientOptions holds all configuration for a Client. Create with NewClientOptions and adjust with the Set* helpers or direct field assignment.

func NewClientOptions 

func NewClientOptions() *ClientOptions

NewClientOptions returns a ClientOptions with the library defaults: heartbeat 10s, connect timeout 30s, auto-reconnect and connect retry enabled, backoff base 1s with 25% jitter.

func (*ClientOptions) AddServer 

func (o *ClientOptions) AddServer(server string) *ClientOptions

AddServer appends a server URL to the server list. A bare address is treated as tcp://; an address starting with ':' gets 127.0.0.1 prepended.

func (*ClientOptions) SetAdaptiveHeartbeat 

func (o *ClientOptions) SetAdaptiveHeartbeat(shortSeconds, maxSeconds int64) *ClientOptions

SetAdaptiveHeartbeat enables adaptive heartbeat. The client probes the maximum sustainable heartbeat interval within [shortSeconds, maxSeconds] (capped by the negotiated server_kp from CONNACK) and settles at ~90% of the found maximum. Replaces a fixed Heartbeat: the CONNECT Keepalive proposal becomes maxSeconds. Incompatible with SetHeartbeat.

func (*ClientOptions) SetConnectTimeout 

func (o *ClientOptions) SetConnectTimeout(k time.Duration) *ClientOptions

SetConnectTimeout sets the timeout for the connection handshake.

func (*ClientOptions) SetConnectionAttemptHandler 

func (o *ClientOptions) SetConnectionAttemptHandler(onConnectAttempt ConnectionAttemptHandler) *ClientOptions

SetConnectionAttemptHandler registers a handler invoked before each connection attempt.

func (*ClientOptions) SetCredential 

func (o *ClientOptions) SetCredential(id string) *ClientOptions

SetCredential sets the credential sent in CONNECT, used by the server for authentication and device identity.

func (*ClientOptions) SetFineStep 

func (o *ClientOptions) SetFineStep(seconds int64) *ClientOptions

SetFineStep sets the nudge step (seconds) used in the fine-tuning probing phase (default 5). Only meaningful with SetAdaptiveHeartbeat.

func (*ClientOptions) SetHeartbeat 

func (o *ClientOptions) SetHeartbeat(k time.Duration) *ClientOptions

SetHeartbeat sets the fixed heartbeat interval (in seconds) sent in CONNECT as the Keepalive proposal. Incompatible with SetAdaptiveHeartbeat.

func (*ClientOptions) SetProbeCount 

func (o *ClientOptions) SetProbeCount(n int) *ClientOptions

SetProbeCount sets the number of consecutive successful short heartbeats required before the probing phase starts (default 3). Only meaningful with SetAdaptiveHeartbeat.

func (*ClientOptions) SetQuicConfig 

func (o *ClientOptions) SetQuicConfig(config *quic.Config) *ClientOptions

SetQuicConfig overrides the QUIC transport settings used for "quic://" servers. Pass nil to restore the library's hardened defaults (MaxIdleTimeout 15min, KeepAlivePeriod 30s).

WARNING: you take full responsibility for your own values. A Config with KeepAlivePeriod <= 0 (which disables transport-level keepalive) or a MaxIdleTimeout shorter than the application's heartbeat/report interval will reproduce the classic periodic "timeout: no recent network activity" drops whenever the adaptive heartbeat grows beyond the idle window — that is the exact bug this library ships its safe defaults to prevent. For your own safety, SetQuicConfig rejects KeepAlivePeriod <= 0 and falls back to the hardened default rather than letting the connection die silently.

func (*ClientOptions) SetReconnectBase 

func (o *ClientOptions) SetReconnectBase(k time.Duration) *ClientOptions

SetReconnectBase sets the base sleep between reconnection attempts.

func (*ClientOptions) SetReconnectJitter 

func (o *ClientOptions) SetReconnectJitter(j float64) *ClientOptions

SetReconnectJitter sets the jitter factor applied to the backoff sleep.

func (*ClientOptions) SetResponseWindow 

func (o *ClientOptions) SetResponseWindow(d time.Duration) *ClientOptions

SetResponseWindow sets the maximum wait for a PINGRESP before counting a probe as failed (default 2s). Only meaningful with SetAdaptiveHeartbeat.

func (*ClientOptions) SetTlsConfig 

func (o *ClientOptions) SetTlsConfig(config *tls.Config) *ClientOptions

SetTlsConfig sets the TLS configuration used for tls://, wss:// and quic:// connections.

func (*ClientOptions) SetWriteTimeout 

func (o *ClientOptions) SetWriteTimeout(k time.Duration) *ClientOptions

SetWriteTimeout sets the timeout for writing an outbound packet.

type ConnectToken 

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

ConnectToken completes when the CONNECT handshake finishes; ReturnCode holds the server's CONNACK return code.

func (*ConnectToken) Done 

func (b *ConnectToken) Done() <-chan struct{}

func (*ConnectToken) Error 

func (b *ConnectToken) Error() error

func (*ConnectToken) HandlerError 

func (b *ConnectToken) HandlerError(e error)

func (*ConnectToken) ReturnCode 

func (c *ConnectToken) ReturnCode() byte

ReturnCode returns the CONNACK return code received from the server.

func (*ConnectToken) SetErrorHandler 

func (b *ConnectToken) SetErrorHandler(f func(error))

func (*ConnectToken) Wait 

func (b *ConnectToken) Wait() bool

func (*ConnectToken) WaitTimeout 

func (b *ConnectToken) WaitTimeout(d time.Duration) bool

type ConnectionAttemptHandler 

type ConnectionAttemptHandler func(server *url.URL, tlsCfg *tls.Config) *tls.Config

ConnectionAttemptHandler is invoked before each connection attempt; it may return a modified *tls.Config for that attempt.

type ConnectionLostHandler 

type ConnectionLostHandler func(Client, error)

ConnectionLostHandler is invoked when the connection is lost.

type DisconnectToken 

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

DisconnectToken completes when the DISCONNECT packet has been sent.

func (*DisconnectToken) Done 

func (b *DisconnectToken) Done() <-chan struct{}

func (*DisconnectToken) Error 

func (b *DisconnectToken) Error() error

func (*DisconnectToken) HandlerError 

func (b *DisconnectToken) HandlerError(e error)

func (*DisconnectToken) SetErrorHandler 

func (b *DisconnectToken) SetErrorHandler(f func(error))

func (*DisconnectToken) Wait 

func (b *DisconnectToken) Wait() bool

func (*DisconnectToken) WaitTimeout 

func (b *DisconnectToken) WaitTimeout(d time.Duration) bool

type Logger 

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

Logger is the pluggable logging interface; it mirrors log.Logger's Println/Printf, so *log.Logger satisfies it directly. 

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 Message 

type Message interface {
	Payload() []byte
}

Message is a PUSH message received from the server.

type MessageHandler 

type MessageHandler func(Client, Message)

MessageHandler receives a PUSH message from the server.

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 PacketAndToken 

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

PacketAndToken pairs a control packet with the token tracking its completion; used for the internal outbound queues.

type PushToken 

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

PushToken completes when the PUSH packet has been written to the connection.

func (*PushToken) Done 

func (b *PushToken) Done() <-chan struct{}

func (*PushToken) Error 

func (b *PushToken) Error() error

func (*PushToken) HandlerError 

func (b *PushToken) HandlerError(e error)

func (*PushToken) SetErrorHandler 

func (b *PushToken) SetErrorHandler(f func(error))

func (*PushToken) Wait 

func (b *PushToken) Wait() bool

func (*PushToken) WaitTimeout 

func (b *PushToken) WaitTimeout(d time.Duration) bool

type ReconnectHandler 

type ReconnectHandler func(Client, *ClientOptions)

ReconnectHandler is invoked before each reconnection attempt.

type Token 

type Token interface {
	Wait() bool
	WaitTimeout(time.Duration) bool
	Done() <-chan struct{}
	Error() error
	SetErrorHandler(func(error))
}

Token tracks the completion of an asynchronous operation (Connect, Push, Disconnect). Wait for completion, then check Error.

type TokenErrorSetter 

type TokenErrorSetter interface {
	// contains filtered or unexported methods
}

TokenErrorSetter is implemented by tokens that can carry an error.

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.