mqtt

Documentation

Index

• Constants
• Variables
• type Client
  • func NewClient(broker string, opts ...ClientOption) (Client, error)
• type ClientOption
  • func WithCleanSession(clean bool) ClientOption
  • func WithClientId(clientId string) ClientOption
  • func WithConnectBackoffMaxInterval(maxInterval time.Duration) ClientOption
  • func WithConnectTimeout(timeout time.Duration) ClientOption
  • func WithConnectionFactory(factory Factory) ClientOption
  • func WithDisconnectTimeout(timeout time.Duration) ClientOption
  • func WithKeepAliveInterval(interval time.Duration) ClientOption
  • func WithKeepAliveTimeout(timeout time.Duration) ClientOption
  • func WithLogger(logger *slog.Logger) ClientOption
  • func WithOnConnectionStatusHandler(handler OnConnectionStatusHandler) ClientOption
  • func WithPassword(password string) ClientOption
  • func WithUsername(username string) ClientOption
• type ConnectionStatus
• type ConnectionStatusCode
• type Factory
  • func NewFactory(dialer *net.Dialer, tlsConfig *tls.Config) Factory
• type Future
  • func NewFuture[T any]() Future[T]
• type OnConnectionStatusHandler
• type OnReceiveHandler
• type QosLevel
• type Result
  • func NewResult[T any](value T, err error) Result[T]

Constants

const (
	SubackSuccessQoS0 = 0
	SubackSuccessQoS1 = 1
	SubackSuccessQoS2 = 2
	SubackFailure     = 128
)

Variables

var ErrAlreadyRunning = errors.New("client is already running")

var ErrConnectionClosedForReconnect = errors.New("connection closed for reconnect")

var ErrConnectionStopped = errors.New("connection stopped")

var ErrEmptyBroker = errors.New("broker must be provided")

var ErrEmptyHost = errors.New("empty host")

var ErrFlushWithNilError = errors.New("signaler must be closed with a non-nil error")

var ErrNoRequestForResponse = errors.New("no request for response")

var ErrNoSubscription = errors.New("no subscription matches packet topic name")

var ErrNotConnack = errors.New("not connack packet")

var ErrRequestReplaced = errors.New("replaced by new request")

var ErrSubackReturnedFailureCode = errors.New("suback returned failure")

var ErrSubscriptionAlreadyRegistered = errors.New("subscription already registered")

var ErrUnsupportedPacketType = errors.New("unsupported packet type")

var ErrUnsupportedScheme = errors.New("unsupported broker scheme")

Types

type Client

type Client interface {
	Run(ctx context.Context) error
	Publish(ctx context.Context, topic string, payload []byte, qos QosLevel) Future[packets.ControlPacket]
	Subscribe(ctx context.Context, topic string, qos QosLevel, handler OnReceiveHandler) Future[packets.SubackPacket]
	Unsubscribe(ctx context.Context, topic string) Future[packets.UnsubackPacket]
	ReconnectWithCredentials(username string, password string)
	Reconnect()
	Status() ConnectionStatus
}

func NewClient

func NewClient(broker string, opts ...ClientOption) (Client, error)

type ClientOption

type ClientOption func(*client)

func WithCleanSession

func WithCleanSession(clean bool) ClientOption

func WithClientId

func WithClientId(clientId string) ClientOption

func WithConnectBackoffMaxInterval

func WithConnectBackoffMaxInterval(maxInterval time.Duration) ClientOption

func WithConnectTimeout

func WithConnectTimeout(timeout time.Duration) ClientOption

func WithConnectionFactory

func WithConnectionFactory(factory Factory) ClientOption

func WithDisconnectTimeout

func WithDisconnectTimeout(timeout time.Duration) ClientOption

func WithKeepAliveInterval

func WithKeepAliveInterval(interval time.Duration) ClientOption

func WithKeepAliveTimeout

func WithKeepAliveTimeout(timeout time.Duration) ClientOption

func WithLogger

func WithLogger(logger *slog.Logger) ClientOption

func WithOnConnectionStatusHandler

func WithOnConnectionStatusHandler(handler OnConnectionStatusHandler) ClientOption

func WithPassword

func WithPassword(password string) ClientOption

func WithUsername

func WithUsername(username string) ClientOption

type ConnectionStatus

type ConnectionStatus struct {
	Code ConnectionStatusCode
	Err  error
}

type ConnectionStatusCode

type ConnectionStatusCode = int

const (
	Stopped ConnectionStatusCode = iota
	Connecting
	Connected
	Disconnected
)

type Factory

type Factory interface {
	Create(broker string) (net.Conn, error)
}

func NewFactory

func NewFactory(dialer *net.Dialer, tlsConfig *tls.Config) Factory

type Future

type Future[T any] interface {
	Value() T
	Error() error
	Done() <-chan any
	Complete(value T, err error) Future[T]
}

func NewFuture

func NewFuture[T any]() Future[T]

type OnConnectionStatusHandler

type OnConnectionStatusHandler func(client Client, status ConnectionStatus)

type OnReceiveHandler

type OnReceiveHandler func(client Client, packet packets.PublishPacket) error

type QosLevel

type QosLevel = byte

const (
	QosLevel0 QosLevel = 0
	QosLevel1 QosLevel = 1
)

type Result

type Result[T any] interface {
	Value() T
	Error() error
}

func NewResult

func NewResult[T any](value T, err error) Result[T]