connect

Documentation

Index

• type IMqttMessageListener
• type MqttConnection
  • func NewMqttConnection() *MqttConnection
  • func (c *MqttConnection) Close(ctx context.Context, correlationId string) error
  • func (c *MqttConnection) Configure(ctx context.Context, config *cconf.ConfigParams)
  • func (c *MqttConnection) CreateQueue() error
  • func (c *MqttConnection) DeleteQueue() error
  • func (c *MqttConnection) GetConnection() mqtt.Client
  • func (c *MqttConnection) IsOpen() bool
  • func (c *MqttConnection) Open(ctx context.Context, correlationId string) error
  • func (c *MqttConnection) Publish(ctx context.Context, topic string, qos byte, retained bool, data []byte) error
  • func (c *MqttConnection) ReadQueueNames() ([]string, error)
  • func (c *MqttConnection) SetReferences(ctx context.Context, references cref.IReferences)
  • func (c *MqttConnection) Subscribe(ctx context.Context, topic string, qos byte, listener IMqttMessageListener) error
  • func (c *MqttConnection) Unsubscribe(ctx context.Context, topic string, listener IMqttMessageListener) error
• type MqttConnectionResolver
  • func NewMqttConnectionResolver() *MqttConnectionResolver
  • func (c *MqttConnectionResolver) Compose(correlationId string, connections []*ccon.ConnectionParams, ...) (*cconf.ConfigParams, error)
  • func (c *MqttConnectionResolver) Configure(ctx context.Context, config *cconf.ConfigParams)
  • func (c *MqttConnectionResolver) Resolve(correlationId string) (*cconf.ConfigParams, error)
  • func (c *MqttConnectionResolver) SetReferences(ctx context.Context, references cref.IReferences)
• type MqttSubscription

Types

type IMqttMessageListener

type IMqttMessageListener interface {
	OnMessage(message mqtt.Message)
}

type MqttConnection

type MqttConnection struct {

	// The logger.
	Logger *clog.CompositeLogger
	// The connection resolver.
	ConnectionResolver *MqttConnectionResolver
	// The configuration options.
	Options *cconf.ConfigParams

	// The MQTT connection object.
	Connection mqtt.Client
	// contains filtered or unexported fields
}

MQTT connection using plain driver. By defining a connection and sharing it through multiple message queues you can reduce number of used connections.

Configuration parameters

• client_id: (optional) name of the client id
• connection(s):
• discovery_key: (optional) a key to retrieve the connection from [[https://pip-services3-node.github.io/pip-services3-components-node/interfaces/connect.idiscovery.html IDiscovery]]
• host: host name or IP address
• port: port number
• uri: resource URI or connection string with all parameters in it
• credential(s):
• store_key: (optional) a key to retrieve the credentials from [[https://pip-services3-node.github.io/pip-services3-components-node/interfaces/auth.icredentialstore.html ICredentialStore]]
• username: user name
• password: user password
• options:
• retry_connect: (optional) turns on/off automated reconnect when connection is log (default: true)
• connect_timeout: (optional) number of milliseconds to wait for connection (default: 30000)
• reconnect_timeout: (optional) number of milliseconds to wait on each reconnection attempt (default: 1000)
• keepalive_timeout: (optional) number of milliseconds to ping broker while inactive (default: 3000)

References

• \*:logger:\*:\*:1.0 (optional) ILogger components to pass log messages
• \*:discovery:\*:\*:1.0 (optional) IDiscovery services
• \*:credential-store:\*:\*:1.0 (optional) Credential stores to resolve credentials

func NewMqttConnection

func NewMqttConnection() *MqttConnection

NewMqttConnection creates a new instance of the connection component.

func (*MqttConnection) Close

func (c *MqttConnection) Close(ctx context.Context, correlationId string) error

Closes component and frees used resources. Parameters:

• ctx context.Context operation context.
• correlationId (optional) transaction id to trace execution through call chain.

Return error or nil no errors occured

func (*MqttConnection) Configure

func (c *MqttConnection) Configure(ctx context.Context, config *cconf.ConfigParams)

Configures component by passing configuration parameters. Parameters:

• ctx context.Context operation context.
• config configuration parameters to be set.

func (*MqttConnection) CreateQueue

func (c *MqttConnection) CreateQueue() error

func (*MqttConnection) DeleteQueue

func (c *MqttConnection) DeleteQueue() error

func (*MqttConnection) GetConnection

func (c *MqttConnection) GetConnection() mqtt.Client

func (*MqttConnection) IsOpen

func (c *MqttConnection) IsOpen() bool

Checks if the component is opened. Returns true if the component has been opened and false otherwise.

func (*MqttConnection) Open

func (c *MqttConnection) Open(ctx context.Context, correlationId string) error

Opens the component. Parameters:

• ctx context.Context operation context.
• correlationId (optional) transaction id to trace execution through call chain.
• Return error or nil no errors occured.

func (*MqttConnection) Publish

func (c *MqttConnection) Publish(ctx context.Context, topic string, qos byte, retained bool, data []byte) error

Publish a message to a specified topic

Parameters:

• ctx context.Context operation context.
• topic a topic name
• qos quality of service (QOS) for the message
• retained retained flag for the message
• data a message to be published

Returns: error or nil for success

func (*MqttConnection) ReadQueueNames

func (c *MqttConnection) ReadQueueNames() ([]string, error)

func (*MqttConnection) SetReferences

func (c *MqttConnection) SetReferences(ctx context.Context, references cref.IReferences)

Sets references to dependent components. Parameters:

• ctx context.Context operation context.
• references references to locate the component dependencies.

func (*MqttConnection) Subscribe

func (c *MqttConnection) Subscribe(ctx context.Context, topic string, qos byte, listener IMqttMessageListener) error

Subscribe to a topic

Parameters: Parameters:

• ctx context.Context operation context.
• topic a topic name
• qos quality of service (QOS) for the subscription
• listener a message listener

Returns: err or nil for success

func (*MqttConnection) Unsubscribe

func (c *MqttConnection) Unsubscribe(ctx context.Context, topic string, listener IMqttMessageListener) error

Unsubscribe from a previously subscribed topic topic

Parameters:

• ctx context.Context operation context.
• topic a topic name
• qos quality of service (QOS) for the subscription
• listener a message listener

Returns: err or nil for success

type MqttConnectionResolver

type MqttConnectionResolver struct {
	ConnectionResolver *ccon.ConnectionResolver
	CredentialResolver *cauth.CredentialResolver
}

MqttConnectionResolver helper class that resolves Mqtt connection and credential parameters, validates them and generates connection options.

Configuration parameters:

• connection(s):
• discovery_key: (optional) a key to retrieve the connection from IDiscovery
• host: host name or IP address
• port: port number
• uri: resource URI or connection string with all parameters in it
• credential(s):
• store_key: (optional) a key to retrieve the credentials from ICredentialStore
• username: user name
• password: user password

References:

• *:discovery:*:*:1.0 (optional) IDiscovery services to resolve connections
• *:credential-store:*:*:1.0 (optional) Credential stores to resolve credentials

func NewMqttConnectionResolver

func NewMqttConnectionResolver() *MqttConnectionResolver

func (*MqttConnectionResolver) Compose

func (c *MqttConnectionResolver) Compose(correlationId string, connections []*ccon.ConnectionParams,
	credential *cauth.CredentialParams) (*cconf.ConfigParams, error)

Compose method are composes Mqtt connection options from connection and credential parameters. Parameters:

• ctx context.Context operation context.
• correlationId string (optional) transaction id to trace execution through call chain.
• connection *ccon.ConnectionParams connection parameters
• credential *cauth.CredentialParams credential parameters

Returns: options *cconf.ConfigParams, err error resolved options or error.

func (*MqttConnectionResolver) Configure

func (c *MqttConnectionResolver) Configure(ctx context.Context, config *cconf.ConfigParams)

Configure are configures component by passing configuration parameters. Parameters:

• ctx context.Context operation context.
• config *cconf.ConfigParams

configuration parameters to be set.

func (*MqttConnectionResolver) Resolve

func (c *MqttConnectionResolver) Resolve(correlationId string) (*cconf.ConfigParams, error)

Resolves MQTT connection options from connection and credential parameters. Parameters:

• ctx context.Context operation context.
• correlationId string (optional) transaction id to trace execution through call chain.

Returns options *cconf.ConfigParams, err error receives resolved options or error.

func (*MqttConnectionResolver) SetReferences

func (c *MqttConnectionResolver) SetReferences(ctx context.Context, references cref.IReferences)

SetReferences are sets references to dependent components. Parameters:

• ctx context.Context operation context.
• references cref.IReferences references to locate the component dependencies.

type MqttSubscription

type MqttSubscription struct {
	Topic    string
	Qos      byte
	Listener IMqttMessageListener
	Skip     int32
}