gotag

README

Gotag

GoTag is a Go pakcage for ThingsPro. It integrates the mqtt client and the protobuffer which makes data exchanging become easily and narrow down the transmission bandwidth.

Installation

Once you have installed Go, run these commands to install the gotag:

    go get github.com/CPtung/gotag

Build a gotag client

import (
    gotag github.com/CPtung/gotag
)

func main() {   
    client, err := gotag.NewClient()
    if err != nil {
        log.Fatal(err)
    }
    defer client.Delete()

    //...
}

Run data pub/sub

As mentioned, gotag use a mqtt client to do the transmission which means it has to pub/sub topics to send/receive data.

Publish data

func PUB(client *gotag.Tagf) {
    value := gotag.NewValue(1.414)
    client.Publish(
        "gotag",
        "test",
        value,
        t.TAG_VALUE_TYPE_DOUBLE,
        1546920188000,
        "unit")
}

Subscribe data

func SUB(client *gotag.Tagf) {
    client.SubscribeCallback(Handler)
    client.Subscribe("gotag", "test")
}

SubscribeCallback

Gotag needs to register a callback function for subscribed topics.

func Handler(source string, tag string, val *t.Value, valtype int32, ts uint64, unit string) {
    fmt.Printf("Source: %v,", source)
    fmt.Printf("Tag: %v,", tag)
    fmt.Printf("Value: %v,", val.GetDouble())
    fmt.Printf("ValueType: %v,", valtype)
    fmt.Printf("At: %v,", ts)
    fmt.Printf("Unit: %v\n", unit)
}

Documentation

Index

• Constants
• func DecodePayload(payload []byte, tag *Tag) error
• func DecodeTopic(topic string) (string, string, error)
• func EncodePayload(source string, tag string, value *Value, valtype int32, at uint64, unit string) []byte
• func EncodeTopic(source, tag string) string
• type MQConfig
• type MsgQueueBase
• type OnTagCallback
• type Tag
• type Tagf
  • func NewClient(host ...string) (*Tagf, error)
  • func (self *Tagf) Delete() error
  • func (self *Tagf) Publish(sourceName string, tagName string, val *Value, valType int32, timestamp uint64, ...) error
  • func (self *Tagf) Subscribe(sourceName, tagName string) error
  • func (self *Tagf) SubscribeCallback(ontag OnTagCallback) error
  • func (self *Tagf) UnSubscribe(sourceName, tagName string) error
• type TpMqtt
  • func NewMqtt(cfg *MQConfig) (*TpMqtt, error)
  • func (self *TpMqtt) Close() error
  • func (self *TpMqtt) OnConnectHandler(client mqtt.Client)
  • func (self *TpMqtt) OnDisconnectHandler(client mqtt.Client, err error)
  • func (self *TpMqtt) OnPublishHandler(client mqtt.Client, message mqtt.Message)
  • func (self *TpMqtt) Publish(topic string, payload []byte) error
  • func (self *TpMqtt) SetLogLevel(level string) error
  • func (self *TpMqtt) Subscribe(topic string) error
  • func (self *TpMqtt) SubscribeCallback(hnd OnTagCallback) error
  • func (self *TpMqtt) UnSubscribe(topic string) error
• type Value
  • func NewValue(value interface{}) *Value
  • func (m *Value) GetBytes() []byte
  • func (m *Value) GetDouble() float64
  • func (m *Value) GetFloat() float32
  • func (m *Value) GetInt() int64
  • func (m *Value) GetStr() string
  • func (m *Value) GetUint() uint64
• type WaitLock
  • func WaitSync(wg *sync.WaitGroup, timeout time.Duration) *WaitLock
  • func (w *WaitLock) PostDelay() bool

Constants

const (
	ERR_SUCESS        = 0
	ERR_FAILED        = 1
	ERR_INVALID_INPUT = 2
	ERR_NULL_ACCESS   = 3
)

const (
	TAG_VALUE_TYPE_BOOLEAN   = 0
	TAG_VALUE_TYPE_INT8      = 1
	TAG_VALUE_TYPE_INT16     = 2
	TAG_VALUE_TYPE_INT32     = 3
	TAG_VALUE_TYPE_INT64     = 4
	TAG_VALUE_TYPE_INT       = 5
	TAG_VALUE_TYPE_UINT8     = 6
	TAG_VALUE_TYPE_UINT16    = 7
	TAG_VALUE_TYPE_UINT32    = 8
	TAG_VALUE_TYPE_UINT64    = 9
	TAG_VALUE_TYPE_UINT      = 10
	TAG_VALUE_TYPE_FLOAT     = 11
	TAG_VALUE_TYPE_DOUBLE    = 12
	TAG_VALUE_TYPE_STRING    = 13
	TAG_VALUE_TYPE_BYTEARRAY = 14
)

Functions

func DecodePayload

func DecodePayload(payload []byte, tag *Tag) error

func DecodeTopic

func DecodeTopic(topic string) (string, string, error)

func EncodePayload

func EncodePayload(source string, tag string, value *Value, valtype int32, at uint64, unit string) []byte

func EncodeTopic

func EncodeTopic(source, tag string) string

Types

type MQConfig

type MQConfig struct {
	Id       string
	Host     string
	Port     string
	Debug    string
	Qos      byte
	Retained bool
}

type MsgQueueBase

type MsgQueueBase interface {
	Publish(topic string, payload []byte) error
	Subscribe(topic string) error
	UnSubscribe(topic string) error
	SubscribeCallback(ontag OnTagCallback) error
	Close() error
}

type OnTagCallback

type OnTagCallback func(
	sourceName string,
	tagName string,
	value *Value,
	valueType int32,
	timestamp uint64,
	unit string)

type Tag

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

type Tagf

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

func NewClient

func NewClient(host ...string) (*Tagf, error)

func (*Tagf) Delete

func (self *Tagf) Delete() error

func (*Tagf) Publish

func (self *Tagf) Publish(sourceName string, tagName string, val *Value, valType int32, timestamp uint64, unit string) error

func (*Tagf) Subscribe

func (self *Tagf) Subscribe(sourceName, tagName string) error

func (*Tagf) SubscribeCallback

func (self *Tagf) SubscribeCallback(ontag OnTagCallback) error

func (*Tagf) UnSubscribe

func (self *Tagf) UnSubscribe(sourceName, tagName string) error

type TpMqtt

type TpMqtt struct {
	MsgQueueBase
	// contains filtered or unexported fields
}

func NewMqtt

func NewMqtt(cfg *MQConfig) (*TpMqtt, error)

func (*TpMqtt) Close

func (self *TpMqtt) Close() error

func (*TpMqtt) OnConnectHandler

func (self *TpMqtt) OnConnectHandler(client mqtt.Client)

func (*TpMqtt) OnDisconnectHandler

func (self *TpMqtt) OnDisconnectHandler(client mqtt.Client, err error)

func (*TpMqtt) OnPublishHandler

func (self *TpMqtt) OnPublishHandler(client mqtt.Client, message mqtt.Message)

func (*TpMqtt) Publish

func (self *TpMqtt) Publish(topic string, payload []byte) error

func (*TpMqtt) SetLogLevel

func (self *TpMqtt) SetLogLevel(level string) error

func (*TpMqtt) Subscribe

func (self *TpMqtt) Subscribe(topic string) error

func (*TpMqtt) SubscribeCallback

func (self *TpMqtt) SubscribeCallback(hnd OnTagCallback) error

func (*TpMqtt) UnSubscribe

func (self *TpMqtt) UnSubscribe(topic string) error

type Value

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

func NewValue

func NewValue(value interface{}) *Value

func (*Value) GetBytes

func (m *Value) GetBytes() []byte

func (*Value) GetDouble

func (m *Value) GetDouble() float64

func (*Value) GetFloat

func (m *Value) GetFloat() float32

func (*Value) GetInt

func (m *Value) GetInt() int64

func (*Value) GetStr

func (m *Value) GetStr() string

func (*Value) GetUint

func (m *Value) GetUint() uint64

type WaitLock

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

func WaitSync

func WaitSync(wg *sync.WaitGroup, timeout time.Duration) *WaitLock

func (*WaitLock) PostDelay

func (w *WaitLock) PostDelay() bool