diff options
Diffstat (limited to 'modules/queue/queue_disk_channel.go')
| -rw-r--r-- | modules/queue/queue_disk_channel.go | 29 |
1 files changed, 22 insertions, 7 deletions
diff --git a/modules/queue/queue_disk_channel.go b/modules/queue/queue_disk_channel.go index c7526714c..91f91f0df 100644 --- a/modules/queue/queue_disk_channel.go +++ b/modules/queue/queue_disk_channel.go @@ -94,7 +94,8 @@ func NewPersistableChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) ( }, Workers: 0, }, - DataDir: config.DataDir, + DataDir: config.DataDir, + QueueName: config.Name + "-level", } levelQueue, err := NewLevelQueue(wrappedHandle, levelCfg, exemplar) @@ -172,16 +173,18 @@ func (q *PersistableChannelQueue) Run(atShutdown, atTerminate func(func())) { atShutdown(q.Shutdown) atTerminate(q.Terminate) - if lq, ok := q.internal.(*LevelQueue); ok && lq.byteFIFO.Len(lq.shutdownCtx) != 0 { + if lq, ok := q.internal.(*LevelQueue); ok && lq.byteFIFO.Len(lq.terminateCtx) != 0 { // Just run the level queue - we shut it down once it's flushed go q.internal.Run(func(_ func()) {}, func(_ func()) {}) go func() { - for !q.IsEmpty() { - _ = q.internal.Flush(0) + for !lq.IsEmpty() { + _ = lq.Flush(0) select { case <-time.After(100 * time.Millisecond): - case <-q.internal.(*LevelQueue).shutdownCtx.Done(): - log.Warn("LevelQueue: %s shut down before completely flushed", q.internal.(*LevelQueue).Name()) + case <-lq.shutdownCtx.Done(): + if lq.byteFIFO.Len(lq.terminateCtx) > 0 { + log.Warn("LevelQueue: %s shut down before completely flushed", q.internal.(*LevelQueue).Name()) + } return } } @@ -316,10 +319,22 @@ func (q *PersistableChannelQueue) Shutdown() { // Redirect all remaining data in the chan to the internal channel log.Trace("PersistableChannelQueue: %s Redirecting remaining data", q.delayedStarter.name) close(q.channelQueue.dataChan) + countOK, countLost := 0, 0 for data := range q.channelQueue.dataChan { - _ = q.internal.Push(data) + err := q.internal.Push(data) + if err != nil { + log.Error("PersistableChannelQueue: %s Unable redirect %v due to: %v", q.delayedStarter.name, data, err) + countLost++ + } else { + countOK++ + } atomic.AddInt64(&q.channelQueue.numInQueue, -1) } + if countLost > 0 { + log.Warn("PersistableChannelQueue: %s %d will be restored on restart, %d lost", q.delayedStarter.name, countOK, countLost) + } else if countOK > 0 { + log.Warn("PersistableChannelQueue: %s %d will be restored on restart", q.delayedStarter.name, countOK) + } log.Trace("PersistableChannelQueue: %s Done Redirecting remaining data", q.delayedStarter.name) log.Debug("PersistableChannelQueue: %s Shutdown", q.delayedStarter.name) |