mqtt

Documentation

Index

• func DetachConnection(clientId string, subscribedTopic string)
• func GetSink() api.Sink
• func GetSource() api.Source
• type AdConf
• type Conf
• type Connection
  • func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*Connection, error)
  • func GetConnection(ctx api.StreamContext, props map[string]any) (*Connection, error)
  • func (conn *Connection) Close()
  • func (conn *Connection) GetClientId() string
  • func (conn *Connection) Ping() error
  • func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error
• type ConnectionConfig
• type Sink
  • func (ms *Sink) Close(ctx api.StreamContext) error
  • func (ms *Sink) Collect(ctx api.StreamContext, item []byte) error
  • func (ms *Sink) Connect(ctx api.StreamContext) error
  • func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error
• type SourceConnector
  • func (ms *SourceConnector) Close(ctx api.StreamContext) error
  • func (ms *SourceConnector) Connect(ctx api.StreamContext) error
  • func (ms *SourceConnector) Ping(props map[string]interface{}) error
  • func (ms *SourceConnector) Provision(ctx api.StreamContext, props map[string]any) error
  • func (ms *SourceConnector) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error
• type SubscriptionInfo

Functions

func DetachConnection

func DetachConnection(clientId string, subscribedTopic string)

func GetSink

func GetSink() api.Sink

func GetSource

func GetSource() api.Source

Types

type AdConf

type AdConf struct {
	Tpc      string `json:"topic"`
	Qos      byte   `json:"qos"`
	Retained bool   `json:"retained"`

	ResendTopic string `json:"resendDestination"`
}

AdConf is the advanced configuration for the mqtt sink

type Conf

type Conf struct {
	Topic string `json:"datasource"`
	Qos   int    `json:"qos"`
}

type Connection

type Connection struct {
	pahoMqtt.Client
	// contains filtered or unexported fields
}

func CreateClient

func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*Connection, error)

CreateClient creates a new mqtt client. It is anonymous and does not require a name.

func GetConnection

func GetConnection(ctx api.StreamContext, props map[string]any) (*Connection, error)

func (*Connection) Close

func (conn *Connection) Close()

func (*Connection) GetClientId

func (conn *Connection) GetClientId() string

func (*Connection) Ping

func (conn *Connection) Ping() error

func (*Connection) Subscribe

func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error

type ConnectionConfig

type ConnectionConfig struct {
	Server   string `json:"server"`
	PVersion string `json:"protocolVersion"`
	ClientId string `json:"clientid"`
	Uname    string `json:"username"`
	Password string `json:"password"`
	// contains filtered or unexported fields
}

type Sink

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

func (*Sink) Close

func (ms *Sink) Close(ctx api.StreamContext) error

func (*Sink) Collect

func (ms *Sink) Collect(ctx api.StreamContext, item []byte) error

func (*Sink) Connect

func (ms *Sink) Connect(ctx api.StreamContext) error

func (*Sink) Provision

func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error

type SourceConnector

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

SourceConnector is the connector for mqtt source When sharing the same connection, each topic will have one single sourceConnector as the shared source node

func (*SourceConnector) Close

func (ms *SourceConnector) Close(ctx api.StreamContext) error

func (*SourceConnector) Connect

func (ms *SourceConnector) Connect(ctx api.StreamContext) error

func (*SourceConnector) Ping

func (ms *SourceConnector) Ping(props map[string]interface{}) error

func (*SourceConnector) Provision

func (ms *SourceConnector) Provision(ctx api.StreamContext, props map[string]any) error

func (*SourceConnector) Subscribe

func (ms *SourceConnector) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error

Subscribe is a one time only operation for source. It connects to the mqtt broker and subscribe to the topic Run open before subscribe

type SubscriptionInfo

type SubscriptionInfo struct {
	Qos        byte
	Handler    pahoMqtt.MessageHandler
	ErrHandler func(error)
}