mqtt

Documentation

Index

• Variables
• func IsTwoWayTLS(conn transport.Conn) bool
• type Client
  • func NewClient(cc ClientInfo, handler Handler, log logger.Logger) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Dying() <-chan struct{}
  • func (c *Client) Send(p packet.Generic) (err error)
• type ClientInfo
• type Dialer
  • func NewDialer(c utils.Certificate) (*Dialer, error)
  • func (d *Dialer) Dial(urlString string) (transport.Conn, error)
• type Dispatcher
  • func NewDispatcher(cc ClientInfo, log logger.Logger) *Dispatcher
  • func (d *Dispatcher) Close() error
  • func (d *Dispatcher) Publish(pid uint16, qos uint32, topic string, payload []byte, retain bool, ...) error
  • func (d *Dispatcher) Send(pkt packet.Generic) error
  • func (d *Dispatcher) Start(h Handler) error
• type Future
  • func NewFuture() *Future
  • func (f *Future) Cancel(err error)
  • func (f *Future) Complete()
  • func (f *Future) Wait(timeout time.Duration) (err error)
• type Handler
• type HandlerWrapper
  • func NewHandlerWrapper(onPublish ProcessPublish, onPuback ProcessPuback, onError ProcessError) *HandlerWrapper
  • func (h *HandlerWrapper) ProcessError(err error)
  • func (h *HandlerWrapper) ProcessPuback(pkt *packet.Puback) error
  • func (h *HandlerWrapper) ProcessPublish(pkt *packet.Publish) error
• type Launcher
  • func NewLauncher(c utils.Certificate) (*Launcher, error)
  • func (l *Launcher) Launch(urlString string) (transport.Server, error)
• type ProcessError
• type ProcessPuback
• type ProcessPublish
• type TopicInfo

Variables

var ErrDispatcherClosed = fmt.Errorf("dispatcher already closed")

ErrDispatcherClosed is returned if the dispatcher is closed

Functions

func IsTwoWayTLS

func IsTwoWayTLS(conn transport.Conn) bool

IsTwoWayTLS check two-way tls connection

Types

type Client

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

A Client connects to a broker and handles the transmission of packets

func NewClient

func NewClient(cc ClientInfo, handler Handler, log logger.Logger) (*Client, error)

NewClient returns a new client

func (*Client) Close

func (c *Client) Close() error

Close closes the client after sending a disconnect packet

func (*Client) Dying

func (c *Client) Dying() <-chan struct{}

Dying returns the channel that can be used to wait until client closed

func (*Client) Send

func (c *Client) Send(p packet.Generic) (err error)

Send sends a generic packet

type ClientInfo

type ClientInfo struct {
	Address           string `yaml:"address" json:"address"`
	Username          string `yaml:"username" json:"username"`
	Password          string `yaml:"password" json:"password"`
	utils.Certificate `yaml:",inline" json:",inline"`
	ClientID          string        `yaml:"clientid" json:"clientid"`
	CleanSession      bool          `yaml:"cleansession" json:"cleansession"`
	Timeout           time.Duration `yaml:"timeout" json:"timeout" default:"30s"`
	Interval          time.Duration `yaml:"interval" json:"interval" default:"1m"`
	KeepAlive         time.Duration `yaml:"keepalive" json:"keepalive" default:"1m"`
	BufferSize        int           `yaml:"buffersize" json:"buffersize" default:"10"`
	ValidateSubs      bool          `yaml:"validatesubs" json:"validatesubs"`
	Subscriptions     []TopicInfo   `yaml:"subscriptions" json:"subscriptions" default:"[]"`
}

ClientInfo mqtt client config

type Dialer

type Dialer struct {
	*transport.Dialer
}

The Dialer handles connecting to a server and creating a connection.

func NewDialer

func NewDialer(c utils.Certificate) (*Dialer, error)

NewDialer returns a new Dialer.

func (*Dialer) Dial

func (d *Dialer) Dial(urlString string) (transport.Conn, error)

Dial initiates a connection based in information extracted from an URL.

type Dispatcher

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

Dispatcher dispatcher of mqtt client

func NewDispatcher

func NewDispatcher(cc ClientInfo, log logger.Logger) *Dispatcher

NewDispatcher creates a new dispatcher

func (*Dispatcher) Close

func (d *Dispatcher) Close() error

Close closes dispatcher

func (*Dispatcher) Publish

func (d *Dispatcher) Publish(pid uint16, qos uint32, topic string, payload []byte, retain bool, duplicate bool) error

Publish sends a publish packet

func (*Dispatcher) Send

func (d *Dispatcher) Send(pkt packet.Generic) error

Send sends a generic packet

func (*Dispatcher) Start

func (d *Dispatcher) Start(h Handler) error

Start starts dispatcher

type Future

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

A Future is a low-level future type that can be extended to transport custom information.

func NewFuture

func NewFuture() *Future

NewFuture will return a new Future.

func (*Future) Cancel

func (f *Future) Cancel(err error)

Cancel will cancel the future with an error.

func (*Future) Complete

func (f *Future) Complete()

Complete will complete the future.

func (*Future) Wait

func (f *Future) Wait(timeout time.Duration) (err error)

Wait will wait the given amount of time and return whether the future has been completed, canceled or the request timed out.

type Handler

type Handler interface {
	ProcessPublish(*packet.Publish) error
	ProcessPuback(*packet.Puback) error
	ProcessError(error)
}

Handler MQTT message handler interface

type HandlerWrapper

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

HandlerWrapper MQTT message handler wrapper

func NewHandlerWrapper

func NewHandlerWrapper(onPublish ProcessPublish, onPuback ProcessPuback, onError ProcessError) *HandlerWrapper

NewHandlerWrapper creates a new handler wrapper

func (*HandlerWrapper) ProcessError

func (h *HandlerWrapper) ProcessError(err error)

ProcessError handles error

func (*HandlerWrapper) ProcessPuback

func (h *HandlerWrapper) ProcessPuback(pkt *packet.Puback) error

ProcessPuback handles puback packet

func (*HandlerWrapper) ProcessPublish

func (h *HandlerWrapper) ProcessPublish(pkt *packet.Publish) error

ProcessPublish handles publish packet

type Launcher

type Launcher struct {
	transport.Launcher
}

The Launcher helps with launching a server and accepting connections.

func NewLauncher

func NewLauncher(c utils.Certificate) (*Launcher, error)

NewLauncher returns a new Launcher.

func (*Launcher) Launch

func (l *Launcher) Launch(urlString string) (transport.Server, error)

Launch will launch a server based on information extracted from an URL.

type ProcessError

type ProcessError func(error)

ProcessError handles error

type ProcessPuback

type ProcessPuback func(*packet.Puback) error

ProcessPuback handles puback packet

type ProcessPublish

type ProcessPublish func(*packet.Publish) error

ProcessPublish handles publish packet

type TopicInfo

type TopicInfo struct {
	QOS   uint32 `yaml:"qos" json:"qos" validate:"min=0, max=1"`
	Topic string `yaml:"topic" json:"topic" validate:"nonzero"`
}

TopicInfo with topic and qos