package sdk
|
|
import (
|
"analysis/logo"
|
"analysis/work"
|
"container/list"
|
"context"
|
"sync"
|
"time"
|
)
|
|
// LockList list
|
type LockList struct {
|
cache *list.List
|
cv *sync.Cond
|
cond bool
|
size int
|
}
|
|
// NewLockList new
|
func NewLockList(size int) *LockList {
|
return &LockList{
|
cache: list.New(),
|
cv: sync.NewCond(&sync.Mutex{}),
|
cond: false,
|
size: size,
|
}
|
}
|
|
// Push push
|
func (l *LockList) Push(v interface{}) {
|
l.cv.L.Lock()
|
l.cache.PushBack(v)
|
|
for l.cache.Len() > l.size {
|
l.cache.Remove(l.cache.Front())
|
}
|
|
l.cond = true
|
l.cv.Signal()
|
l.cv.L.Unlock()
|
}
|
|
// Pop pop
|
func (l *LockList) Pop() interface{} {
|
l.cv.L.Lock()
|
|
for !l.cond {
|
l.cv.Wait()
|
}
|
|
elem := l.cache.Front().Value
|
|
l.cache.Remove(l.cache.Front())
|
l.cond = false
|
l.cv.L.Unlock()
|
|
return elem
|
}
|
|
/////////////////////////////////////////////////////////////////
|
|
func flowSimpleWork(ctx context.Context, out chan<- work.MsgRS, typ string,
|
fnConsume func() interface{}, fnRun func(work.MsgRS, chan<- work.MsgRS, string)) {
|
|
tm := time.Now()
|
sc := 0
|
|
for {
|
select {
|
case <-ctx.Done():
|
return
|
default:
|
|
rMsg := fnConsume().(work.MsgRS)
|
|
fnRun(rMsg, out, typ)
|
|
sc++
|
if sc == 25 {
|
logo.Infoln(typ, " RUN 25 FRAME USE TIME: ", time.Since(tm))
|
sc = 0
|
tm = time.Now()
|
}
|
if time.Since(tm) > time.Second {
|
logo.Infof("%s RUN %d FRAME USE TIME: %v", typ, sc, time.Since(tm))
|
sc = 0
|
tm = time.Now()
|
}
|
}
|
}
|
|
}
|
|
// FlowSimple wrap
|
func FlowSimple(ctx context.Context, in <-chan work.MsgRS, out chan<- work.MsgRS, typ string,
|
fnProduce func(interface{}), fnConsume func() interface{},
|
fnRun func(work.MsgRS, chan<- work.MsgRS, string), fnClose func()) {
|
|
go flowSimpleWork(ctx, out, typ, fnConsume, fnRun)
|
|
for {
|
select {
|
case <-ctx.Done():
|
fnClose()
|
return
|
default:
|
rMsg := <-in
|
if !validRemoteMessage(rMsg, typ) {
|
logo.Errorln(typ, " validremotemessage invalid")
|
ejectResult(nil, rMsg, out)
|
continue
|
}
|
fnProduce(rMsg)
|
}
|
}
|
}
|
