package common
|
|
import (
|
"context"
|
|
"time"
|
|
"basic.com/pubsub/protomsg.git"
|
"basic.com/valib/deliver.git"
|
"github.com/gogo/protobuf/proto"
|
)
|
|
// Reciever recv from ipc
|
type Reciever struct {
|
ctx context.Context
|
ipcURL string
|
chMsg chan<- MsgRS
|
|
shm bool
|
fnLogger func(...interface{})
|
}
|
|
// NewReciever new recv
|
func NewReciever(url string, chMsg chan<- MsgRS, shm bool, fn func(...interface{})) *Reciever {
|
return &Reciever{
|
ipcURL: url,
|
chMsg: chMsg,
|
shm: shm,
|
fnLogger: fn,
|
}
|
}
|
|
func (r *Reciever) unserilizeProto(ctx context.Context, data <-chan []byte) {
|
for {
|
select {
|
case <-ctx.Done():
|
return
|
default:
|
d := <-data
|
if len(d) < 100 {
|
continue
|
}
|
// logo.Infoln(len(d), "reciver数据")
|
msg := protomsg.SdkMessage{}
|
if err := proto.Unmarshal(d, &msg); err != nil {
|
r.fnLogger(err, " msg 处理异常")
|
continue
|
}
|
outMsg := MsgRS{Msg: msg}
|
r.chMsg <- outMsg
|
}
|
}
|
}
|
|
// Run run a IPC client
|
func (r *Reciever) Run(ctx context.Context) {
|
|
if r.shm {
|
r.runShm(ctx)
|
} else {
|
r.run(ctx, deliver.NewClient(mode, r.ipcURL))
|
}
|
}
|
|
func (r *Reciever) run(ctx context.Context, i deliver.Deliver) {
|
|
dataChan := make(chan []byte, 3)
|
|
go r.unserilizeProto(ctx, dataChan)
|
|
count := 0
|
|
for {
|
select {
|
case <-ctx.Done():
|
i.Close()
|
return
|
default:
|
|
if r.shm {
|
if d, err := i.Recv(); err != nil {
|
i.Close()
|
r.fnLogger("ANALYSIS RECV ERROR: ", err)
|
|
c, err := deliver.NewClientWithError(deliver.Shm, r.ipcURL)
|
for {
|
if err == nil {
|
break
|
}
|
time.Sleep(time.Second)
|
c, err = deliver.NewClientWithError(deliver.Shm, r.ipcURL)
|
r.fnLogger("ANALYSIS CREATE FAILED : ", err)
|
}
|
i = c
|
r.fnLogger("ANALYSIS CREATE SHM")
|
} else {
|
if d != nil {
|
count++
|
if count > 10 {
|
count = 0
|
r.fnLogger("~~~shm recv image:", len(d))
|
}
|
dataChan <- d
|
}
|
}
|
} else {
|
if msg, err := i.Recv(); err != nil {
|
// logo.Errorln("recv error : ", err, " url: ", r.ipcURL)
|
} else {
|
count++
|
if count > 10 {
|
count = 0
|
r.fnLogger("~~~mangos recv image:", len(msg))
|
}
|
dataChan <- msg
|
}
|
}
|
|
}
|
}
|
}
|
|
func (r *Reciever) runShm(ctx context.Context) {
|
c, err := deliver.NewClientWithError(deliver.Shm, r.ipcURL)
|
for {
|
if err == nil {
|
break
|
}
|
time.Sleep(1 * time.Second)
|
c, err = deliver.NewClientWithError(deliver.Shm, r.ipcURL)
|
r.fnLogger("CLIENT CREATE FAILED : ", err)
|
}
|
r.run(ctx, c)
|
}
|
