package common
|
|
import (
|
"container/list"
|
"context"
|
"sync"
|
"time"
|
|
"basic.com/valib/deliver.git"
|
// "basic.com/pubsub/protomsg.git"
|
// "github.com/gogo/protobuf/proto"
|
)
|
|
type runResult struct {
|
data []byte
|
valid bool
|
}
|
|
// ToRule ipc
|
type ToRule struct {
|
ipcURL string
|
maxSize int
|
cache *list.List
|
cv *sync.Cond
|
cond bool
|
fnLogger func(...interface{})
|
}
|
|
// NewToRule send to ruleprocess
|
func NewToRule(ipcURL string, maxSize int, fn func(...interface{})) *ToRule {
|
return &ToRule{
|
ipcURL: ipcURL,
|
maxSize: maxSize,
|
cache: list.New(),
|
cv: sync.NewCond(&sync.Mutex{}),
|
cond: false,
|
fnLogger: fn,
|
}
|
}
|
|
// Push data
|
func (t *ToRule) Push(data []byte, valid bool) {
|
|
t.cv.L.Lock()
|
result := runResult{data, valid}
|
t.cache.PushBack(result)
|
if t.cache.Len() > t.maxSize {
|
for i := 0; i < t.cache.Len(); {
|
d := t.cache.Front().Value.(runResult)
|
if d.valid == false {
|
t.cache.Remove(t.cache.Front())
|
i = i + 2
|
} else {
|
i = i + 1
|
}
|
}
|
}
|
if t.cache.Len() > t.maxSize {
|
for i := 0; i < t.cache.Len(); {
|
t.cache.Remove(t.cache.Front())
|
i = i + 2
|
}
|
}
|
// logo.Infof("push to cache count : %d\n", t.cache.Len())
|
t.cond = true
|
t.cv.Signal()
|
t.cv.L.Unlock()
|
}
|
|
// Run forever
|
func (t *ToRule) Run(ctx context.Context) {
|
|
var i deliver.Deliver
|
var err error
|
|
for {
|
i, err = deliver.NewClientWithError(deliver.PushPull, t.ipcURL)
|
if err != nil {
|
time.Sleep(time.Second)
|
t.fnLogger("wait create to rule ipc", err)
|
continue
|
}
|
break
|
}
|
|
count := 0
|
|
for {
|
select {
|
case <-ctx.Done():
|
return
|
default:
|
|
var d []byte
|
t.cv.L.Lock()
|
|
for !t.cond {
|
t.cv.Wait()
|
}
|
|
for j := 0; j < 8; j++ {
|
if t.cache.Len() <= 0 {
|
break
|
}
|
|
d = t.cache.Front().Value.(runResult).data
|
if i != nil && d != nil {
|
|
err := i.Send(d)
|
if err != nil {
|
t.fnLogger("!!!!!!!!!!!!!!!!!!!!!!!!!!!", err)
|
} else {
|
count++
|
if count > 5 {
|
count = 0
|
t.fnLogger("~~~~~~SEND TO RULE CORRECT")
|
}
|
}
|
}
|
t.cache.Remove(t.cache.Front())
|
}
|
|
t.cond = false
|
t.cv.L.Unlock()
|
|
}
|
}
|
}
|
