40b8afb150
Backend stability and security improvements: * internal/util/ — common RandomHex helper, removed 3 duplicates * ESL: deduplicated readMessage (locked/unlocked), net.JoinHostPort for IPv6 * AMI: synchronous reconnect() in readEventsLoop, net.JoinHostPort for IPv6 * Auth: /api/auth/refresh accepts Authorization header only (no ?token=) * decodeJSON: http.MaxBytesReader(1<<20) body limit * Trunks: gatewayParams() uses configured ESL.GatewayPrefix * Config: jwt_secret_env env-var fallback * FSCollector: time.After → time.NewTimer with defer Stop * Monitoring: Prometheus counters (route_requests, nodes_total/healthy, uptime) * go fmt pass across all internal/ packages
91 lines
2.7 KiB
Go
91 lines
2.7 KiB
Go
package esl
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"time"
|
|
)
|
|
|
|
// EventHandler — callback для обработки событий ESL.
|
|
// Принимает имя события и заголовки, вызывается из readEventsLoop.
|
|
type EventHandler func(eventName string, headers map[string]string, body string)
|
|
|
|
// StartEventLoop запускает основной цикл реконнекта и чтения событий.
|
|
// Блокирует до тех пор, пока контекст не будет отменён (ctx.Done()).
|
|
// При успешном подключении:
|
|
// - Вызывает onConnect (например, для GatewaySyncAll)
|
|
// - Подписывается на события SOFIA::gateway_register/unregister
|
|
// - Запускает чтение асинхронных событий
|
|
//
|
|
// При разрыве соединения — автоматический реконнект.
|
|
func (c *Client) StartEventLoop(ctx context.Context, profile string, onConnect func(), onEvent EventHandler) {
|
|
const subscribeEvents = "SOFIA::gateway_register SOFIA::gateway_unregister SOFIA::gateway_expire"
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
c.Disconnect()
|
|
return
|
|
default:
|
|
}
|
|
|
|
// Пытаемся подключиться
|
|
log.Printf("[esl] попытка подключения к %s:%d", c.host, c.port)
|
|
if err := c.Connect(); err != nil {
|
|
log.Printf("[esl] ошибка подключения: %v, повтор через %v", err, c.backoff)
|
|
time.Sleep(c.backoff)
|
|
c.backoff *= 2
|
|
if c.backoff > 60*time.Second {
|
|
c.backoff = 60 * time.Second
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Успешное подключение
|
|
c.backoff = 1 * time.Second
|
|
|
|
// Подписываемся на события
|
|
if err := c.Subscribe(subscribeEvents); err != nil {
|
|
log.Printf("[esl] ошибка подписки: %v", err)
|
|
c.Disconnect()
|
|
continue
|
|
}
|
|
|
|
// Вызываем callback (GatewaySyncAll)
|
|
if onConnect != nil {
|
|
onConnect()
|
|
}
|
|
|
|
// Читаем события пока соединение живо
|
|
readEvents := make(chan struct{})
|
|
go func() {
|
|
defer close(readEvents)
|
|
c.readEventsLoop()
|
|
}()
|
|
|
|
// Основной цикл обработки событий
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
c.Disconnect()
|
|
return
|
|
case <-readEvents:
|
|
// Соединение разорвалось — идём на реконнект
|
|
log.Printf("[esl] чтение событий остановлено")
|
|
goto reconnect
|
|
case evt, ok := <-c.Events():
|
|
if !ok {
|
|
goto reconnect
|
|
}
|
|
if onEvent != nil {
|
|
onEvent(evt.Name, evt.Headers, evt.Body)
|
|
}
|
|
}
|
|
}
|
|
|
|
reconnect:
|
|
c.Disconnect()
|
|
time.Sleep(c.backoff)
|
|
}
|
|
}
|