mqtt

Documentation

Index

• Constants
• Variables
• type Client
• type Config
  • func NewConfig() Config
  • func (c Config) Equal(o Config) bool
  • func (c Config) NewClient() (Client, error)
  • func (c *Config) SetNewClientF(fn func(c Config) (Client, error))
  • func (c Config) Validate() error
• type Configs
  • func (cs Configs) Validate() error
• type Diagnostic
• type HandlerConfig
• type PahoClient
  • func (p *PahoClient) Connect() error
  • func (p *PahoClient) Disconnect()
  • func (p *PahoClient) Publish(topic string, qos QoSLevel, retained bool, message []byte) error
• type QoSLevel
  • func (q QoSLevel) MarshalText() (text []byte, err error)
  • func (q *QoSLevel) UnmarshalText(text []byte) error
• type Service
  • func NewService(cs Configs, d Diagnostic) (*Service, error)
  • func (s *Service) Alert(brokerName, topic string, qos QoSLevel, retained bool, message string) error
  • func (s *Service) Close() error
  • func (s *Service) Handler(c HandlerConfig, ctx ...keyvalue.T) (alert.Handler, error)
  • func (s *Service) Open() error
  • func (s *Service) Test(o interface{}) error
  • func (s *Service) TestOptions() interface{}
  • func (s *Service) Update(newConfigs []interface{}) error

Constants

const DefaultQuiesceTimeout time.Duration = 250 * time.Millisecond

DefaultQuiesceTimeout is the duration the client will wait for outstanding messages to be published before forcing a disconnection

Variables

var (
	ErrInvalidQoS = errors.New("invalid QoS")
)

Types

type Client

type Client interface {
	Connect() error
	Disconnect()
	Publish(topic string, qos QoSLevel, retained bool, message []byte) error
}

Client describes an immutable MQTT client, designed to accommodate the incongruencies between real clients and mock clients.

type Config

type Config struct {
	// Enabled indicates whether the service should be enabled
	Enabled bool   `toml:"enabled" override:"enabled"`
	Name    string `toml:"name" override:"name"`
	Default bool   `toml:"default" override:"default"`
	// URL of the MQTT Broker.
	// Valid URLs include tcp://host:port, ws://host:port or ssl://host:port.
	// If using ssl://host:port, one must also specify the SSL configuration options.
	URL string `toml:"url" override:"url"`

	// Path to CA file
	SSLCA string `toml:"ssl-ca" override:"ssl-ca"`
	// Path to host cert file
	SSLCert string `toml:"ssl-cert" override:"ssl-cert"`
	// Path to cert key file
	SSLKey string `toml:"ssl-key" override:"ssl-key"`
	// Use SSL but skip chain & host verification
	InsecureSkipVerify bool `toml:"insecure-skip-verify" override:"insecure-skip-verify"`

	// ClientID is the unique ID advertised to the MQTT broker from this client.
	// Defaults to Name if empty.
	ClientID string `toml:"client-id" override:"client-id"`
	Username string `toml:"username" override:"username"`
	Password string `toml:"password" override:"password,redact"`
	// contains filtered or unexported fields
}

func NewConfig

func NewConfig() Config

func (Config) Equal

func (c Config) Equal(o Config) bool

func (Config) NewClient

func (c Config) NewClient() (Client, error)

NewClient creates a new client based off this configuration.

func (*Config) SetNewClientF

func (c *Config) SetNewClientF(fn func(c Config) (Client, error))

SetNewClientF sets the newClientF on a Config. It is used exclusively for testing.

func (Config) Validate

func (c Config) Validate() error

type Configs

type Configs []Config

func (Configs) Validate

func (cs Configs) Validate() error

Validate calls config.Validate for each element in Configs

type Diagnostic

type Diagnostic interface {
	WithContext(ctx ...keyvalue.T) Diagnostic
	Error(msg string, err error)
	CreatingAlertHandler(c HandlerConfig)
	HandlingEvent()
}

type HandlerConfig

type HandlerConfig struct {
	BrokerName    string         `mapstructure:"broker-name"`
	Topic         string         `mapstructure:"topic"`
	QoS           QoSLevel       `mapstructure:"qos"`
	Retained      bool           `mapstructure:"retained"`
	TopicTemplate *text.Template `mapstructure:"topic-template"`
}

type PahoClient

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

func (*PahoClient) Connect

func (p *PahoClient) Connect() error

func (*PahoClient) Disconnect

func (p *PahoClient) Disconnect()

func (*PahoClient) Publish

func (p *PahoClient) Publish(topic string, qos QoSLevel, retained bool, message []byte) error

type QoSLevel

type QoSLevel byte

QoSLevel indicates the quality of service for messages delivered to a broker.

const (
	// best effort delivery. "fire and forget"
	AtMostOnce QoSLevel = iota
	// guarantees delivery to at least one receiver. May deliver multiple times.
	AtLeastOnce
	// guarantees delivery only once. Safest and slowest.
	ExactlyOnce
)

func (QoSLevel) MarshalText

func (q QoSLevel) MarshalText() (text []byte, err error)

func (*QoSLevel) UnmarshalText

func (q *QoSLevel) UnmarshalText(text []byte) error

type Service

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

func NewService

func NewService(cs Configs, d Diagnostic) (*Service, error)

func (*Service) Alert

func (s *Service) Alert(brokerName, topic string, qos QoSLevel, retained bool, message string) error

func (*Service) Close

func (s *Service) Close() error

func (*Service) Handler

func (s *Service) Handler(c HandlerConfig, ctx ...keyvalue.T) (alert.Handler, error)

func (*Service) Open

func (s *Service) Open() error

func (*Service) Test

func (s *Service) Test(o interface{}) error

func (*Service) TestOptions

func (s *Service) TestOptions() interface{}

func (*Service) Update

func (s *Service) Update(newConfigs []interface{}) error