@@ -94,7 +94,8 @@ func NewPersistableChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (
94
94
},
95
95
Workers : 0 ,
96
96
},
97
- DataDir : config .DataDir ,
97
+ DataDir : config .DataDir ,
98
+ QueueName : config .Name + "-level" ,
98
99
}
99
100
100
101
levelQueue , err := NewLevelQueue (wrappedHandle , levelCfg , exemplar )
@@ -172,16 +173,18 @@ func (q *PersistableChannelQueue) Run(atShutdown, atTerminate func(func())) {
172
173
atShutdown (q .Shutdown )
173
174
atTerminate (q .Terminate )
174
175
175
- if lq , ok := q .internal .(* LevelQueue ); ok && lq .byteFIFO .Len (lq .shutdownCtx ) != 0 {
176
+ if lq , ok := q .internal .(* LevelQueue ); ok && lq .byteFIFO .Len (lq .terminateCtx ) != 0 {
176
177
// Just run the level queue - we shut it down once it's flushed
177
178
go q .internal .Run (func (_ func ()) {}, func (_ func ()) {})
178
179
go func () {
179
- for ! q .IsEmpty () {
180
- _ = q . internal .Flush (0 )
180
+ for ! lq .IsEmpty () {
181
+ _ = lq .Flush (0 )
181
182
select {
182
183
case <- time .After (100 * time .Millisecond ):
183
- case <- q .internal .(* LevelQueue ).shutdownCtx .Done ():
184
- log .Warn ("LevelQueue: %s shut down before completely flushed" , q .internal .(* LevelQueue ).Name ())
184
+ case <- lq .shutdownCtx .Done ():
185
+ if lq .byteFIFO .Len (lq .terminateCtx ) > 0 {
186
+ log .Warn ("LevelQueue: %s shut down before completely flushed" , q .internal .(* LevelQueue ).Name ())
187
+ }
185
188
return
186
189
}
187
190
}
@@ -316,10 +319,22 @@ func (q *PersistableChannelQueue) Shutdown() {
316
319
// Redirect all remaining data in the chan to the internal channel
317
320
log .Trace ("PersistableChannelQueue: %s Redirecting remaining data" , q .delayedStarter .name )
318
321
close (q .channelQueue .dataChan )
322
+ countOK , countLost := 0 , 0
319
323
for data := range q .channelQueue .dataChan {
320
- _ = q .internal .Push (data )
324
+ err := q .internal .Push (data )
325
+ if err != nil {
326
+ log .Error ("PersistableChannelQueue: %s Unable redirect %v due to: %v" , q .delayedStarter .name , data , err )
327
+ countLost ++
328
+ } else {
329
+ countOK ++
330
+ }
321
331
atomic .AddInt64 (& q .channelQueue .numInQueue , - 1 )
322
332
}
333
+ if countLost > 0 {
334
+ log .Warn ("PersistableChannelQueue: %s %d will be restored on restart, %d lost" , q .delayedStarter .name , countOK , countLost )
335
+ } else if countOK > 0 {
336
+ log .Warn ("PersistableChannelQueue: %s %d will be restored on restart" , q .delayedStarter .name , countOK )
337
+ }
323
338
log .Trace ("PersistableChannelQueue: %s Done Redirecting remaining data" , q .delayedStarter .name )
324
339
325
340
log .Debug ("PersistableChannelQueue: %s Shutdown" , q .delayedStarter .name )
0 commit comments