mqtt

Documentation

Index

• type Client
  • func NewClient(opts *mqtt.ClientOptions, logger ...*zap.Logger) *Client
  • func (c *Client) Connect() error
  • func (c *Client) Disconnect()
  • func (c *Client) MessageToPostgres(client mqtt.Client, msg mqtt.Message)
  • func (c *Client) Publish(topic string, qos byte, retained bool, payload interface{}) error
  • func (c *Client) Subscribe(topic string, qos byte, callback mqtt.MessageHandler) error
• type ClientOptions
• type PeerMQTT
  • func (p *PeerMQTT) Init(config json.RawMessage, args ...any) error
  • func (p *PeerMQTT) Publish(event logrepl.PostgresCDC) error

Types

type Client

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

Client represents an MQTT client with PostgREST forwarding capabilities.

func NewClient

func NewClient(opts *mqtt.ClientOptions, logger ...*zap.Logger) *Client

NewClient creates a new MQTT client with the given options and logger.

func (*Client) Connect

func (c *Client) Connect() error

Connect establishes a connection to the MQTT broker.

func (*Client) Disconnect

func (c *Client) Disconnect()

Disconnect closes the connection to the MQTT broker.

func (*Client) MessageToPostgres

func (c *Client) MessageToPostgres(client mqtt.Client, msg mqtt.Message)

MessageToPostgres persists an MQTT message to PostgreSQL. It expects the topic to be in the following format: topic: /pgo/TABLE_NAME/OPERATION (insert, update, delete) payload: JSON mosquitto_pub -t /pgo/users/insert -m '{"name":"some1"}'

func (*Client) Publish

func (c *Client) Publish(topic string, qos byte, retained bool, payload interface{}) error

Publish sends a message to the specified MQTT topic.

func (*Client) Subscribe

func (c *Client) Subscribe(topic string, qos byte, callback mqtt.MessageHandler) error

Subscribe registers a callback for messages on the specified MQTT topic.

type ClientOptions

type ClientOptions struct {
	Servers                []*url.URL `json:"servers"`
	ClientID               string     `json:"clientID"`
	Username               string     `json:"username"`
	Password               string     `json:"password"`
	CredentialsProvider    mqtt.CredentialsProvider
	CleanSession           bool
	Order                  bool
	WillEnabled            bool
	WillTopic              string
	WillPayload            []byte
	WillQos                byte
	WillRetained           bool
	ProtocolVersion        uint
	TLSConfig              *tls.Config
	KeepAlive              int64 // Warning: Some brokers may reject connections with Keepalive = 0.
	PingTimeout            time.Duration
	ConnectTimeout         time.Duration
	MaxReconnectInterval   time.Duration
	AutoReconnect          bool
	ConnectRetryInterval   time.Duration
	ConnectRetry           bool
	Store                  mqtt.Store
	DefaultPublishHandler  mqtt.MessageHandler
	OnConnect              mqtt.OnConnectHandler
	OnConnectionLost       mqtt.ConnectionLostHandler
	OnReconnecting         mqtt.ReconnectHandler
	OnConnectAttempt       mqtt.ConnectionAttemptHandler
	WriteTimeout           time.Duration
	MessageChannelDepth    uint
	ResumeSubs             bool
	HTTPHeaders            http.Header
	WebsocketOptions       *mqtt.WebsocketOptions
	MaxResumePubInFlight   int // // 0 = no limit; otherwise this is the maximum simultaneous messages sent while resuming
	Dialer                 *net.Dialer
	CustomOpenConnectionFn mqtt.OpenConnectionFunc
	AutoAckDisabled        bool
}

type PeerMQTT

type PeerMQTT struct {
	*Client
}

func (*PeerMQTT) Init

func (p *PeerMQTT) Init(config json.RawMessage, args ...any) error

func (*PeerMQTT) Publish

func (p *PeerMQTT) Publish(event logrepl.PostgresCDC) error