Подправлена работа с redis-ом, добавлена настройка таймаута соединения
This commit is contained in:
@@ -309,6 +309,10 @@ func readConfig() {
|
|||||||
sampleConfig.Port = 6379
|
sampleConfig.Port = 6379
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if sampleConfig.Timeout == 0 {
|
||||||
|
sampleConfig.Timeout = 10
|
||||||
|
}
|
||||||
|
|
||||||
if sampleConfig.Loglevel == "" {
|
if sampleConfig.Loglevel == "" {
|
||||||
sampleConfig.Loglevel = "info"
|
sampleConfig.Loglevel = "info"
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,6 +5,9 @@
|
|||||||
# 6379 if not set
|
# 6379 if not set
|
||||||
"port" : 6379,
|
"port" : 6379,
|
||||||
|
|
||||||
|
# Redis timeout, 10 seconds if not set
|
||||||
|
"timeout" : 10,
|
||||||
|
|
||||||
# info if not set
|
# info if not set
|
||||||
"loglevel" : "info",
|
"loglevel" : "info",
|
||||||
|
|
||||||
|
|||||||
40
main.go
40
main.go
@@ -1,7 +1,6 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
@@ -39,13 +38,18 @@ func init() {
|
|||||||
default:
|
default:
|
||||||
log.SetLevel(log.InfoLevel)
|
log.SetLevel(log.InfoLevel)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Иницализируем клиента Редиски
|
||||||
|
redisClient = redis.NewClient(&redis.Options{
|
||||||
|
Addr: fmt.Sprintf("%s:%d", config.Server, config.Port),
|
||||||
|
}).WithContext(ctx).WithTimeout(time.Duration(config.Timeout) * time.Second)
|
||||||
|
|
||||||
|
// Обозначим, что хотим после соединения подписаться на события из канала config.Channel
|
||||||
|
subscriber = redisClient.Subscribe(ctx, config.Channel)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Основная функция программы, не добавить и не убавить
|
// Основная функция программы, не добавить и не убавить
|
||||||
func main() {
|
func main() {
|
||||||
// Main context
|
|
||||||
var ctx = context.Background()
|
|
||||||
|
|
||||||
// Откроем лог и скормим его логгеру
|
// Откроем лог и скормим его логгеру
|
||||||
if config.Log != "" {
|
if config.Log != "" {
|
||||||
logfile, err := os.OpenFile(config.Log, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0644)
|
logfile, err := os.OpenFile(config.Log, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0644)
|
||||||
@@ -57,14 +61,6 @@ func main() {
|
|||||||
log.SetOutput(logfile)
|
log.SetOutput(logfile)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Иницализируем клиента
|
|
||||||
redisClient = redis.NewClient(&redis.Options{
|
|
||||||
Addr: fmt.Sprintf("%s:%d", config.Server, config.Port),
|
|
||||||
})
|
|
||||||
|
|
||||||
log.Debugf("Lazy connect() to redis at %s:%d", config.Server, config.Port)
|
|
||||||
subscriber = redisClient.Subscribe(ctx, config.Channel)
|
|
||||||
|
|
||||||
// Самое время поставить траппер сигналов
|
// Самое время поставить траппер сигналов
|
||||||
signal.Notify(sigChan,
|
signal.Notify(sigChan,
|
||||||
syscall.SIGINT,
|
syscall.SIGINT,
|
||||||
@@ -73,25 +69,13 @@ func main() {
|
|||||||
|
|
||||||
go sigHandler()
|
go sigHandler()
|
||||||
|
|
||||||
// Обработчик событий от редиски
|
// Начнём выгребать события из редиски (длина ковеера/буфера канала по-умолчанию - 100 сообщений)
|
||||||
for {
|
ch := subscriber.Channel()
|
||||||
if shutdown {
|
|
||||||
time.Sleep(1 * time.Second)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
msg, err := subscriber.ReceiveMessage(ctx)
|
for msg := range ch {
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
if !shutdown {
|
if !shutdown {
|
||||||
log.Warnf("Unable to connect to redis at %s:%d: %s", config.Server, config.Port, err)
|
msgParser(ctx, msg.Payload)
|
||||||
}
|
}
|
||||||
|
|
||||||
time.Sleep(1 * time.Second)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
go msgParser(ctx, msg.Payload)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
1
types.go
1
types.go
@@ -4,6 +4,7 @@ package main
|
|||||||
type myConfig struct {
|
type myConfig struct {
|
||||||
Server string `json:"server,omitempty"`
|
Server string `json:"server,omitempty"`
|
||||||
Port int `json:"port,omitempty"`
|
Port int `json:"port,omitempty"`
|
||||||
|
Timeout int `json:"timeout,omitempty"`
|
||||||
Loglevel string `json:"loglevel,omitempty"`
|
Loglevel string `json:"loglevel,omitempty"`
|
||||||
Log string `json:"log,omitempty"`
|
Log string `json:"log,omitempty"`
|
||||||
Channel string `json:"channel,omitempty"`
|
Channel string `json:"channel,omitempty"`
|
||||||
|
|||||||
Reference in New Issue
Block a user