Documentation
¶
Overview ¶
Package client manages a single autopaho.ConnectionManager shared by all publishers and subscribers within a vinculum MQTT client block.
Usage ¶
Create a client, add publishers and subscribers, then start:
c, err := client.NewClient(client.ClientConfig{
ServerURLs: []*url.URL{mustURL("mqtt://localhost:1883")},
ClientID: "my-client",
KeepAlive: 30 * time.Second,
})
c.AddPublisher(pub)
c.AddSubscriber(sub)
if err := c.Start(ctx); err != nil { ... }
defer c.Stop(ctx)
Start blocks until the first connection is established and all subscriptions are registered. Publishers receive their publish function during Start.
Connection lifecycle ¶
The underlying autopaho.ConnectionManager reconnects automatically on disconnect. On each reconnect, broker subscriptions are re-registered. Publisher publish functions remain valid across reconnects.
Use Stop to send a graceful MQTT DISCONNECT and wait for the connection manager to shut down. Stop is safe to call before Start (no-op).
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ClientConfig ¶
type ClientConfig struct {
// ServerURLs — one or more broker URLs (mqtt://, mqtts://, ws://, wss://).
ServerURLs []*url.URL
// ClientID — MQTT client identifier. Must be unique per broker connection.
ClientID string
// KeepAlive — PINGREQ interval. Converted to seconds for the broker.
// Zero uses the autopaho default (no keepalive).
KeepAlive time.Duration
// CleanStart — whether to request a clean session on the initial connection.
CleanStart bool
// SessionExpiryInterval — seconds the broker retains session state after
// disconnect. Zero means the session ends when the connection closes.
SessionExpiryInterval uint32
// TLSConfig — optional TLS configuration for mqtts:// or wss:// connections.
TLSConfig *tls.Config
// Username / Password — optional MQTT credentials.
Username string
Password []byte
// WillMessage — optional Last Will and Testament configuration.
WillMessage *WillConfig
// ReconnectBackoffFunc maps reconnect attempt number to wait duration.
// nil uses autopaho's default constant 10s backoff.
ReconnectBackoffFunc func(attempt int) time.Duration
// OnConnect is called from within OnConnectionUp after subscriptions are
// registered. It runs synchronously; keep it fast.
OnConnect func(ctx context.Context)
// OnDisconnect is called from within OnConnectionDown when the connection
// drops unexpectedly. It runs synchronously; keep it fast.
OnDisconnect func(ctx context.Context)
// MetricsProvider — optional o11y.MetricsProvider for connection metrics.
MetricsProvider o11y.MetricsProvider
// Logger — optional logger.
Logger *zap.Logger
}
ClientConfig holds all configuration for a single MQTT connection. It is populated by the vinculum config layer and passed to NewClient.
type ClientMetrics ¶
type ClientMetrics struct {
// contains filtered or unexported fields
}
ClientMetrics holds the o11y instruments for an MQTTClient. A nil *ClientMetrics is valid and results in no-op recording.
func NewClientMetrics ¶
func NewClientMetrics(provider o11y.MetricsProvider) *ClientMetrics
NewClientMetrics creates a ClientMetrics using the given provider. Returns nil if provider is nil, which is safe to call all methods on.
func (*ClientMetrics) IncrReconnects ¶
func (m *ClientMetrics) IncrReconnects(ctx context.Context)
IncrReconnects increments the reconnection counter.
func (*ClientMetrics) SetConnected ¶
func (m *ClientMetrics) SetConnected(ctx context.Context, up bool)
SetConnected sets the connected gauge to 1 (connected) or 0 (not connected).
type MQTTClient ¶
type MQTTClient struct {
// contains filtered or unexported fields
}
MQTTClient manages a single autopaho.ConnectionManager, wiring zero or more MQTTPublishers and MQTTSubscribers to a shared MQTT connection.
Call AddPublisher / AddSubscriber before Start. Start blocks until the first connection is established and all subscriptions are registered. Stop sends a graceful DISCONNECT and waits for the connection to close.
func NewClient ¶
func NewClient(cfg ClientConfig) (*MQTTClient, error)
NewClient constructs an MQTTClient. Validates that at least one server URL is provided.
func (*MQTTClient) AddPublisher ¶
func (c *MQTTClient) AddPublisher(p *mqttpublisher.MQTTPublisher)
AddPublisher registers a publisher. Must be called before Start. At Start time the publisher receives the shared publish function.
func (*MQTTClient) AddSubscriber ¶
func (c *MQTTClient) AddSubscriber(s *mqttsubscriber.MQTTSubscriber)
AddSubscriber registers a subscriber. Must be called before Start. At Start time the subscriber's broker topics are subscribed to and its HandleMessage is wired into the MQTT message router.
func (*MQTTClient) Start ¶
func (c *MQTTClient) Start(ctx context.Context) error
Start connects to the MQTT broker, registers all subscriptions, injects the publish function into all publishers, and returns. Blocks until the first connection is fully established (OnConnectionUp has run) or ctx is cancelled.
Source Files