mqtt

Documentation

Index

• Constants
• func ConnectorFactory(URL url.URL) (mqhub.Connector, error)
• func Encode(msg mqhub.Message) ([]byte, error)
• func EndpointTopic(topicBase, name string) string
• func ParseTopic(topic, prefix string) (compID, endpoint string)
• func ParseTopicRel(relativeTopic string) (compID, endpoint string)
• func SubCompTopic(topicBase string, id ...string) string
• func TokenizeTopic(topic string) []string
• type Connector
  • func NewConnector(options *Options) *Connector
  • func (c *Connector) Close() error
  • func (c *Connector) Connect() mqhub.Future
  • func (c *Connector) Describe(componentID string) mqhub.Descriptor
  • func (c *Connector) Publish(comp mqhub.Component) (mqhub.Publication, error)
  • func (c *Connector) Watch(sink mqhub.MessageSink) (mqhub.Watcher, error)
• type DataEmitter
  • func (e *DataEmitter) ConsumeMessage(msg mqhub.Message) mqhub.Future
• type DataSink
  • func (s *DataSink) ConsumeMessage(msg mqhub.Message) mqhub.Future
• type Descriptor
  • func (d *Descriptor) Endpoint(name string) mqhub.EndpointRef
  • func (d *Descriptor) ID() string
  • func (d *Descriptor) SubComponent(id ...string) mqhub.Descriptor
  • func (d *Descriptor) Watch(sink mqhub.MessageSink) (mqhub.Watcher, error)
• type EndpointRef
  • func (r *EndpointRef) ConsumeMessage(msg mqhub.Message) mqhub.Future
  • func (r *EndpointRef) Watch(sink mqhub.MessageSink) (mqhub.Watcher, error)
• type Future
  • func (f *Future) Wait() error
• type HandlerRef
  • func MakeHandlerRef(handler paho.MessageHandler) *HandlerRef
• type Message
  • func NewMessage(prefix string, msg paho.Message) *Message
  • func (m *Message) As(out interface{}) error
  • func (m *Message) Component() string
  • func (m *Message) Endpoint() string
  • func (m *Message) IsState() bool
  • func (m *Message) Payload() ([]byte, error)
  • func (m *Message) Value() (interface{}, bool)
• type Options
  • func NewOptions() *Options
  • func (o *Options) AddServer(server string) *Options
  • func (o *Options) Auth(username, password string) *Options
  • func (o *Options) SetClientID(id string) *Options
• type Publication
  • func (p *Publication) Close() error
  • func (p *Publication) Component() mqhub.Component
• type TopicFilter
  • func NewTopicFilter(filter string) *TopicFilter
  • func (f *TopicFilter) Matches(topic string) bool
  • func (f *TopicFilter) MatchesTokenized(tokens []string) bool
  • func (f *TopicFilter) String() string
• type TopicHandlerMap
  • func NewTopicHandlerMap() *TopicHandlerMap
  • func (m *TopicHandlerMap) Add(filters []string, handler *HandlerRef) (subs []string)
  • func (m *TopicHandlerMap) Del(filters []string, handler *HandlerRef) (unsubs []string)
  • func (m *TopicHandlerMap) HandleMessage(client paho.Client, msg paho.Message)

Constants

const (
	// Protocol is the name of protocol for connector
	Protocol = "mqtt"

	// OptClientID is the property name in URL query
	OptClientID = "client-id"
)

Functions

func ConnectorFactory

func ConnectorFactory(URL url.URL) (mqhub.Connector, error)

ConnectorFactory implements mqhub.ConnectorFactory

func Encode

func Encode(msg mqhub.Message) ([]byte, error)

Encode encodes original message into bytes

func EndpointTopic

func EndpointTopic(topicBase, name string) string

EndpointTopic creates a topic for endpoints

func ParseTopic

func ParseTopic(topic, prefix string) (compID, endpoint string)

ParseTopic parse topic including the prefix

func ParseTopicRel

func ParseTopicRel(relativeTopic string) (compID, endpoint string)

ParseTopicRel parse topic without prefix (start with component ID)

func SubCompTopic

func SubCompTopic(topicBase string, id ...string) string

SubCompTopic creates a topic for sub-components

func TokenizeTopic

func TokenizeTopic(topic string) []string

TokenizeTopic split topic into tokens

Types

type Connector

type Connector struct {
	Client paho.Client
	// contains filtered or unexported fields
}

Connector connects to MQTT

func NewConnector

func NewConnector(options *Options) *Connector

NewConnector creates a connector

func (*Connector) Close

func (c *Connector) Close() error

Close implements io.Closer

func (*Connector) Connect

func (c *Connector) Connect() mqhub.Future

Connect connects to server

func (*Connector) Describe

func (c *Connector) Describe(componentID string) mqhub.Descriptor

Describe creates a descriptor

func (*Connector) Publish

func (c *Connector) Publish(comp mqhub.Component) (mqhub.Publication, error)

Publish implements Publisher

func (*Connector) Watch

func (c *Connector) Watch(sink mqhub.MessageSink) (mqhub.Watcher, error)

Watch implements Watchable

type DataEmitter

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

DataEmitter is a consumer which publish the data to hub

func (*DataEmitter) ConsumeMessage

func (e *DataEmitter) ConsumeMessage(msg mqhub.Message) mqhub.Future

ConsumeMessage emits the message

type DataSink

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

DataSink is a consumer receives messages from hub

func (*DataSink) ConsumeMessage

func (s *DataSink) ConsumeMessage(msg mqhub.Message) mqhub.Future

ConsumeMessage implements MessageSink

type Descriptor

type Descriptor struct {
	ComponentID string `json:"id"`
	SubTopic    string `json:"topic"`
	// contains filtered or unexported fields
}

Descriptor represents advertisements and implements mqhub.Descriptor

func (*Descriptor) Endpoint

func (d *Descriptor) Endpoint(name string) mqhub.EndpointRef

Endpoint implements Descriptor

func (*Descriptor) ID

func (d *Descriptor) ID() string

ID implements Descriptor

func (*Descriptor) SubComponent

func (d *Descriptor) SubComponent(id ...string) mqhub.Descriptor

SubComponent implements Descriptor

func (*Descriptor) Watch

func (d *Descriptor) Watch(sink mqhub.MessageSink) (mqhub.Watcher, error)

Watch implements Descriptor

type EndpointRef

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

EndpointRef implements mqhub.EndpointRef

func (*EndpointRef) ConsumeMessage

func (r *EndpointRef) ConsumeMessage(msg mqhub.Message) mqhub.Future

ConsumeMessage implements MessageSink

func (*EndpointRef) Watch

func (r *EndpointRef) Watch(sink mqhub.MessageSink) (mqhub.Watcher, error)

Watch implements EndpointRef

type Future

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

Future implements mqhub.Future

func (*Future) Wait

func (f *Future) Wait() error

Wait implements Future

type HandlerRef

type HandlerRef struct {
	Handler paho.MessageHandler
}

HandlerRef wraps over a handler

func MakeHandlerRef

func MakeHandlerRef(handler paho.MessageHandler) *HandlerRef

MakeHandlerRef builds a HandlerRef

type Message

type Message struct {
	ComponentID  string
	EndpointName string
	Raw          paho.Message
}

Message implements Message

func NewMessage

func NewMessage(prefix string, msg paho.Message) *Message

NewMessage wraps mqtt message

func (*Message) As

func (m *Message) As(out interface{}) error

As implements Message

func (*Message) Component

func (m *Message) Component() string

Component implements Message

func (*Message) Endpoint

func (m *Message) Endpoint() string

Endpoint implements Message

func (*Message) IsState

func (m *Message) IsState() bool

IsState implements Message

func (*Message) Payload

func (m *Message) Payload() ([]byte, error)

Payload implements EncodedPayload

func (*Message) Value

func (m *Message) Value() (interface{}, bool)

Value implements Message

type Options

type Options struct {
	Servers   []*url.URL
	TLS       tls.Config
	Username  string
	Password  string
	ClientID  string
	Namespace string
}

Options defines configuration for connection

func NewOptions

func NewOptions() *Options

NewOptions creates options

func (*Options) AddServer

func (o *Options) AddServer(server string) *Options

AddServer adds a server

func (*Options) Auth

func (o *Options) Auth(username, password string) *Options

Auth sets username and password

func (*Options) SetClientID

func (o *Options) SetClientID(id string) *Options

SetClientID sets client ID

type Publication

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

Publication implements mqhub.Publication

func (*Publication) Close

func (p *Publication) Close() error

Close implements Publication

func (*Publication) Component

func (p *Publication) Component() mqhub.Component

Component implements Publication

type TopicFilter

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

TopicFilter defines a parsed topic filter

func NewTopicFilter

func NewTopicFilter(filter string) *TopicFilter

NewTopicFilter parses a topic filter in string

func (*TopicFilter) Matches

func (f *TopicFilter) Matches(topic string) bool

Matches indicates f matches the topic

func (*TopicFilter) MatchesTokenized

func (f *TopicFilter) MatchesTokenized(tokens []string) bool

MatchesTokenized indicates f matches the tokenized topic

func (*TopicFilter) String

func (f *TopicFilter) String() string

String returns the filter in string

type TopicHandlerMap

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

TopicHandlerMap maps topic filter to handlers

func NewTopicHandlerMap

func NewTopicHandlerMap() *TopicHandlerMap

NewTopicHandlerMap creates a new TopicHandlerMap

func (*TopicHandlerMap) Add

func (m *TopicHandlerMap) Add(filters []string, handler *HandlerRef) (subs []string)

Add inserts filters and corresponding handler the returns subs require a SUBSCRIBE

func (*TopicHandlerMap) Del

func (m *TopicHandlerMap) Del(filters []string, handler *HandlerRef) (unsubs []string)

Del removes filters/handler, the returned topics require UNSUBSCRIBE

func (*TopicHandlerMap) HandleMessage

func (m *TopicHandlerMap) HandleMessage(client paho.Client, msg paho.Message)

HandleMessage implements paho.MessageHandler