Add cancellable context for Redis operation to prevent hanging redis ops on shutdown
- Rename Ctx to RedisCtx with CancelRedisCtx function - Signal handler now cancels context before closing Redis
This commit is contained in:
@@ -40,11 +40,11 @@ func main() {
|
|||||||
MaxRetries: -1,
|
MaxRetries: -1,
|
||||||
MinRetryBackoff: time.Second,
|
MinRetryBackoff: time.Second,
|
||||||
MaxRetryBackoff: 30 * time.Second,
|
MaxRetryBackoff: 30 * time.Second,
|
||||||
}).WithContext(misc.Ctx).WithTimeout(time.Duration(misc.Config.Timeout) * time.Second)
|
}).WithContext(misc.RedisCtx).WithTimeout(time.Duration(misc.Config.Timeout) * time.Second)
|
||||||
|
|
||||||
// Проверим, что редиска доступна. Повторяем попытки, пока не подключится.
|
// Проверим, что редиска доступна. Повторяем попытки, пока не подключится.
|
||||||
for {
|
for {
|
||||||
if err := misc.RedisClient.Ping(misc.Ctx).Err(); err == nil {
|
if err := misc.RedisClient.Ping(misc.RedisCtx).Err(); err == nil {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -53,7 +53,7 @@ func main() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Обозначим, что хотим после соединения подписаться на события из канала config.Channel.
|
// Обозначим, что хотим после соединения подписаться на события из канала config.Channel.
|
||||||
misc.Subscriber = misc.RedisClient.Subscribe(misc.Ctx, misc.Config.Channel)
|
misc.Subscriber = misc.RedisClient.Subscribe(misc.RedisCtx, misc.Config.Channel)
|
||||||
|
|
||||||
// Самое время поставить траппер сигналов.
|
// Самое время поставить траппер сигналов.
|
||||||
signal.Notify(misc.SigChan,
|
signal.Notify(misc.SigChan,
|
||||||
@@ -69,7 +69,7 @@ func main() {
|
|||||||
log.Info("Aleesa-misc-go started")
|
log.Info("Aleesa-misc-go started")
|
||||||
|
|
||||||
for msg := range ch {
|
for msg := range ch {
|
||||||
misc.MsgParser(misc.Ctx, msg.Payload)
|
misc.MsgParser(misc.RedisCtx, msg.Payload)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package misc
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"os"
|
"os"
|
||||||
|
"sync"
|
||||||
|
|
||||||
"github.com/go-redis/redis/v8"
|
"github.com/go-redis/redis/v8"
|
||||||
)
|
)
|
||||||
@@ -20,7 +21,14 @@ var Subscriber *redis.PubSub
|
|||||||
// Канал, в который приходят уведомления для хэндлера сигналов от траппера сигналов.
|
// Канал, в который приходят уведомления для хэндлера сигналов от траппера сигналов.
|
||||||
var SigChan = make(chan os.Signal, 1)
|
var SigChan = make(chan os.Signal, 1)
|
||||||
|
|
||||||
// Main context.
|
// RedisCtx - контекст для операций с редиской. Отменяется при завершении.
|
||||||
var Ctx = context.Background()
|
var RedisCtx context.Context
|
||||||
|
var CancelRedisCtx context.CancelFunc
|
||||||
|
|
||||||
|
var once sync.Once
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
RedisCtx, CancelRedisCtx = context.WithCancel(context.Background())
|
||||||
|
}
|
||||||
|
|
||||||
/* vim: set ft=go noet ai ts=4 sw=4 sts=4: */
|
/* vim: set ft=go noet ai ts=4 sw=4 sts=4: */
|
||||||
|
|||||||
@@ -28,8 +28,11 @@ func SigHandler() {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Отменяем контекст редиски, чтобы все ожидающие операции завершились.
|
||||||
|
CancelRedisCtx()
|
||||||
|
|
||||||
// Отпишемся от всех каналов и закроем коннект к редиске.
|
// Отпишемся от всех каналов и закроем коннект к редиске.
|
||||||
if err = Subscriber.Unsubscribe(Ctx); err != nil {
|
if err = Subscriber.Unsubscribe(RedisCtx); err != nil {
|
||||||
log.Errorf("Unable to unsubscribe from redis channels cleanly: %s", err)
|
log.Errorf("Unable to unsubscribe from redis channels cleanly: %s", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user