diff --git a/config/config.toml b/config/config.toml index 26e2268..114ce4e 100644 --- a/config/config.toml +++ b/config/config.toml @@ -59,7 +59,7 @@ receiveBuffer = 4096 marketSubscribeLimit = 16 consumeBatch = 1024 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume -httpProxy = "http://192.168.1.5:7890" +httpProxy = "http://192.168.1.6:7890" # 模拟盘API交易地址如下: # REST:https://www.okx.com diff --git a/pkg/utils/promise/promise.go b/pkg/utils/promise/promise.go new file mode 100644 index 0000000..4e5d756 --- /dev/null +++ b/pkg/utils/promise/promise.go @@ -0,0 +1,117 @@ +package promise + +import ( + "errors" + "sync" + "time" +) + +var ( + promiseTicker *time.Ticker + promiseTicks []*PromiseAll // todo 内存释放 capacity / len > ? + promiseTickerLock sync.Mutex + + ErrorTimeout error = errors.New("promise timeout") + NoExpire time.Duration = 0 +) + +func init() { + promiseTicker = time.NewTicker(100 * time.Millisecond) + go func() { + for { + now := <-promiseTicker.C + promiseTickerLock.Lock() + index := 0 + for _, tick := range promiseTicks { + if finish := tick.tick(now); !finish { + promiseTicks[index] = tick + index++ + } + } + promiseTicks = promiseTicks[:index] + promiseTickerLock.Unlock() + } + }() +} + +type PromiseAll struct { + stime time.Time + timeout time.Duration + subs int + keys map[string]bool + data map[string]any + finallyCall func(data map[string]any, err error) + sync.Mutex +} + +func NewPromiseAll(timeout time.Duration) *PromiseAll { + return &PromiseAll{ + timeout: timeout, + keys: make(map[string]bool), + data: make(map[string]any), + } +} + +func (p *PromiseAll) tick(now time.Time) (finish bool) { + if p.stime.Add(p.timeout).After(now) { + p.finish(ErrorTimeout) + return true + } + return p.finallyCall != nil +} + +func (p *PromiseAll) finish(err error) { + if p.finallyCall == nil { + return + } + // 执行结束回调函数 + go p.finallyCall(p.data, err) + // promise 状态结束 + p.finallyCall = nil +} + +func (p *PromiseAll) Subscribe(keys ...string) *PromiseAll { + for _, key := range keys { + p.keys[key] = false + } + p.subs = len(p.keys) + return p +} + +func (p *PromiseAll) Finally(func(data map[string]any, err error)) *PromiseAll { + if p.subs <= 0 { + p.finish(nil) + return p + } + + if p.timeout != NoExpire { + promiseTickerLock.Lock() + promiseTicks = append(promiseTicks, p) + promiseTickerLock.Unlock() + p.stime = time.Now() + } + return p +} + +func (p *PromiseAll) Update(k string, v any, err error) *PromiseAll { + // options: error stop, concurrent limit + p.Lock() + defer p.Unlock() + + finish, ok := p.keys[k] + if !ok || finish { + return p + } + p.keys[k] = true + p.subs-- + if err != nil { + p.finish(err) + return p + } + + p.data[k] = v + if p.subs <= 0 { + p.finish(nil) + } + return p +} diff --git a/pkg/utils/promise/promise_test.go b/pkg/utils/promise/promise_test.go new file mode 100644 index 0000000..f9b3f4f --- /dev/null +++ b/pkg/utils/promise/promise_test.go @@ -0,0 +1,12 @@ +package promise + +import ( + "testing" +) + +func TestPromiseAll(t *testing.T) { + v1, v2, v3 := 1, 2, 3 + _, _, _ = v1, v2, v3 + // pub kline 1 data, i1 1 data, pub i2 1 data, pub i3 1 data + // resolve 1 [data, data, data, data] +}