mqtt

Documentation

Index

• func NewAgentOptions(mqttOptions *MQTTOptions, clusterName, agentID string) *options.CloudEventsAgentOptions
• func NewSourceOptions(mqttOptions *MQTTOptions, clientID, sourceID string) *options.CloudEventsSourceOptions
• type MQTTConfig
• type MQTTDialer
  • func (d *MQTTDialer) Close() error
  • func (d *MQTTDialer) Dial() (net.Conn, error)
• type MQTTOptions
  • func BuildMQTTOptionsFromFlags(configPath string) (*MQTTOptions, error)
  • func (o *MQTTOptions) GetCloudEventsProtocol(ctx context.Context, clientID string, errorHandler func(error), ...) (options.CloudEventsProtocol, error)
  • func (o *MQTTOptions) GetMQTTConnectOption(clientID string) *paho.Connect
• type PubTopic
• type TopicKey

Functions

func NewAgentOptions

func NewAgentOptions(mqttOptions *MQTTOptions, clusterName, agentID string) *options.CloudEventsAgentOptions

func NewSourceOptions

func NewSourceOptions(mqttOptions *MQTTOptions, clientID, sourceID string) *options.CloudEventsSourceOptions

Types

type MQTTConfig

type MQTTConfig struct {
	// BrokerHost is the host of the MQTT broker (hostname:port).
	BrokerHost string `json:"brokerHost" yaml:"brokerHost"`

	// Username is the username for basic authentication to connect the MQTT broker.
	Username string `json:"username,omitempty" yaml:"username,omitempty"`
	// Password is the password for basic authentication to connect the MQTT broker.
	Password string `json:"password,omitempty" yaml:"password,omitempty"`

	// CAFile is the file path to a cert file for the MQTT broker certificate authority.
	CAFile string `json:"caFile,omitempty" yaml:"caFile,omitempty"`
	// ClientCertFile is the file path to a client cert file for TLS.
	ClientCertFile string `json:"clientCertFile,omitempty" yaml:"clientCertFile,omitempty"`
	// ClientKeyFile is the file path to a client key file for TLS.
	ClientKeyFile string `json:"clientKeyFile,omitempty" yaml:"clientKeyFile,omitempty"`

	// KeepAlive is the keep alive time in seconds for MQTT clients, by default is 60s
	KeepAlive *uint16 `json:"keepAlive,omitempty" yaml:"keepAlive,omitempty"`

	// DialTimeout is the timeout when establishing a MQTT TCP connection, by default is 60s
	DialTimeout *time.Duration `json:"dialTimeout,omitempty" yaml:"dialTimeout,omitempty"`

	// PubQoS is the QoS for publish, by default is 1
	PubQoS *int `json:"pubQoS,omitempty" yaml:"pubQoS,omitempty"`
	// SubQoS is the Qos for subscribe, by default is 1
	SubQoS *int `json:"subQoS,omitempty" yaml:"subQoS,omitempty"`

	// Topics are MQTT topics for resource spec, status and resync.
	Topics *types.Topics `json:"topics,omitempty" yaml:"topics,omitempty"`
}

MQTTConfig holds the information needed to build connect to MQTT broker as a given user.

type MQTTDialer

type MQTTDialer struct {
	TLSConfig  *tls.Config
	BrokerHost string
	Timeout    time.Duration
	// contains filtered or unexported fields
}

func (*MQTTDialer) Close

func (d *MQTTDialer) Close() error

func (*MQTTDialer) Dial

func (d *MQTTDialer) Dial() (net.Conn, error)

type MQTTOptions

type MQTTOptions struct {
	Topics    types.Topics
	Username  string
	Password  string
	KeepAlive uint16
	PubQoS    int
	SubQoS    int

	Dialer *MQTTDialer
}

MQTTOptions holds the options that are used to build MQTT client.

func BuildMQTTOptionsFromFlags

func BuildMQTTOptionsFromFlags(configPath string) (*MQTTOptions, error)

BuildMQTTOptionsFromFlags builds configs from a config filepath.

func (*MQTTOptions) GetCloudEventsProtocol

func (o *MQTTOptions) GetCloudEventsProtocol(
	ctx context.Context,
	clientID string,
	errorHandler func(error),
	clientOpts ...cloudeventsmqtt.Option,
) (options.CloudEventsProtocol, error)

func (*MQTTOptions) GetMQTTConnectOption

func (o *MQTTOptions) GetMQTTConnectOption(clientID string) *paho.Connect

type PubTopic

type PubTopic string

type TopicKey

type TopicKey string

const (
	MQTT_SOURCE_PUB_TOPIC_KEY TopicKey = "mqtt_source_pub_topic"
	MQTT_AGENT_PUB_TOPIC_KEY  TopicKey = "mqtt_agent_pub_topic"
)