mqtt

Nov 6, 2025

Changes in this version

+ var ErrUnsupported = NewRpcError("unsupported operation")

+ func NewRpcError(msg string) error

+ func WithPublishFunc(ctx context.Context, fn PublishFunc) context.Context

+ type BrokerConfig struct

+ ClientID string

+ Password string

+ URL string

+ Username string

+ type Call struct

+ MaxWait time.Duration

+ QoS int

+ Request any

+ Response any

+ Retain bool

+ Topic string

+ Unmarshal func(payload []byte, resp any) error

+ func (c *Call) Payload() ([]byte, error)

+ type Config struct

+ AutoSubscribe bool

+ Broker BrokerConfig

+ HandlerTimeout time.Duration

+ Logger *zerolog.Logger

+ MessageHandlers map[string]MessageHandler

+ RequestHandlers map[string]RequestHandler

+ Subscriptions []string

+ type Endpoint interface

+ Call func(ctx context.Context, c *Call) error

+ Connect func(ctx context.Context) error

+ Disconnect func(ctx context.Context) error

+ Publish func(ctx context.Context, msg *Message) error

+ RegisterMessageHandler func(topic string, h MessageHandler)

+ RegisterRequestHandler func(topic string, h RequestHandler)

+ Subscribe func(ctx context.Context, topic string) error

+ UnregisterHandler func(topic string)

+ Unsubscribe func(ctx context.Context, topic string) error

+ type HTTPDoer struct

+ func NewHTTPDoer(config HTTPDoerConfig) *HTTPDoer

+ func (d *HTTPDoer) Do(req *http.Request) (*http.Response, error)

+ type HTTPDoerConfig struct

+ Endpoint Endpoint

+ MaxWait time.Duration

+ QoS int

+ Retain bool

+ Topic string

+ type HTTPServer struct

+ func NewHTTPServer(config HTTPServerConfig) *HTTPServer

+ func (s *HTTPServer) Start(ctx context.Context) error

+ func (s *HTTPServer) Stop(ctx context.Context) error

+ type HTTPServerConfig struct

+ Broker BrokerConfig

+ Handler http.Handler

+ HandlerTimeout time.Duration

+ Logger *zerolog.Logger

+ Topic string

+ type MQTTEndpoint struct

+ func New(config Config) *MQTTEndpoint

+ func (e *MQTTEndpoint) Call(ctx context.Context, c *Call) error

+ func (e *MQTTEndpoint) Connect(ctx context.Context) error

+ func (e *MQTTEndpoint) Disconnect(ctx context.Context) error

+ func (e *MQTTEndpoint) Publish(ctx context.Context, msg *Message) error

+ func (e *MQTTEndpoint) RegisterMessageHandler(topic string, h MessageHandler)

+ func (e *MQTTEndpoint) RegisterRequestHandler(topic string, h RequestHandler)

+ func (e *MQTTEndpoint) Subscribe(ctx context.Context, topic string) error

+ func (e *MQTTEndpoint) UnregisterHandler(topic string)

+ func (e *MQTTEndpoint) Unsubscribe(ctx context.Context, topic string) error

+ type Message struct

+ Expiry time.Duration

+ Payload []byte

+ QoS int

+ Retain bool

+ Topic string

+ type MessageHandler func(ctx context.Context, msg *Message)

+ type PublishFunc func(ctx context.Context, msg *Message) error

+ func GetPublishFunc(ctx context.Context) (PublishFunc, bool)

+ type RequestHandler func(ctx context.Context, msg *Message) (any, error)