mqtt

Documentation

Index

• Variables
• type Admin
• type Config
• type GMqttServer
• type Message
• type MessageHandler
• type Mqtt
  • func (m *Mqtt) Admin() Admin
  • func (m *Mqtt) Authenticator() mqtt_authenticator.MqttAuthenticator
  • func (m *Mqtt) NewClient(name string) (client MqttCli)
  • func (m *Mqtt) Publish(topic string, payload []byte, qos uint8, retain bool) (err error)
  • func (m *Mqtt) RemoveClient(name string)
  • func (m *Mqtt) Shutdown() (err error)
  • func (m *Mqtt) Start()
• type MqttBind
  • func NewMqttBind(mqtt MqttServ) *MqttBind
  • func (m MqttBind) Publish(topic string, payload []byte, qos uint8, retain bool)
• type MqttCli
  • func NewClient(mqtt *Mqtt, name string) MqttCli
• type MqttServ
  • func NewMqtt(lc fx.Lifecycle, cfg *Config, ...) (mqtt MqttServ)

Variables

var (
	// ErrInvalidTopicFilter ...
	ErrInvalidTopicFilter = errors.New("invalid topic filter")
	// ErrInvalidQos ...
	ErrInvalidQos = errors.New("invalid Qos")
	// ErrInvalidUtf8String ...
	ErrInvalidUtf8String = errors.New("invalid utf-8 string")
)

Types

type Admin

type Admin interface {
	GetClients(limit, offset uint) (list []*admin.ClientInfo, total uint32, err error)
	GetClient(clientId string) (client *admin.ClientInfo, err error)
	GetSessions(limit, offset uint) (list []*admin.SessionInfo, total int, err error)
	GetSession(clientId string) (session *admin.SessionInfo, err error)
	GetSubscriptions(clientId string, limit, offset uint) (list []*admin.SubscriptionInfo, total int, err error)
	Subscribe(clientId, topic string, qos int) (err error)
	Unsubscribe(clientId, topic string) (err error)
	Publish(topic string, qos int, payload []byte, retain bool) (err error)
	CloseClient(clientId string) (err error)
	SearchTopic(query string) (result []*admin.SubscriptionInfo, err error)
}

Admin ...

type Config

type Config struct {
	Port                       int
	RetryInterval              time.Duration
	RetryCheckInterval         time.Duration
	SessionExpiryInterval      time.Duration
	SessionExpireCheckInterval time.Duration
	QueueQos0Messages          bool
	MaxInflight                int
	MaxAwaitRel                int
	MaxMsgQueue                int
	Logging                    bool
	DebugMode                  common.RunMode
}

Config ...

type GMqttServer

type GMqttServer interface {
	// ...
	Run() error
	// ...
	Stop(ctx context.Context) error
	// ...
	Init(opts ...server.Options) error
	// SubscriptionStore returns the subscription.Store.
	SubscriptionService() server.SubscriptionService
	// RetainedStore returns the retained.Store.
	RetainedStore() retained.Store
	// Publisher returns the Publisher
	Publisher() server.Publisher
	// client return the ClientService
	ClientService() server.ClientService
	// GetConfig returns the config of the server
	GetConfig() config.Config
	// StatsManager returns StatsReader
	StatsManager() server.StatsReader
}

GMqttServer ...

type Message

type Message struct {
	Dup      bool
	Qos      uint8
	Retained bool
	Topic    string
	PacketID uint16
	Payload  []byte
}

Message ...

type MessageHandler

type MessageHandler func(MqttCli, Message)

MessageHandler ...

type Mqtt

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

Mqtt ...

func (*Mqtt) Admin

func (m *Mqtt) Admin() Admin

Admin ...

func (*Mqtt) Authenticator

func (m *Mqtt) Authenticator() mqtt_authenticator.MqttAuthenticator

Authenticator ...

func (*Mqtt) NewClient

func (m *Mqtt) NewClient(name string) (client MqttCli)

NewClient ...

func (*Mqtt) Publish

func (m *Mqtt) Publish(topic string, payload []byte, qos uint8, retain bool) (err error)

Publish ...

func (*Mqtt) RemoveClient

func (m *Mqtt) RemoveClient(name string)

RemoveClient ...

func (*Mqtt) Shutdown

func (m *Mqtt) Shutdown() (err error)

Shutdown ...

func (*Mqtt) Start

func (m *Mqtt) Start()

Start ...

type MqttBind

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

MqttBind ...

func NewMqttBind

func NewMqttBind(mqtt MqttServ) *MqttBind

NewMqttBind ...

func (MqttBind) Publish

func (m MqttBind) Publish(topic string, payload []byte, qos uint8, retain bool)

Publish ...

type MqttCli

type MqttCli interface {
	Publish(topic string, payload []byte) error
	Subscribe(topic string, handler MessageHandler) error
	Unsubscribe(topic string)
	UnsubscribeAll()
	OnMsgArrived(ctx context.Context, client server.Client, req *server.MsgArrivedRequest)
}

MqttCli ...

func NewClient

func NewClient(mqtt *Mqtt, name string) MqttCli

NewClient ...

type MqttServ

type MqttServ interface {
	Shutdown() error
	Start()
	Publish(topic string, payload []byte, qos uint8, retain bool) error
	NewClient(name string) MqttCli
	RemoveClient(name string)
	Admin() Admin
	Authenticator() mqtt_authenticator.MqttAuthenticator
}

MqttServ ...

func NewMqtt

func NewMqtt(lc fx.Lifecycle,
	cfg *Config,
	authenticator mqtt_authenticator.MqttAuthenticator,
	scriptService scripts.ScriptService,
) (mqtt MqttServ)

NewMqtt ...