72 lines
1.5 KiB
Go
72 lines
1.5 KiB
Go
package sync
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/unistack-org/micro/v3/store"
|
|
)
|
|
|
|
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)
|
|
}
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|