Documentation
¶
Overview ¶
Package mqtt provides a bidirectional bridge between a DDS participant and an MQTT broker. Samples published on either side are forwarded to the other using a configurable topic-mapping and QoS-mapping policy.
Dependency-free design ¶
go-DDS does not import an MQTT client library. Instead this package defines the MQTTClient interface which callers implement using their preferred client (e.g., github.com/eclipse/paho.mqtt.golang). This keeps go-DDS's go.mod free of external dependencies while still letting callers use any standards-compliant MQTT 3.1.1 / 5.0 client.
Quick start ¶
client := /* paho.NewClient(...) or any MQTTClient adapter */
ddsParticipant, _ := mock.New(dds.Domain(0))
bridge, err := mqtt.NewBridge(ddsParticipant, client, mqtt.Options{
TopicMap: mqtt.PrefixMap("vehicles/", ""),
QoSMap: mqtt.DefaultQoSMap,
})
if err != nil {
log.Fatal(err)
}
defer bridge.Close()
Topic mapping ¶
By default the bridge uses a 1:1 topic name mapping (the DDS topic and the MQTT topic are identical). Provide a TopicMapper to rename topics as they cross the boundary in either direction.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Bridge ¶
type Bridge struct {
// contains filtered or unexported fields
}
Bridge forwards samples bidirectionally between a DDS participant and an MQTT client. It is safe for concurrent use.
func NewBridge ¶
func NewBridge(p dds.Participant, client MQTTClient, opts Options) (*Bridge, error)
NewBridge creates a Bridge and starts forwarding between DDS and MQTT. The caller is responsible for closing the bridge when done.
type MQTTClient ¶
type MQTTClient interface {
// Publish sends payload to an MQTT topic.
// qos: 0=at-most-once, 1=at-least-once, 2=exactly-once.
// retained: true means the broker stores the last message for late joiners.
Publish(topic string, qos byte, retained bool, payload []byte) Token
// Subscribe registers handler for all messages on an MQTT topic pattern.
// The pattern may use MQTT wildcards (+ and #).
Subscribe(topic string, qos byte, handler MessageHandler) Token
// Unsubscribe cancels a previous subscription.
Unsubscribe(topics ...string) Token
// IsConnected reports whether the client is currently connected.
IsConnected() bool
}
MQTTClient is the interface that must be implemented to connect this bridge to an MQTT broker. github.com/eclipse/paho.mqtt.golang satisfies this interface with a thin adapter (see the adapters example in the package docs).
type MessageHandler ¶
MessageHandler is called for each message received from the MQTT broker. topic is the MQTT topic; payload is the raw message bytes.
type Options ¶
type Options struct {
// TopicMap controls how topic names are translated. Default: IdentityMap.
TopicMap TopicMapper
// QoSMap controls how QoS policies are translated. Default: DefaultQoSMap.
QoSMap QoSMapper
// DDSTopics is the set of DDS topics the bridge subscribes to and forwards
// to MQTT. An empty slice means the bridge listens on all topics via a
// wildcard subscriber ("#" on the DDS side).
//
// Note: the DDS participant must support wildcard subscriptions if this
// field is left empty.
DDSTopics []string
// MQTTTopics is the set of MQTT topic patterns the bridge subscribes to and
// forwards to DDS. An empty slice defaults to "#" (all topics).
MQTTTopics []string
// DDSQoS is the QoS used when the bridge creates DDS publishers for
// messages arriving from MQTT. Default: dds.DefaultQoS.
DDSQoS dds.QoS
}
Options configures the bridge.
type QoSMapper ¶
type QoSMapper interface {
// DDSToMQTT returns the MQTT QoS byte (0/1/2) and the retained flag for a
// given DDS QoS.
DDSToMQTT(q dds.QoS) (qos byte, retained bool)
// MQTTToDDS returns the DDS QoS appropriate for a message received with the
// given MQTT QoS and retained flag.
MQTTToDDS(qos byte, retained bool) dds.QoS
}
QoSMapper converts between DDS QoS and MQTT QoS levels.
var DefaultQoSMap QoSMapper = defaultQoSMapper{}
DefaultQoSMap is the built-in QoSMapper:
- DDS BestEffort → MQTT QoS 0, not retained
- DDS Reliable / TransientLocal → MQTT QoS 1, retained
- MQTT QoS 0 → DDS BestEffort + Volatile
- MQTT QoS 1/2 → DDS Reliable + TransientLocal
type Token ¶
type Token interface {
// Wait blocks until the operation completes.
Wait() bool
// Error returns the error (if any) from the completed operation.
Error() error
}
Token represents an asynchronous MQTT operation. Implement it to wrap your client library's token / future type.
type TopicMapper ¶
type TopicMapper interface {
// DDSToMQTT translates a DDS topic name to an MQTT topic.
DDSToMQTT(ddsTopic string) string
// MQTTToDDS translates an MQTT topic to a DDS topic name.
// Returns "", false to discard the message (no matching DDS topic).
MQTTToDDS(mqttTopic string) (string, bool)
}
TopicMapper converts topic names as they cross the DDS↔MQTT boundary.
var IdentityMap TopicMapper = identityMapper{}
IdentityMap is the default TopicMapper: DDS and MQTT use the same topic names.
func PrefixMap ¶
func PrefixMap(ddsPrefix, mqttPrefix string) TopicMapper
PrefixMap returns a TopicMapper that prepends ddsPrefix to DDS topics and mqttPrefix to MQTT topics. Stripping happens on the opposite side.
Example — DDS "signals/rpm" ↔ MQTT "vehicle/signals/rpm":
PrefixMap("signals/", "vehicle/signals/")
Source Files