mqtt

Documentation

Index

• Constants
• Variables
• func TopicBuilder(opt BuilderOption, format string, key ...any) (t string)
• func TopicResolver(tpl string, topic string) (res map[string]string, err error)
• type BuilderOption
• type ClientOption
• type Session
  • func New(ctx context.Context, options ...ClientOption) Session

Constants

const (
	TopicGroup = "default"
)

Variables

var (
	ErrPublishTimeout = errors.New("mqtt publish wait timeout")
)

var WithAuth = func(username, password string) ClientOption {
	return func(options *proxy.ClientOptions) *proxy.ClientOptions {
		options = options.SetUsername(username)
		if len(password) > 0 {
			options = options.SetPassword(password)
		}
		return options
	}
}

var WithBroker = func(broker string, others ...string) ClientOption {
	return func(options *proxy.ClientOptions) *proxy.ClientOptions {
		options = options.AddBroker(broker)
		for _, b := range others {
			options = options.AddBroker(b)
		}
		return options
	}
}

var WithClientId = func(clientId string) ClientOption {
	return func(options *proxy.ClientOptions) *proxy.ClientOptions {
		options = options.SetClientID(clientId)
		return options
	}
}

var WithOnConnectHandler = func(h proxy.OnConnectHandler) ClientOption {
	return func(options *proxy.ClientOptions) *proxy.ClientOptions {
		options = options.SetOnConnectHandler(h)
		return options
	}
}

var WithOnLostHandler = func(h proxy.ConnectionLostHandler) ClientOption {
	return func(options *proxy.ClientOptions) *proxy.ClientOptions {
		options = options.SetConnectionLostHandler(h)
		return options
	}
}

var WithRandomClientId = func() ClientOption {
	return func(options *proxy.ClientOptions) *proxy.ClientOptions {
		options = options.SetClientID(util.UUIDString())
		log.Debug("ClientId:%s", options.ClientID)
		return options
	}
}

var WithSessionControl = func(keepAlive, pingTimeout, writeTimeout time.Duration) ClientOption {
	return func(options *proxy.ClientOptions) *proxy.ClientOptions {
		options = options.
			SetKeepAlive(keepAlive).
			SetPingTimeout(pingTimeout).
			SetWriteTimeout(writeTimeout)
		return options
	}
}

Functions

func TopicBuilder

func TopicBuilder(opt BuilderOption, format string, key ...any) (t string)

func TopicResolver

func TopicResolver(tpl string, topic string) (res map[string]string, err error)

Types

type BuilderOption

type BuilderOption struct {
	Share bool   // 是否共享topic
	Tpl   bool   // true: /a/b/{c}/d  false:/a/b/c/d
	Group string // group名称, 默认default
}

type ClientOption

type ClientOption func(*proxy.ClientOptions) *proxy.ClientOptions

type Session

type Session interface {
	Connect() error
	Publish(topic string, qos byte, retained bool, data []byte) error
	Subscribe(handler func(topic string, payload []byte), qos byte, topic string) error
	Unsubscribe(topics ...string)
}

func New

func New(ctx context.Context, options ...ClientOption) Session