micro/util/sync/manager.go

72 lines
1.5 KiB
Go
Raw Normal View History

2020-04-11 11:33:10 +03:00
package sync
import (
"fmt"
"time"
"github.com/unistack-org/micro/v3/store"
)
2020-04-11 11:33:10 +03:00
func (c *syncStore) syncManager() {
tickerAggregator := make(chan struct{ index int })
for i, ticker := range c.pendingWriteTickers {
go func(index int, c chan struct{ index int }, t *time.Ticker) {
for range t.C {
c <- struct{ index int }{index: index}
}
}(i, tickerAggregator, ticker)
}
for {
select {
case i := <-tickerAggregator:
println(i.index, "ticked")
c.processQueue(i.index)
}
}
}
2020-04-11 11:33:10 +03:00
func (c *syncStore) processQueue(index int) {
c.Lock()
defer c.Unlock()
q := c.pendingWrites[index]
for i := 0; i < q.Len(); i++ {
r, ok := q.PopFront()
if !ok {
panic(fmt.Errorf("retrieved an invalid value from the L%d sync queue", index+1))
}
ir, ok := r.(*internalRecord)
if !ok {
panic(fmt.Errorf("retrieved a non-internal record from the L%d sync queue", index+1))
}
if !ir.expiresAt.IsZero() && time.Now().After(ir.expiresAt) {
continue
}
nr := &store.Record{
Key: ir.key,
}
nr.Value = make([]byte, len(ir.value))
copy(nr.Value, ir.value)
if !ir.expiresAt.IsZero() {
nr.Expiry = time.Until(ir.expiresAt)
}
// Todo = internal queue also has to hold the corresponding store.WriteOptions
if err := c.syncOpts.Stores[index+1].Write(c.storeOpts.Context, nr); err != nil {
// some error, so queue for retry and bail
q.PushBack(ir)
return
}
}
}
func intpow(x, y int64) int64 {
result := int64(1)
for 0 != y {
if 0 != (y & 1) {
result *= x
}
y >>= 1
x *= x
}
return result
}