mqtttransport

Documentation

Index

• Constants
• func MakeTopic(msgType, agentID, thingID, name string, clientID string) string
• func SplitTopic(topic string) (msgType, agentID, thingID, name string, senderID string, err error)
• type Handler
  • func NewHandler(ctx context.Context, c *paho.Client) (*Handler, error)
  • func (h *Handler) Request(ctx context.Context, pb *paho.Publish) (*paho.Publish, error)
• type MqttHubTransport
  • func NewMqttTransport(fullURL string, clientID string, caCert *x509.Certificate) *MqttHubTransport
  • func (hc *MqttHubTransport) AddressTokens() (sep string, wc string, rem string)
  • func (ht *MqttHubTransport) ConnectWithConn(credentials string, conn net.Conn) error
  • func (ht *MqttHubTransport) ConnectWithPassword(password string) error
  • func (ht *MqttHubTransport) ConnectWithToken(kp string, jwtToken string) error
  • func (ht *MqttHubTransport) CreateKeyPair() (serializedKP string, serializedPub string)
  • func (ht *MqttHubTransport) Disconnect()
  • func (ht *MqttHubTransport) ParseResponse(data []byte, resp interface{}) error
  • func (ht *MqttHubTransport) Pub(topic string, payload []byte) (err error)
  • func (ht *MqttHubTransport) PubRequest(topic string, payload []byte) (resp []byte, err error)
  • func (ht *MqttHubTransport) Sub(topic string, cb func(topic string, msg []byte)) (transports.ISubscription, error)
  • func (ht *MqttHubTransport) SubRequest(topic string, cb func(topic string, payload []byte) (reply []byte, err error)) (transports.ISubscription, error)
• type PahoSubscription
  • func (sub *PahoSubscription) Unsubscribe() error

Constants

const InboxTopicFormat = "_INBOX/%s"

InboxTopicFormat is the INBOX subscription topic used by the client Used to send replies to requests.

const MqttInMemUDSProd = "@/MqttInMemUDSProd" // production UDS name

const MqttInMemUDSTest = "@/MqttInMemUDSTest" // test server UDS name

Functions

func MakeTopic

func MakeTopic(msgType, agentID, thingID, name string, clientID string) string

MakeTopic creates a mqtt message topic optionally with wildcards This uses the hiveot topic format: {msgType}/{deviceID}/{thingID}/{name}[/{clientID}]

msgType is the message type: "event", "action", "config" or "rpc".
agentID is the device or service being addressed. Use "" for wildcard
thingID is the ID of the thing managed by the publisher. Use "" for wildcard
name is the event or action name. Use "" for wildcard.
clientID is the login ID of the sender. Use "" for subscribe.

func SplitTopic

func SplitTopic(topic string) (msgType, agentID, thingID, name string, senderID string, err error)

SplitTopic separates a topic into its components

topic is a hiveot mqtt topic. eg: msgType/things/deviceID/thingID/name/clientID

msgType of "things", "rpc" or "_INBOX"
agentID is the device or service that is being addressed.
thingID is the thing of the topic, or capability for services.
name is the event or action name
senderID is the client publishing the request.

Types

type Handler

type Handler struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Handler is the struct providing a request/response functionality for the paho MQTT v5 client

func NewHandler

func NewHandler(ctx context.Context, c *paho.Client) (*Handler, error)

func (*Handler) Request

func (h *Handler) Request(ctx context.Context, pb *paho.Publish) (*paho.Publish, error)

type MqttHubTransport

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

MqttHubTransport manages the hub server connection with hub event and action messaging This implements the IHubClient interface. This implementation is based on the Mqtt messaging system.

func NewMqttTransport

func NewMqttTransport(fullURL string, clientID string, caCert *x509.Certificate) *MqttHubTransport

NewMqttTransport creates a new instance of the mqtt client.

fullURL is the url with schema. If omitted this uses the in-memory UDS address, which only works with ConnectWithToken.

fullURL of broker to connect to, starting with "tls://", "wss://", "unix://"
clientID to connect as
keyPair with previously saved serialized public/private key pair, or "" to create one
caCert of the server to validate the server or nil to not check the server cert

func (*MqttHubTransport) AddressTokens

func (hc *MqttHubTransport) AddressTokens() (sep string, wc string, rem string)

AddressTokens returns the address separator and wildcards

func (*MqttHubTransport) ConnectWithConn

func (ht *MqttHubTransport) ConnectWithConn(credentials string, conn net.Conn) error

ConnectWithConn connects to a mqtt broker using the pre-established network connection This allows creating connections with any available means including tcp/tls/wss/uds/pipes

credentials is either the password or a signed JWT token
conn network connection to utilize

func (*MqttHubTransport) ConnectWithPassword

func (ht *MqttHubTransport) ConnectWithPassword(password string) error

ConnectWithPassword connects to the Hub server using a login ID and password.

func (*MqttHubTransport) ConnectWithToken

func (ht *MqttHubTransport) ConnectWithToken(kp string, jwtToken string) error

ConnectWithToken connects to the Hub server using a user JWT credentials secret The token clientID must match that of the client A private key might be required in future. This supports UDS connections with @/path or unix://@/path

kp is a serialized public/private key-pair of this client
jwtToken is the token obtained with login or refresh.

func (*MqttHubTransport) CreateKeyPair

func (ht *MqttHubTransport) CreateKeyPair() (serializedKP string, serializedPub string)

CreateKeyPair returns a new set of serialized public/private key pair

func (*MqttHubTransport) Disconnect

func (ht *MqttHubTransport) Disconnect()

Disconnect from the MQTT broker and unsubscribe from all topics and set device state to disconnected

func (*MqttHubTransport) ParseResponse

func (ht *MqttHubTransport) ParseResponse(data []byte, resp interface{}) error

ParseResponse helper message to parse response and check for errors

func (*MqttHubTransport) Pub

func (ht *MqttHubTransport) Pub(topic string, payload []byte) (err error)

Pub publishes a message and returns

func (*MqttHubTransport) PubRequest

func (ht *MqttHubTransport) PubRequest(topic string, payload []byte) (resp []byte, err error)

PubRequest publishes a request message and waits for an answer or until timeout

func (*MqttHubTransport) Sub

func (ht *MqttHubTransport) Sub(topic string, cb func(topic string, msg []byte)) (transports.ISubscription, error)

Sub subscribes to a topic

func (*MqttHubTransport) SubRequest

func (ht *MqttHubTransport) SubRequest(
	topic string, cb func(topic string, payload []byte) (reply []byte, err error)) (
	transports.ISubscription, error)

SubRequest subscribes to a requests and sends a response Intended for actions, config and rpc requests

type PahoSubscription

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

PahoSubscription implements hubclient.ISubscription interface

func (*PahoSubscription) Unsubscribe

func (sub *PahoSubscription) Unsubscribe() error

Unsubscribe from the subscription