package pubsub
|
|
import (
|
"context"
|
"encoding/json"
|
"fmt"
|
"nanomsg.org/go-mangos"
|
"nanomsg.org/go-mangos/protocol/pub"
|
"nanomsg.org/go-mangos/protocol/sub"
|
"nanomsg.org/go-mangos/transport/ipc"
|
"nanomsg.org/go-mangos/transport/tcp"
|
)
|
|
type mangosPubSub struct {
|
url string
|
|
ctx context.Context
|
|
sock mangos.Socket
|
|
pubCh chan []byte //publish msg chan
|
|
recvCh chan Message //recv msg chan
|
}
|
|
func newPub(url string) (*mangosPubSub,error) {
|
var sock mangos.Socket
|
var err error
|
|
sock, err = pub.NewSocket()
|
if err != nil {
|
return nil, err
|
}
|
sock.AddTransport(ipc.NewTransport())
|
sock.AddTransport(tcp.NewTransport())
|
|
err = sock.Listen(url)
|
if err != nil {
|
return nil, err
|
}
|
ctx, cancel := context.WithCancel(context.Background())
|
pub := &mangosPubSub{
|
url: url,
|
ctx: ctx,
|
sock: sock,
|
pubCh: make(chan []byte),
|
}
|
go func() {
|
for {
|
select {
|
case <-ctx.Done():
|
close(pub.pubCh)
|
cancel()
|
return
|
case msg := <-pub.pubCh:
|
err := pub.sock.Send(msg)
|
if err != nil {
|
fmt.Println("Error PUBLISH MSG to the socket:", err.Error())
|
}
|
}
|
}
|
}()
|
return pub,nil
|
}
|
|
func newSub(url string, topics []string) (*mangosPubSub,error) {
|
var sock mangos.Socket
|
var err error
|
|
sock, err = sub.NewSocket()
|
if err != nil {
|
return nil, err
|
}
|
sock.AddTransport(ipc.NewTransport())
|
sock.AddTransport(tcp.NewTransport())
|
|
err = sock.Dial(url)
|
if err != nil {
|
return nil, err
|
}
|
// subscribes to everything
|
err = sock.SetOption(mangos.OptionSubscribe, []byte(""))
|
if err != nil {
|
return nil, err
|
}
|
ctx, cancel := context.WithCancel(context.Background())
|
sub := &mangosPubSub{
|
url:url,
|
ctx: ctx,
|
sock: sock,
|
recvCh: make(chan Message,50),
|
}
|
|
var msg []byte
|
go func() {
|
for {
|
select {
|
case <-ctx.Done():
|
close(sub.recvCh)
|
cancel()
|
return
|
default:
|
msg, err = sub.sock.Recv()
|
if err != nil {
|
fmt.Println("Cannot SUBSCRIBE MSG,ERR:", err.Error())
|
} else {
|
//判断是否是想要的主题消息
|
var recvMsg Message
|
if unmarshlErr := json.Unmarshal(msg, &recvMsg);unmarshlErr ==nil {
|
if matchTopic(recvMsg.Topic, topics) {
|
sub.recvCh <- recvMsg
|
}
|
}
|
}
|
}
|
}
|
}()
|
|
return sub,nil
|
}
|
|
func matchTopic(topic string,subTopics []string) bool {
|
if subTopics ==nil && len(subTopics) ==0 {
|
return true
|
}
|
for _,t := range subTopics {
|
if topic == t {
|
return true
|
}
|
}
|
return false
|
}
|
|
func (ps *mangosPubSub) Publish(msg []byte) {
|
ps.pubCh <- msg
|
}
|
|
func (ps *mangosPubSub) Recv() chan Message {
|
return ps.recvCh
|
}
|
