mqtt

Documentation

Index

• Constants
• Variables
• type MessageDirection
• type MessageState
• type MqttFactory
  • func (m *MqttFactory) New(protocol string, c core.Config, ch chan base.ServiceCommand) (base.Service, error)
• type SessionTopic
  • func (p *SessionTopic) Encode() ([]byte, error)
  • func (p *SessionTopic) Length() int
• type Storage
  • func NewStorage(name string, c core.Config) (Storage, error)
• type StorageDevice
• type StorageMessage
• type StorageSession
• type StorageTopic
• type TenantTopic

Constants

const (
	// Protocol version
	PROTOCOL_NAME_V31    = "MQIsdp"
	PROTOCOL_VERSION_V31 = 3

	PROTOCOL_NAME_V311    = "MQTT"
	PROTOCOL_VERSION_V311 = 4

	// Message types
	INVALID     = 0x00
	CONNECT     = 0x10
	CONNACK     = 0x20
	PUBLISH     = 0x30
	PUBACK      = 0x40
	PUBREC      = 0x50
	PUBREL      = 0x60
	PUBCOMP     = 0x70
	SUBSCRIBE   = 0x80
	SUBACK      = 0x90
	UNSUBSCRIBE = 0xA0
	UNSUBACK    = 0xB0
	PINGREQ     = 0xC0
	PINGRESP    = 0xD0
	DISCONNECT  = 0xE0

	// CONNACK result
	CONNACK_ACCEPTED                      = 0
	CONNACK_REFUSED_PROTOCOL_VERSION      = 1
	CONNACK_REFUSED_IDENTIFIER_REJECTED   = 2
	CONNACK_REFUSED_SERVER_UNAVAILABLE    = 3
	CONNACK_REFUSED_BAD_USERNAME_PASSWORD = 4
	CONNACK_REFUSED_NOT_AUTHORIZED        = 5

	MQTT_MAX_PAYLOAD = 268435455
)

const (
	ObjectActionRegister   = "register"
	ObjectActionUnregister = "unregister"
	ObjectActionRetrieve   = "retrieve"
	ObjectActionDelete     = "delete"
	ObjectActionUpdate     = "update"
)

const (
	TopicNameSession = "mqtt-session"
)

Variables

var Configs = map[string]string{
	"bind_address":       "localhost:1883",
	"max_connections":    "1000",
	"loglevel":           "debug",
	"message_size_limit": "500",
	"allow_anonymous":    "true",
	"session_queue_size": "20",
}

Types

type MessageDirection

type MessageDirection int

const (
	MessageDirectionIn  MessageDirection = 0
	MessageDirectionOut MessageDirection = 1
)

type MessageState

type MessageState int

MessageState ...

type MqttFactory

type MqttFactory struct{}

MqttFactory

func (*MqttFactory) New

func (m *MqttFactory) New(protocol string, c core.Config, ch chan base.ServiceCommand) (base.Service, error)

New create mqtt service factory

type SessionTopic

type SessionTopic struct {
	Launcher  string `json:"launcher"`
	SessionId string `json:"sessionId"`
	Action    string `json:"action"`
	State     uint8  `json:"oldState"`
	// contains filtered or unexported fields
}

func (*SessionTopic) Encode

func (p *SessionTopic) Encode() ([]byte, error)

func (*SessionTopic) Length

func (p *SessionTopic) Length() int

type Storage

type Storage interface {
	Open() error
	Close()
	Backup(shutdown bool) error
	Restore() error

	// Session
	FindSession(id string) (*mqttSession, error)
	DeleteSession(id string) error
	UpdateSession(s *mqttSession) error
	RegisterSession(s *mqttSession) error

	// Subscription
	AddSubscription(sessionid string, topic string, qos uint8) error
	RetainSubscription(sessionid string, topic string, qos uint8) error
	RemoveSubscription(sessionid string, topic string) error

	// Message Management
	FindMessage(clientid string, mid uint16) (bool, error)
	StoreMessage(clientid string, msg StorageMessage) error
	DeleteMessageWithValidator(clientid string, validator func(StorageMessage) bool)
	DeleteMessage(clientid string, mid uint16, direction MessageDirection) error

	QueueMessage(clientid string, msg StorageMessage) error
	GetMessageTotalCount(clientid string) int
	InsertMessage(clientid string, mid uint16, direction MessageDirection, msg StorageMessage) error
	ReleaseMessage(clientid string, mid uint16, direction MessageDirection) error
	UpdateMessage(clientid string, mid uint16, direction MessageDirection, state MessageState)
}

Storage ...

func NewStorage

func NewStorage(name string, c core.Config) (Storage, error)

NewStorage lookup registered storage list, create a new storage instance

type StorageDevice

type StorageDevice struct{}

type StorageMessage

type StorageMessage struct {
	ID        uint
	SourceID  string
	Topic     string
	Direction MessageDirection
	State     MessageState
	Qos       uint8
	Retain    bool
	Payload   []uint8
}

StorageMessage ...

type StorageSession

type StorageSession struct {
	Id                 string
	Username           string
	Password           string
	Keepalive          uint16
	LastMid            uint16
	State              uint8
	LastMessageInTime  time.Time
	LastMessageOutTime time.Time
	Ping               time.Time
	CleanSession       uint8
	SubscribeCount     uint32
	Protocol           uint8
	RefCount           uint8
}

Session storage

type StorageTopic

type StorageTopic struct {
	Name string
}

type TenantTopic

type TenantTopic struct{}