mqtt

package
v0.51.0 Latest Latest
Warning

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

 Go to latest Published: Aug 5, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultKeepAlive       = 60 * time.Second
	DefaultConnectTimeout  = 10 * time.Second
	DefaultWriteTimeout    = 5 * time.Second
	DefaultAckTimeout      = 10 * time.Second
	DefaultPingTimeout     = 5 * time.Second
	DefaultReconnectMin    = 1 * time.Second
	DefaultReconnectMax    = 2 * time.Minute
	DefaultMaxInflight     = 100
	DefaultMessageChanSize = 256
)

Default values.

Variables

View Source 
var (
	// Configuration errors.
	ErrNoServers       = errors.New("no servers configured")
	ErrEmptyClientID   = errors.New("client ID cannot be empty")
	ErrInvalidProtocol = errors.New("invalid protocol version (must be 4 or 5)")
	ErrNilOptions      = errors.New("options cannot be nil")

	// Connection errors.
	ErrNotConnected     = errors.New("client not connected")
	ErrAlreadyConnected = errors.New("client already connected")
	ErrConnectFailed    = errors.New("connection failed")
	ErrConnectRejected  = errors.New("connection rejected by broker")
	ErrConnectTimeout   = errors.New("connection timeout")
	ErrPingTimeout      = errors.New("ping response timeout")

	// Operation errors.
	ErrTimeout              = errors.New("operation timed out")
	ErrMaxInflight          = errors.New("maximum inflight messages exceeded")
	ErrConnectionLost       = errors.New("connection lost")
	ErrClientClosed         = errors.New("client has been closed")
	ErrDraining             = errors.New("client is draining")
	ErrInvalidMessage       = errors.New("invalid message")
	ErrInvalidSubscribeOpt  = errors.New("invalid subscribe option")
	ErrInvalidQoS           = errors.New("invalid QoS level (must be 0, 1, or 2)")
	ErrInvalidTopic         = errors.New("invalid topic")
	ErrSubscribeFailed      = errors.New("subscription failed")
	ErrSlowConsumer         = errors.New("slow consumer: message dropped")
	ErrOutboundBackpressure = errors.New("outbound publish backpressure: message dropped")
	ErrReconnectBufferFull  = errors.New("reconnect publish buffer is full")
	ErrQueueAckRequiresV5   = errors.New("queue acknowledgments require MQTT v5 user properties")
	ErrQueueAckMissingGroup = errors.New("group-id required for queue acknowledgment")

	// Authentication errors.
	ErrAuthFailed         = errors.New("enhanced authentication failed")
	ErrNoAuthHandler      = errors.New("server sent AUTH but no OnAuth handler configured")
	ErrAuthNotV5          = errors.New("enhanced authentication requires MQTT 5.0")
	ErrAuthMethodMissing  = errors.New("enhanced authentication method not set")
	ErrAuthMethodMismatch = errors.New("enhanced authentication method mismatch")

	// Protocol errors.
	ErrUnexpectedPacket = errors.New("unexpected packet type")
	ErrMalformedPacket  = errors.New("malformed packet")
)

Client errors.

Functions

This section is empty.

Types

type Client 

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

Client is a thread-safe MQTT client.

func New 

func New(opts *Options) (*Client, error)

New creates a new MQTT client with the given options.

func (*Client) Ack 

func (c *Client) Ack(ctx context.Context, queueName, messageID string) error

Ack acknowledges successful processing of a queue message.

func (*Client) AckWithGroup 

func (c *Client) AckWithGroup(ctx context.Context, queueName, messageID, groupID string) error

AckWithGroup acknowledges a queue message with an explicit consumer group.

func (*Client) Close 

func (c *Client) Close(ctx context.Context) error

func (*Client) Connect 

func (c *Client) Connect(ctx context.Context) error

func (*Client) Disconnect 

func (c *Client) Disconnect(ctx context.Context) error

func (*Client) DisconnectWithReason 

func (c *Client) DisconnectWithReason(ctx context.Context, reasonCode byte, sessionExpiry uint32, reasonString string) error

func (*Client) Drain 

func (c *Client) Drain(ctx context.Context) error

func (*Client) DroppedMessages 

func (c *Client) DroppedMessages() uint64

func (*Client) IsConnected 

func (c *Client) IsConnected() bool

func (*Client) Nack 

func (c *Client) Nack(ctx context.Context, queueName, messageID string) error

Nack negatively acknowledges a queue message, triggering retry.

func (*Client) NackWithGroup 

func (c *Client) NackWithGroup(ctx context.Context, queueName, messageID, groupID string) error

NackWithGroup negatively acknowledges a queue message with an explicit consumer group.

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, topic string, payload []byte, qos byte, retain bool) error

func (*Client) PublishAsync 

func (c *Client) PublishAsync(ctx context.Context, topic string, payload []byte, qos byte, retain bool) *PublishToken

func (*Client) PublishMessage 

func (c *Client) PublishMessage(ctx context.Context, msg *Message, waitAck bool) (*pendingOp, uint16, error)

PublishMessage sends a message with optional MQTT 5.0 publish properties. For MQTT 3.1.1, publish properties are ignored.

func (*Client) PublishMessageAsync 

func (c *Client) PublishMessageAsync(ctx context.Context, msg *Message) *PublishToken

func (*Client) PublishToQueue 

func (c *Client) PublishToQueue(ctx context.Context, queueName string, payload []byte) error

PublishToQueue publishes a message to a durable queue. The queueName should NOT include the "$queue/" prefix - it will be added automatically.

func (*Client) PublishToQueueWithOptions 

func (c *Client) PublishToQueueWithOptions(ctx context.Context, opts *QueuePublishOptions) error

PublishToQueueWithOptions publishes a message to a durable queue with full control. The queueName should NOT include the "$queue/" prefix - it will be added automatically.

func (*Client) Reject 

func (c *Client) Reject(ctx context.Context, queueName, messageID string) error

Reject rejects a queue message, sending it to the dead-letter queue.

func (*Client) RejectWithGroup 

func (c *Client) RejectWithGroup(ctx context.Context, queueName, messageID, groupID string) error

RejectWithGroup rejects a queue message with an explicit consumer group.

func (*Client) RejectWithGroupReason 

func (c *Client) RejectWithGroupReason(ctx context.Context, queueName, messageID, groupID, reason string) error

RejectWithGroupReason rejects a queue message with explicit group and reason.

func (*Client) RejectWithReason 

func (c *Client) RejectWithReason(ctx context.Context, queueName, messageID, reason string) error

RejectWithReason rejects a queue message with a broker-visible reason.

func (*Client) SendAuth 

func (c *Client) SendAuth(reasonCode byte, authData []byte) error

func (*Client) ServerCapabilities 

func (c *Client) ServerCapabilities() *ServerCapabilities

func (*Client) State 

func (c *Client) State() State

func (*Client) Subscribe 

func (c *Client) Subscribe(ctx context.Context, topics map[string]byte) error

func (*Client) SubscribeAsync 

func (c *Client) SubscribeAsync(ctx context.Context, topics map[string]byte) *SubscribeToken

SubscribeAsync subscribes in a background goroutine and returns a completion token.

func (*Client) SubscribeSingle 

func (c *Client) SubscribeSingle(ctx context.Context, topic string, qos byte) error

func (*Client) SubscribeToQueue 

func (c *Client) SubscribeToQueue(ctx context.Context, queueName, consumerGroup string, handler QueueMessageHandler) error

SubscribeToQueue subscribes to a durable queue with a consumer group. The queueName should NOT include the "$queue/" prefix - it will be added automatically. The handler will be called for each message received from the queue.

func (*Client) SubscribeWithOptions 

func (c *Client) SubscribeWithOptions(ctx context.Context, opts ...*SubscribeOption) error

func (*Client) SubscribeWithOptionsAsync 

func (c *Client) SubscribeWithOptionsAsync(ctx context.Context, opts ...*SubscribeOption) *SubscribeToken

func (*Client) Unsubscribe 

func (c *Client) Unsubscribe(ctx context.Context, topics ...string) error

Unsubscribe unsubscribes from one or more topics.

func (*Client) UnsubscribeAsync 

func (c *Client) UnsubscribeAsync(ctx context.Context, topics ...string) *UnsubscribeToken

func (*Client) UnsubscribeFromQueue 

func (c *Client) UnsubscribeFromQueue(ctx context.Context, queueName string) error

UnsubscribeFromQueue unsubscribes from a durable queue. The queueName should NOT include the "$queue/" prefix - it will be added automatically.

type ConnAckError 

type ConnAckError byte

ConnAckError represents MQTT CONNACK return codes. 

const (
	ErrConnAccepted           ConnAckError = 0x00
	ErrConnRefusedProtocol    ConnAckError = 0x01
	ErrConnRefusedIDRejected  ConnAckError = 0x02
	ErrConnRefusedUnavailable ConnAckError = 0x03
	ErrConnRefusedBadAuth     ConnAckError = 0x04
	ErrConnRefusedNotAuth     ConnAckError = 0x05
)

MQTT 3.1.1 CONNACK return codes.

func (ConnAckError) Error 

func (c ConnAckError) Error() string

Error implements the error interface.

func (ConnAckError) String 

func (c ConnAckError) String() string

String returns a human-readable description of the CONNACK code.

type DroppedMessage 

type DroppedMessage struct {
	Direction   DroppedMessageDirection
	Reason      DroppedMessageReason
	Topic       string
	QoS         byte
	PayloadSize int
	Timestamp   time.Time
}

DroppedMessage describes a message dropped by pressure policies.

type DroppedMessageDirection 

type DroppedMessageDirection string

DroppedMessageDirection identifies inbound or outbound drop path. 

const (
	DroppedMessageInbound  DroppedMessageDirection = "inbound"
	DroppedMessageOutbound DroppedMessageDirection = "outbound"
)

type DroppedMessageReason 

type DroppedMessageReason string

DroppedMessageReason describes why the message was dropped. 

const (
	DroppedReasonSlowConsumer         DroppedMessageReason = "slow_consumer"
	DroppedReasonOutboundBackpressure DroppedMessageReason = "outbound_backpressure"
	DroppedReasonReconnectBufferFull  DroppedMessageReason = "reconnect_buffer_full"
)

type Message 

type Message struct {
	Topic     string
	Payload   []byte
	QoS       byte
	Retain    bool
	Dup       bool
	PacketID  uint16
	Timestamp time.Time

	// MQTT 5.0 properties
	PayloadFormat   *byte
	MessageExpiry   *uint32
	ContentType     string
	ResponseTopic   string
	CorrelationData []byte
	UserProperties  map[string]string
	SubscriptionIDs []uint32
}

Message represents an MQTT message.

func NewMessage 

func NewMessage(topic string, payload []byte, qos byte, retain bool) *Message

NewMessage creates a new message with the given parameters.

func (*Message) Copy 

func (m *Message) Copy() *Message

Copy creates a deep copy of the message.

type MessageStore 

type MessageStore interface {
	// StoreOutbound stores an outbound message awaiting acknowledgment.
	StoreOutbound(packetID uint16, msg *Message) error

	// GetOutbound retrieves an outbound message by packet ID.
	GetOutbound(packetID uint16) (*Message, bool)

	// DeleteOutbound removes an outbound message after acknowledgment.
	DeleteOutbound(packetID uint16) error

	// GetAllOutbound returns all stored outbound messages (for reconnection).
	GetAllOutbound() []*Message

	// StoreInbound stores an inbound QoS 2 message awaiting PUBREL.
	StoreInbound(packetID uint16, msg *Message) error

	// GetInbound retrieves an inbound QoS 2 message by packet ID.
	GetInbound(packetID uint16) (*Message, bool)

	// DeleteInbound removes an inbound message after PUBREL/PUBCOMP.
	DeleteInbound(packetID uint16) error

	// Reset clears all stored messages.
	Reset() error

	// Close releases any resources.
	Close() error
}

MessageStore provides persistence for QoS 1/2 messages. It stores outbound messages waiting for acknowledgment and inbound QoS 2 messages waiting for PUBREL.

func NewBadgerStore 

func NewBadgerStore(opts *storebadger.Options) (MessageStore, error)

NewBadgerStore creates a MessageStore backed by BadgerDB key/value storage.

func NewMemoryStore 

func NewMemoryStore() MessageStore

NewMemoryStore creates a MessageStore backed by in-memory key/value storage.

type Options 

type Options struct {
	// Connection
	Servers        []string      // List of broker addresses (host:port)
	ClientID       string        // Client identifier
	Username       string        // Optional username
	Password       string        // Optional password
	TLSConfig      *tls.Config   // TLS configuration (nil for plain TCP)
	ConnectTimeout time.Duration // Timeout for connection attempts
	WriteTimeout   time.Duration // Timeout for write operations
	KeepAlive      time.Duration // Keep-alive interval (0 to disable)
	PingTimeout    time.Duration // Timeout waiting for PINGRESP

	// Session
	CleanSession    bool   // Start with clean session
	SessionExpiry   uint32 // Session expiry interval (MQTT 5.0, seconds)
	ProtocolVersion byte   // 4 for MQTT 3.1.1, 5 for MQTT 5.0

	// MQTT 5.0 Connect Properties
	ReceiveMaximum      uint16 // Maximum inflight messages client accepts (0 = use default 65535)
	MaximumPacketSize   uint32 // Maximum packet size client accepts (0 = no limit)
	TopicAliasMaximum   uint16 // Maximum topic aliases client accepts (0 = disabled)
	RequestResponseInfo bool   // Request server to send response information in CONNACK
	RequestProblemInfo  bool   // Request detailed error information (default true)

	// Will
	Will *WillMessage // Last will and testament

	// QoS
	AckTimeout  time.Duration // Timeout waiting for PUBACK/SUBACK
	MaxInflight int           // Maximum inflight messages

	// Reconnection
	AutoReconnect        bool          // Enable automatic reconnection
	ReconnectBackoff     time.Duration // Initial reconnect delay
	MaxReconnectWait     time.Duration // Maximum reconnect delay
	ReconnectJitter      time.Duration // Random reconnect delay component [0, ReconnectJitter]
	MaxReconnectAttempts int           // Maximum reconnect attempts (0 = unlimited)
	ReconnectBufSize     int           // Max buffered outbound publish bytes while disconnected (0 = disabled)

	// Callbacks
	OnConnect            func()                                                                    // Called on successful connection
	OnConnectionLost     func(error)                                                               // Called when connection is lost
	OnReconnecting       func(attempt int)                                                         // Called before each reconnect attempt
	OnReconnectFailed    func(error)                                                               // Called when reconnect retries are exhausted
	OnAsyncError         func(error)                                                               // Called for asynchronous non-fatal errors (slow consumer, reconnect buffer overflow)
	OnDroppedMessage     func(*DroppedMessage)                                                     // Called when inbound/outbound messages are dropped by pressure policies
	OnMessage            func(topic string, payload []byte, qos byte)                              // Called for incoming messages (basic)
	OnMessageV2          func(msg *Message)                                                        // Called for incoming messages (full context, takes precedence over OnMessage)
	OnServerCapabilities func(*ServerCapabilities)                                                 // Called when server capabilities received (MQTT 5.0)
	OnAuth               func(reasonCode byte, authMethod string, authData []byte) ([]byte, error) // Called for enhanced authentication (MQTT 5.0)

	// Enhanced Authentication (MQTT 5.0)
	AuthMethod string // Authentication method for enhanced auth
	AuthData   []byte // Authentication data for initial CONNECT

	// Advanced
	MessageChanSize            int                        // Size of internal message channel
	MaxPendingMessages         int                        // Max queued callback messages before dropping (0 = disabled)
	MaxPendingBytes            int64                      // Max queued callback bytes before dropping (0 = disabled)
	SlowConsumerPolicy         SlowConsumerPolicy         // Slow-consumer behavior for unordered callbacks
	SlowConsumerBlockTimeout   time.Duration              // Blocking duration for SlowConsumerBlockWithTimeout
	MaxOutboundPendingMessages int                        // Max pending outbound publish writes before pressure policy applies (0 = disabled)
	MaxOutboundPendingBytes    int64                      // Max pending outbound publish write bytes before pressure policy applies (0 = disabled)
	OutboundBackpressurePolicy OutboundBackpressurePolicy // Outbound publish pressure behavior
	OutboundBlockTimeout       time.Duration              // Blocking duration for OutboundBackpressureBlockWithTimeout
	OrderMatters               bool                       // Maintain message order (may reduce throughput)
	Store                      MessageStore               // Message store for QoS 1/2 (nil = in-memory)
}

Options configures the MQTT client.

func NewOptions 

func NewOptions() *Options

NewOptions creates Options with sensible defaults.

func (*Options) SetAckTimeout 

func (o *Options) SetAckTimeout(d time.Duration) *Options

SetAckTimeout sets the acknowledgment timeout.

func (*Options) SetAuthData 

func (o *Options) SetAuthData(data []byte) *Options

SetAuthData sets the initial authentication data for enhanced auth (MQTT 5.0).

func (*Options) SetAuthMethod 

func (o *Options) SetAuthMethod(method string) *Options

SetAuthMethod sets the authentication method for enhanced auth (MQTT 5.0).

func (*Options) SetAutoReconnect 

func (o *Options) SetAutoReconnect(enable bool) *Options

SetAutoReconnect enables or disables automatic reconnection.

func (*Options) SetCleanSession 

func (o *Options) SetCleanSession(clean bool) *Options

SetCleanSession sets the clean session flag.

func (*Options) SetClientID 

func (o *Options) SetClientID(id string) *Options

SetClientID sets the client identifier.

func (*Options) SetConnectTimeout 

func (o *Options) SetConnectTimeout(d time.Duration) *Options

SetConnectTimeout sets the connection timeout.

func (*Options) SetCredentials 

func (o *Options) SetCredentials(username, password string) *Options

SetCredentials sets username and password.

func (*Options) SetKeepAlive 

func (o *Options) SetKeepAlive(d time.Duration) *Options

SetKeepAlive sets the keep-alive interval.

func (*Options) SetMaxInflight 

func (o *Options) SetMaxInflight(max int) *Options

SetMaxInflight sets the maximum number of inflight messages.

func (*Options) SetMaxOutboundPendingBytes 

func (o *Options) SetMaxOutboundPendingBytes(max int64) *Options

SetMaxOutboundPendingBytes sets max pending outbound publish write bytes before applying pressure policy. A value <= 0 disables this limit.

func (*Options) SetMaxOutboundPendingMessages 

func (o *Options) SetMaxOutboundPendingMessages(max int) *Options

SetMaxOutboundPendingMessages sets max pending outbound publish writes before applying pressure policy. A value <= 0 disables this limit.

func (*Options) SetMaxPendingBytes 

func (o *Options) SetMaxPendingBytes(max int64) *Options

SetMaxPendingBytes sets max queued callback bytes before dropping. A value <= 0 disables this limit.

func (*Options) SetMaxPendingMessages 

func (o *Options) SetMaxPendingMessages(max int) *Options

SetMaxPendingMessages sets max queued callback messages before dropping. A value <= 0 disables this limit.

func (*Options) SetMaxReconnectAttempts 

func (o *Options) SetMaxReconnectAttempts(max int) *Options

SetMaxReconnectAttempts sets maximum reconnect attempts (0 = unlimited).

func (*Options) SetMaxReconnectWait 

func (o *Options) SetMaxReconnectWait(d time.Duration) *Options

SetMaxReconnectWait sets maximum reconnect delay.

func (*Options) SetMaximumPacketSize 

func (o *Options) SetMaximumPacketSize(size uint32) *Options

SetMaximumPacketSize sets the maximum packet size the client accepts (MQTT 5.0). 0 means no limit beyond protocol maximum (256 MB).

func (*Options) SetMessageChanSize 

func (o *Options) SetMessageChanSize(size int) *Options

SetMessageChanSize sets internal message dispatch channel size.

func (*Options) SetOnAsyncError 

func (o *Options) SetOnAsyncError(fn func(error)) *Options

SetOnAsyncError sets callback for non-fatal asynchronous errors.

func (*Options) SetOnAuth 

func (o *Options) SetOnAuth(fn func(reasonCode byte, authMethod string, authData []byte) ([]byte, error)) *Options

SetOnAuth sets the enhanced authentication callback (MQTT 5.0). The callback receives the server's reason code, auth method, and auth data, and should return response auth data or an error to abort.

func (*Options) SetOnConnect 

func (o *Options) SetOnConnect(fn func()) *Options

SetOnConnect sets the connection callback.

func (*Options) SetOnConnectionLost 

func (o *Options) SetOnConnectionLost(fn func(error)) *Options

SetOnConnectionLost sets the connection lost callback.

func (*Options) SetOnDroppedMessage 

func (o *Options) SetOnDroppedMessage(fn func(*DroppedMessage)) *Options

SetOnDroppedMessage sets callback for dropped inbound/outbound messages due to pressure policies.

func (*Options) SetOnMessage 

func (o *Options) SetOnMessage(fn func(topic string, payload []byte, qos byte)) *Options

SetOnMessage sets the message handler callback.

func (*Options) SetOnMessageV2 

func (o *Options) SetOnMessageV2(fn func(msg *Message)) *Options

SetOnMessageV2 sets the enhanced message handler callback with full message context. This takes precedence over OnMessage if both are set. The Message includes MQTT v5 properties, user properties, and other metadata.

func (*Options) SetOnReconnectFailed 

func (o *Options) SetOnReconnectFailed(fn func(error)) *Options

SetOnReconnectFailed sets callback for reconnect exhaustion.

func (*Options) SetOnReconnecting 

func (o *Options) SetOnReconnecting(fn func(attempt int)) *Options

SetOnReconnecting sets the reconnecting callback.

func (*Options) SetOnServerCapabilities 

func (o *Options) SetOnServerCapabilities(fn func(*ServerCapabilities)) *Options

SetOnServerCapabilities sets the server capabilities callback (MQTT 5.0).

func (*Options) SetOrderMatters 

func (o *Options) SetOrderMatters(orderMatters bool) *Options

SetOrderMatters controls whether message callback ordering is preserved.

func (*Options) SetOutboundBackpressurePolicy 

func (o *Options) SetOutboundBackpressurePolicy(policy OutboundBackpressurePolicy) *Options

SetOutboundBackpressurePolicy sets outbound publish pressure behavior.

func (*Options) SetOutboundBlockTimeout 

func (o *Options) SetOutboundBlockTimeout(timeout time.Duration) *Options

SetOutboundBlockTimeout sets blocking duration for OutboundBackpressureBlockWithTimeout.

func (*Options) SetPingTimeout 

func (o *Options) SetPingTimeout(d time.Duration) *Options

SetPingTimeout sets timeout waiting for PINGRESP.

func (*Options) SetProtocolVersion 

func (o *Options) SetProtocolVersion(v byte) *Options

SetProtocolVersion sets MQTT protocol version (4 or 5).

func (*Options) SetReceiveMaximum 

func (o *Options) SetReceiveMaximum(max uint16) *Options

SetReceiveMaximum sets the maximum inflight messages the client accepts (MQTT 5.0). Default is 65535 if not set. Must be > 0.

func (*Options) SetReconnectBackoff 

func (o *Options) SetReconnectBackoff(d time.Duration) *Options

SetReconnectBackoff sets initial reconnect delay.

func (*Options) SetReconnectBufferSize 

func (o *Options) SetReconnectBufferSize(bytes int) *Options

SetReconnectBufferSize sets max buffered publish bytes while disconnected. A value <= 0 disables buffering.

func (*Options) SetReconnectJitter 

func (o *Options) SetReconnectJitter(d time.Duration) *Options

SetReconnectJitter sets reconnect jitter added to each reconnect sleep.

func (*Options) SetRequestProblemInfo 

func (o *Options) SetRequestProblemInfo(request bool) *Options

SetRequestProblemInfo requests detailed error information from server (MQTT 5.0). When true, server includes reason strings and user properties in error responses. Default is true.

func (*Options) SetRequestResponseInfo 

func (o *Options) SetRequestResponseInfo(request bool) *Options

SetRequestResponseInfo requests the server to send response information (MQTT 5.0). The server may include response information in CONNACK which can be used for request/response patterns.

func (*Options) SetServers 

func (o *Options) SetServers(servers ...string) *Options

SetServers sets the broker addresses.

func (*Options) SetSessionExpiry 

func (o *Options) SetSessionExpiry(seconds uint32) *Options

SetSessionExpiry sets the session expiry interval in seconds (MQTT 5.0). 0 means the session expires when the network connection closes.

func (*Options) SetSlowConsumerBlockTimeout 

func (o *Options) SetSlowConsumerBlockTimeout(timeout time.Duration) *Options

SetSlowConsumerBlockTimeout sets blocking duration for SlowConsumerBlockWithTimeout.

func (*Options) SetSlowConsumerPolicy 

func (o *Options) SetSlowConsumerPolicy(policy SlowConsumerPolicy) *Options

SetSlowConsumerPolicy sets callback queue pressure policy.

func (*Options) SetStore 

func (o *Options) SetStore(store MessageStore) *Options

SetStore sets the message store for QoS 1/2 persistence.

func (*Options) SetTLSConfig 

func (o *Options) SetTLSConfig(cfg *tls.Config) *Options

SetTLSConfig sets TLS configuration.

func (*Options) SetTopicAliasMaximum 

func (o *Options) SetTopicAliasMaximum(max uint16) *Options

SetTopicAliasMaximum sets the maximum topic aliases the client accepts (MQTT 5.0). 0 means topic aliases are disabled. Server cannot use topic aliases if set to 0.

func (*Options) SetWill 

func (o *Options) SetWill(topic string, payload []byte, qos byte, retain bool) *Options

SetWill sets the last will and testament.

func (*Options) SetWriteTimeout 

func (o *Options) SetWriteTimeout(d time.Duration) *Options

SetWriteTimeout sets the write timeout.

func (*Options) Validate 

func (o *Options) Validate() error

Validate checks the options for errors.

type OutboundBackpressurePolicy 

type OutboundBackpressurePolicy string

OutboundBackpressurePolicy controls behavior when outbound publish pressure is hit. 

const (
	// OutboundBackpressureBlock blocks until outbound capacity is available.
	OutboundBackpressureBlock OutboundBackpressurePolicy = "block"
	// OutboundBackpressureBlockWithTimeout blocks up to OutboundBlockTimeout.
	OutboundBackpressureBlockWithTimeout OutboundBackpressurePolicy = "block_with_timeout"
	// OutboundBackpressureDropNew drops the new publish when outbound pressure is hit.
	OutboundBackpressureDropNew OutboundBackpressurePolicy = "drop_new"
)

type PublishToken 

type PublishToken struct {
	MessageID uint16
	// contains filtered or unexported fields
}

PublishToken is returned by Publish operations.

func (*PublishToken) Done 

func (t *PublishToken) Done() <-chan struct{}

Done returns a channel closed when publish completes.

func (*PublishToken) Error 

func (t *PublishToken) Error() error

Error returns publish error if already completed.

func (*PublishToken) Wait 

func (t *PublishToken) Wait() error

Wait blocks until publish completes.

func (*PublishToken) WaitTimeout 

func (t *PublishToken) WaitTimeout(timeout time.Duration) error

WaitTimeout blocks until publish completes or timeout occurs.

type QueueMessage 

type QueueMessage struct {
	*Message // Embedded standard MQTT message

	MessageID string // Unique message ID for acknowledgment
	GroupID   string // Consumer group ID for acknowledgment
	Offset    uint64 // Queue offset
	Sequence  uint64 // Legacy alias for Offset
	// contains filtered or unexported fields
}

QueueMessage represents a message received from a durable queue.

func (*QueueMessage) Ack 

func (qm *QueueMessage) Ack(ctx context.Context) error

Ack acknowledges successful message processing. The message will be removed from the queue.

func (*QueueMessage) Nack 

func (qm *QueueMessage) Nack(ctx context.Context) error

Nack negatively acknowledges the message, triggering a retry. The message will be redelivered according to the retry policy.

func (*QueueMessage) Reject 

func (qm *QueueMessage) Reject(ctx context.Context) error

Reject rejects the message, sending it to the dead-letter queue. The message will not be retried.

func (*QueueMessage) RejectWithReason 

func (qm *QueueMessage) RejectWithReason(ctx context.Context, reason string) error

RejectWithReason rejects the message and provides a broker-visible reason.

type QueueMessageHandler 

type QueueMessageHandler func(msg *QueueMessage)

QueueMessageHandler is called when a queue message is received.

type QueuePublishOptions 

type QueuePublishOptions struct {
	QueueName  string            // Queue topic name (without $queue/ prefix)
	Payload    []byte            // Message payload
	Properties map[string]string // Additional user properties (optional)
	QoS        byte              // Quality of Service (default 1)
}

QueuePublishOptions configures queue message publishing.

type ServerCapabilities 

type ServerCapabilities struct {
	// SessionExpiryInterval is the session expiry interval negotiated by the server.
	// If nil, the server accepted the client's requested value.
	SessionExpiryInterval *uint32

	// ReceiveMaximum is the maximum number of QoS 1 and QoS 2 publications
	// that the server is willing to process concurrently.
	// Default is 65535 if not present.
	ReceiveMaximum uint16

	// MaximumQoS is the maximum QoS level the server supports.
	// Default is 2 (QoS 0, 1, and 2 supported).
	MaximumQoS byte

	// RetainAvailable indicates whether the server supports retained messages.
	// Default is true if not present.
	RetainAvailable bool

	// MaximumPacketSize is the maximum packet size the server is willing to accept.
	// If nil, there is no limit beyond the protocol maximum.
	MaximumPacketSize *uint32

	// AssignedClientID is the client identifier assigned by the server if the
	// client connected with an empty client ID.
	AssignedClientID string

	// TopicAliasMaximum is the maximum topic alias value the server accepts.
	// 0 means topic aliases are not supported by the server.
	TopicAliasMaximum uint16

	// ReasonString provides additional diagnostic information.
	ReasonString string

	// UserProperties contains user-defined properties from the server.
	UserProperties map[string]string

	// WildcardSubscriptionAvailable indicates whether wildcard subscriptions are supported.
	// Default is true if not present.
	WildcardSubscriptionAvailable bool

	// SubscriptionIdentifiersAvailable indicates whether subscription identifiers are supported.
	// Default is true if not present.
	SubscriptionIdentifiersAvailable bool

	// SharedSubscriptionAvailable indicates whether shared subscriptions are supported.
	// Default is true if not present.
	SharedSubscriptionAvailable bool

	// ServerKeepAlive is the keep alive time assigned by the server.
	// If nil, the server accepted the client's requested value.
	ServerKeepAlive *uint16

	// ResponseInformation can be used by the client to construct response topics.
	ResponseInformation string

	// ServerReference indicates another server the client can use.
	ServerReference string

	// AuthenticationMethod is the authentication method used.
	AuthenticationMethod string

	// AuthenticationData contains authentication-specific data.
	AuthenticationData []byte
}

ServerCapabilities represents the capabilities and limits advertised by the server in the CONNACK packet (MQTT 5.0).

type SlowConsumerPolicy 

type SlowConsumerPolicy string

SlowConsumerPolicy controls behavior when callback queue pressure is hit. 

const (
	// SlowConsumerDropNew drops the newly received message.
	SlowConsumerDropNew SlowConsumerPolicy = "drop_new"
	// SlowConsumerDropOldest drops one queued message and keeps the new one.
	SlowConsumerDropOldest SlowConsumerPolicy = "drop_oldest"
	// SlowConsumerBlockWithTimeout waits for callback queue space up to SlowConsumerBlockTimeout.
	SlowConsumerBlockWithTimeout SlowConsumerPolicy = "block_with_timeout"
)

type State 

type State uint32

State represents the client connection state. 

const (
	StateDisconnected State = iota
	StateConnecting
	StateConnected
	StateReconnecting
	StateDisconnecting
	StateClosed
)

func (State) String 

func (s State) String() string

String returns the state name.

type SubscribeOption 

type SubscribeOption struct {
	// Topic is the topic filter to subscribe to.
	Topic string

	// QoS is the maximum QoS level the client wishes to receive.
	QoS byte

	// NoLocal prevents the client from receiving messages it published itself.
	// Only applies to MQTT 5.0. Default is false.
	NoLocal bool

	// RetainAsPublished preserves the RETAIN flag as it was set by the publisher.
	// If false, the server clears the RETAIN flag before forwarding.
	// Only applies to MQTT 5.0. Default is false.
	RetainAsPublished bool

	// RetainHandling controls when retained messages are sent:
	//   0 = Send retained messages at subscription time (default)
	//   1 = Send retained messages only if this is a new subscription
	//   2 = Don't send retained messages at all
	// Only applies to MQTT 5.0. Default is 0.
	RetainHandling byte

	// SubscriptionID is a numeric identifier for this subscription.
	// When the server sends a PUBLISH for this subscription, it will include this ID.
	// This helps the client determine which subscription(s) triggered the message.
	// 0 means no subscription identifier.
	// Only applies to MQTT 5.0. Default is 0.
	SubscriptionID uint32
}

SubscribeOption represents subscription options for MQTT 5.0.

func NewSubscribeOption 

func NewSubscribeOption(topic string, qos byte) *SubscribeOption

NewSubscribeOption creates a basic subscribe option with just topic and QoS.

func (*SubscribeOption) SetNoLocal 

func (o *SubscribeOption) SetNoLocal(noLocal bool) *SubscribeOption

SetNoLocal sets the NoLocal flag.

func (*SubscribeOption) SetRetainAsPublished 

func (o *SubscribeOption) SetRetainAsPublished(retain bool) *SubscribeOption

SetRetainAsPublished sets the RetainAsPublished flag.

func (*SubscribeOption) SetRetainHandling 

func (o *SubscribeOption) SetRetainHandling(handling byte) *SubscribeOption

SetRetainHandling sets the retain handling option (0, 1, or 2).

func (*SubscribeOption) SetSubscriptionID 

func (o *SubscribeOption) SetSubscriptionID(id uint32) *SubscribeOption

SetSubscriptionID sets the subscription identifier. The ID will be included in PUBLISH packets for messages matching this subscription.

type SubscribeToken 

type SubscribeToken struct {
	ReturnCodes []byte
	// contains filtered or unexported fields
}

SubscribeToken is returned by Subscribe operations.

func (SubscribeToken) Done 

func (t SubscribeToken) Done() <-chan struct{}

Done returns a channel that closes when the operation completes.

func (SubscribeToken) Error 

func (t SubscribeToken) Error() error

Error returns the operation error (may be nil).

func (SubscribeToken) Wait 

func (t SubscribeToken) Wait() error

Wait blocks until the operation completes.

func (SubscribeToken) WaitTimeout 

func (t SubscribeToken) WaitTimeout(timeout time.Duration) error

WaitTimeout blocks until the operation completes or times out.

type Token 

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

Token represents an asynchronous operation result.

type UnsubscribeToken 

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

UnsubscribeToken is returned by Unsubscribe operations.

func (UnsubscribeToken) Done 

func (t UnsubscribeToken) Done() <-chan struct{}

Done returns a channel that closes when the operation completes.

func (UnsubscribeToken) Error 

func (t UnsubscribeToken) Error() error

Error returns the operation error (may be nil).

func (UnsubscribeToken) Wait 

func (t UnsubscribeToken) Wait() error

Wait blocks until the operation completes.

func (UnsubscribeToken) WaitTimeout 

func (t UnsubscribeToken) WaitTimeout(timeout time.Duration) error

WaitTimeout blocks until the operation completes or times out.

type WillMessage 

type WillMessage struct {
	Topic   string
	Payload []byte
	QoS     byte
	Retain  bool

	// MQTT 5.0 Will Properties
	WillDelayInterval uint32            // Delay before sending will (seconds)
	PayloadFormat     *byte             // 0=bytes, 1=UTF-8
	MessageExpiry     uint32            // Will message lifetime (seconds)
	ContentType       string            // MIME type
	ResponseTopic     string            // Response topic for request/response
	CorrelationData   []byte            // Correlation data for request/response
	UserProperties    map[string]string // User-defined properties
}

WillMessage represents a last will and testament message.

Source Files

View all Source files

Directories

Path Synopsis
store
badger
memory

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.