slink

README

slink

Base on mqtt protocal.

go get "github.com/USWS/slink"

Features

1. Service discovery
2. RPC
3. Subscribe & Publish

Connect

slk := slink.New("mqtt_broker_host:port").SetUserName("user").SetPassword("password")
if _, err := dis.Connect(); err != nil {
   panic(err)
}

Service discovery

// service provider side :
slk.ProvideService(&slink.ServiceInfo{
   Name: "service name",
   IP:   "ip address",
   Port: port number,
})

// client side or other service :
SvcGetter, err := slk.SetOnServiceUp("service_name", nil)
if err != nil {
   log.Println(err)
}

go func() {
   for {
      log.Println(SvcGetter())
      time.Sleep(1 * time.Second)
   }
}()

RPC

//RPC service provide side：
Register, err := slk.InitRpcServer("service1")
if err != nil {
   panic(err)
}
Register("method_name_1", func(param string) (string, error) {
   return "name", errors.New("some error")
})
Register("method_name_2", func(param string) (string, error) {
   return "return value", nil
})
Register("method_name_3", func(param string) (string, error) {
   return "return value", nil
})
Register("method_name_4", func(param string) (string, error) {
   return "return value", nil
})

//RPC client side：
Service1Invoker, err := slk.InitRpcClient("service1")
if err != nil {
   panic(err)
}

//async invoke
if err := Service1Invoker(&slink.InvokeParam{
   Method:      "method_name_1",
   Param:       `json string`,
}, func(result string, err error) {
   log.Println(result, err)
}, time.Second*5); err != nil {
   log.Println(err)
}

//sync invoke
slk.SyncInvoker(Service1Invoker,&slink.InvokeParam{
   Method:      "method_name_1",
   Param:       `json string`,
}, func(result string, err error) {
   log.Println(result, err)
}, time.Second*5);err != nil {
   log.Println(err)
}

//multi clients
if clis, err := slk.NewRpcClients([]string{
   "svc1", "svc2", "svc3",
}); err != nil {
   log.Println(err)
} else {
   if err := clis["svc1"](&slink.InvokeParam{
      Method: "GetName",
      Param:  "",
   }, func(result string, err error) {
      log.Println(result, err)
   }, time.Second*5); err != nil {
      log.Println(err)
   }
}

Subscribe & Publish

if err := slk.Subscribe("abc", func(msg []byte) {
	log.Println(string(msg))
}); err != nil {
	panic(err)
}

if err := slk.Publish("abc", "123"); err != nil {
   log.Println(err)
}

if mul, err := slk.NewMultiSubscribe("topic"); err != nil {
   log.Println(err)
} else {
   var a slink.TopicHandler = func(msg []byte) {
      log.Println(string(msg))
   }
   mul.Subscribe(&a)
   var b slink.TopicHandler = func(msg []byte) {
      log.Println(msg)
   }
   mul.Subscribe(&b)
   _ = slk.Publish("topic", "multi")
   mul.Unsubscribe(&a)
   _ = slk.Publish("topic", "end")
}

ms, _ := slk.NewMultiSubscribes([]string{
   "topic", "topic1",
})
var a slink.TopicHandler = func(msg []byte) {
   log.Println(string(msg))
}
ms["topic"].Subscribe(&a)
var b slink.TopicHandler = func(msg []byte) {
   log.Println(string(msg))
}
ms["topic1"].Subscribe(&b)
_ = slk.Publish("topic", "topic")
_ = slk.Publish("topic1", "topic1")

Documentation

Index

• Constants
• func SyncInvoker(invoker Invoker, param *InvokeParam, on OnResult, timeout time.Duration) error
• type InvokeParam
• type Invoker
• type MethodRequest
• type MultiSubscribe
• type OnResult
• type OnServiceUp
• type Register
• type SLink
  • func New(host string) *SLink
  • func (p *SLink) Connect() (MQTT.Client, error)
  • func (p *SLink) GetClient() MQTT.Client
  • func (p *SLink) InitRpcClient(serviceName string) (Invoker, error)
  • func (p *SLink) InitRpcServer(serviceName string) (Register, error)
  • func (p *SLink) NewMultiSubscribe(topic string) (*MultiSubscribe, error)
  • func (p *SLink) NewMultiSubscribes(topics []string) (map[string]*MultiSubscribe, error)
  • func (p *SLink) NewRpcClients(svcnames []string) (map[string]Invoker, error)
  • func (p *SLink) ProvideService(sInfo *ServiceInfo)
  • func (p *SLink) Publish(topic string, msg string) error
  • func (p *SLink) SetDeadTime(t time.Duration) *SLink
  • func (p *SLink) SetKeepAlive(t time.Duration) *SLink
  • func (p *SLink) SetOnServiceUp(serviceName string, onServiceUp OnServiceUp) (ServiceGetter, error)
  • func (p *SLink) SetPassword(password string) *SLink
  • func (p *SLink) SetUserName(name string) *SLink
  • func (p *SLink) Subscribe(topic string, th TopicHandler) error
  • func (p *SLink) Unsubscribe(topic string) error
• type ServiceGetter
• type ServiceInfo
• type TopicHandler

Constants

const (
	ServiceUp        = "usws/svc/up"
	ServiceRpcUp     = "usws/rpc/up"
	ServiceRpcIn     = "usws/rpc/in"
	ServiceRpcOut    = "usws/rpc/out"
	RecyleDuration   = time.Second * 5
	DefaultKeepAlive = time.Second * 5
	DefaultDeadTime  = time.Second * 15
)

Functions

func SyncInvoker

func SyncInvoker(invoker Invoker, param *InvokeParam, on OnResult, timeout time.Duration) error

Types

type InvokeParam

type InvokeParam struct {
	Method string `json:"m"`
	Param  string `json:"p"`
}

type Invoker

type Invoker func(param *InvokeParam, on OnResult, timeout time.Duration) error

type MethodRequest

type MethodRequest func(param string) (string, error)

type MultiSubscribe

type MultiSubscribe struct {
	Subscribe   func(lis *TopicHandler)
	Unsubscribe func(lis *TopicHandler)
}

type OnResult

type OnResult func(result string, err error)

type OnServiceUp

type OnServiceUp func(p *ServiceInfo)

type Register

type Register func(method string, f MethodRequest)

type SLink

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

func New

func New(host string) *SLink

func (*SLink) Connect

func (p *SLink) Connect() (MQTT.Client, error)

func (*SLink) GetClient

func (p *SLink) GetClient() MQTT.Client

func (*SLink) InitRpcClient

func (p *SLink) InitRpcClient(serviceName string) (Invoker, error)

func (*SLink) InitRpcServer

func (p *SLink) InitRpcServer(serviceName string) (Register, error)

func (*SLink) NewMultiSubscribe

func (p *SLink) NewMultiSubscribe(topic string) (*MultiSubscribe, error)

func (*SLink) NewMultiSubscribes

func (p *SLink) NewMultiSubscribes(topics []string) (map[string]*MultiSubscribe, error)

func (*SLink) NewRpcClients

func (p *SLink) NewRpcClients(svcnames []string) (map[string]Invoker, error)

func (*SLink) ProvideService

func (p *SLink) ProvideService(sInfo *ServiceInfo)

func (*SLink) Publish

func (p *SLink) Publish(topic string, msg string) error

func (*SLink) SetDeadTime

func (p *SLink) SetDeadTime(t time.Duration) *SLink

func (*SLink) SetKeepAlive

func (p *SLink) SetKeepAlive(t time.Duration) *SLink

func (*SLink) SetOnServiceUp

func (p *SLink) SetOnServiceUp(serviceName string, onServiceUp OnServiceUp) (ServiceGetter, error)

func (*SLink) SetPassword

func (p *SLink) SetPassword(password string) *SLink

func (*SLink) SetUserName

func (p *SLink) SetUserName(name string) *SLink

func (*SLink) Subscribe

func (p *SLink) Subscribe(topic string, th TopicHandler) error

func (*SLink) Unsubscribe

func (p *SLink) Unsubscribe(topic string) error

type ServiceGetter

type ServiceGetter func() *ServiceInfo

type ServiceInfo

type ServiceInfo struct {
	Name string `json:"n"`
	IP   string `json:"i"`
	Port int    `json:"p"`
}

type TopicHandler

type TopicHandler func(msg []byte)