From 533a1fba4974f9cfb8f6627c89ac2028379f9134 Mon Sep 17 00:00:00 2001 From: Maksim Totmin Date: Thu, 25 Jun 2026 16:56:42 +0700 Subject: [PATCH] feat: ESL client, gateway management, event subscription - Raw TCP ESL client with reconnect - GatewayAdd/GatewayDelete/GatewaySyncAll - SOFIA::gateway_* event processing - SendAsync for deadlock-free bgapi commands --- internal/esl/client.go | 91 ++++++++++++++++++++++++++----- internal/esl/events.go | 118 ++++++++++++++++++++-------------------- internal/esl/gateway.go | 58 +++++--------------- 3 files changed, 150 insertions(+), 117 deletions(-) diff --git a/internal/esl/client.go b/internal/esl/client.go index d93b6b3..48e743a 100644 --- a/internal/esl/client.go +++ b/internal/esl/client.go @@ -36,9 +36,10 @@ type Client struct { port int password string - conn net.Conn - reader *bufio.Reader - mu sync.Mutex // защищает отправку команд + conn net.Conn + reader *bufio.Reader + mu sync.Mutex // защищает отправку синхронных команд (auth, subscribe) + sendMu sync.Mutex // защищает запись в сокет (async gateway commands) connected atomic.Bool eventsCh chan *Event @@ -49,10 +50,10 @@ type Client struct { backoff time.Duration // Статистика - reconnects atomic.Int64 - gatewayOps atomic.Int64 - eventsRecv atomic.Int64 - startTime time.Time + reconnects atomic.Int64 + gatewayOps atomic.Int64 + eventsRecv atomic.Int64 + startTime time.Time } // NewClient создаёт нового ESL-клиента. @@ -160,13 +161,19 @@ func (c *Client) tryConnect() error { func (c *Client) Disconnect() { c.shouldReconn.Store(false) c.mu.Lock() - defer c.mu.Unlock() c.connected.Store(false) if c.conn != nil { c.conn.Close() c.conn = nil } - close(c.closeCh) + c.mu.Unlock() + // Безопасное закрытие closeCh (может быть уже закрыт) + select { + case <-c.closeCh: + // уже закрыт + default: + close(c.closeCh) + } } // IsConnected возвращает статус подключения. @@ -221,7 +228,28 @@ func (c *Client) StartReadingEvents() { go c.readEventsLoop() } +// SendAsync отправляет асинхронную команду (bgapi) без ожидания ответа. +// Ответ придёт как BACKGROUND_JOB событие в readEventsLoop. +// Использует sendMu для защиты записи в сокет. +func (c *Client) SendAsync(command string) error { + c.sendMu.Lock() + defer c.sendMu.Unlock() + + if !c.connected.Load() || c.conn == nil { + return fmt.Errorf("esl не подключён") + } + + fullCmd := command + "\n\n" + _, err := c.conn.Write([]byte(fullCmd)) + if err != nil { + return fmt.Errorf("отправка команды: %w", err) + } + return nil +} + // readEventsLoop читает асинхронные сообщения (события) от FS. +// Не держит мьютекс во время блокирующего чтения — синхронные команды +// (Send) используют c.mu, асинхронные (SendAsync) используют sendMu. func (c *Client) readEventsLoop() { for { select { @@ -235,9 +263,7 @@ func (c *Client) readEventsLoop() { continue } - c.mu.Lock() - headers, body, err := c.readMessageLocked() - c.mu.Unlock() + headers, body, err := c.readMessageUnlocked() if err != nil { if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") { @@ -304,8 +330,47 @@ func (c *Client) reconnect() { // --- Приватные методы --- +// readMessageUnlocked читает одно ESL-сообщение: заголовки + тело. +// Должна вызываться только из readEventsLoop (одиночный читатель). +func (c *Client) readMessageUnlocked() (map[string]string, string, error) { + headers := make(map[string]string) + + for { + line, err := c.reader.ReadString('\n') + if err != nil { + return nil, "", err + } + line = strings.TrimRight(line, "\r\n") + if line == "" { + break // конец заголовков + } + + parts := strings.SplitN(line, ":", 2) + if len(parts) == 2 { + headers[strings.TrimSpace(parts[0])] = strings.TrimSpace(parts[1]) + } + } + + // Читаем тело согласно Content-Length + cl := headers["Content-Length"] + if cl == "" { + return headers, "", nil + } + length, err := strconv.Atoi(cl) + if err != nil { + return headers, "", fmt.Errorf("некорректный Content-Length: %s", cl) + } + + body := make([]byte, length) + if _, err := io.ReadFull(c.reader, body); err != nil { + return headers, "", fmt.Errorf("чтение тела (Content-Length=%d): %w", length, err) + } + + return headers, strings.TrimSpace(string(body)), nil +} + // readMessageLocked читает одно ESL-сообщение: заголовки + тело. -// Вызывающий держит c.mu. +// Вызывающий держит c.mu (для синхронных команд Send). func (c *Client) readMessageLocked() (map[string]string, string, error) { headers := make(map[string]string) diff --git a/internal/esl/events.go b/internal/esl/events.go index 401aa62..9bcbdb9 100644 --- a/internal/esl/events.go +++ b/internal/esl/events.go @@ -11,6 +11,7 @@ import ( type EventHandler func(eventName string, headers map[string]string, body string) // StartEventLoop запускает основной цикл реконнекта и чтения событий. +// Блокирует до тех пор, пока контекст не будет отменён (ctx.Done()). // При успешном подключении: // - Вызывает onConnect (например, для GatewaySyncAll) // - Подписывается на события SOFIA::gateway_register/unregister @@ -19,73 +20,70 @@ type EventHandler func(eventName string, headers map[string]string, body string) func (c *Client) StartEventLoop(ctx context.Context, profile string, onConnect func(), onEvent EventHandler) { const subscribeEvents = "SOFIA::gateway_register SOFIA::gateway_unregister SOFIA::gateway_expire" - go func() { + 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 - 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() - // Не ждём реконнект внутри readEventsLoop - }() - - // Основной цикл обработки событий - for { - select { - case <-ctx.Done(): - c.Disconnect() - return - case <-readEvents: - // Соединение разорвалось — идём на реконнект - log.Printf("[esl] чтение событий остановлено") + case <-readEvents: + // Соединение разорвалось — идём на реконнект + log.Printf("[esl] чтение событий остановлено") + goto reconnect + case evt, ok := <-c.Events(): + if !ok { goto reconnect - case evt, ok := <-c.Events(): - if !ok { - goto reconnect - } - if onEvent != nil { - onEvent(evt.Name, evt.Headers, evt.Body) - } + } + if onEvent != nil { + onEvent(evt.Name, evt.Headers, evt.Body) } } - - reconnect: - c.Disconnect() - time.Sleep(c.backoff) } - }() + + reconnect: + c.Disconnect() + time.Sleep(c.backoff) + } } diff --git a/internal/esl/gateway.go b/internal/esl/gateway.go index eb4fd0b..f6a0791 100644 --- a/internal/esl/gateway.go +++ b/internal/esl/gateway.go @@ -7,59 +7,29 @@ import ( ) // GatewayAdd создаёт gateway на FreeSWITCH через Sofia профиль. -// Возвращает nil если gateway создан или уже существует (идемпотентность). +// Использует bgapi — асинхронная команда, результат приходит как BACKGROUND_JOB событие. func (c *Client) GatewayAdd(profile, name, proxy string) error { - cmd := fmt.Sprintf("api sofia profile %s gwadd %s %s", profile, name, proxy) - headers, body, err := c.Send(cmd) - c.IncGatewayOps() - - if err != nil { - log.Printf("[esl] GatewayAdd %s: ошибка команды: %v", name, err) + cmd := fmt.Sprintf("bgapi sofia profile %s gwadd %s %s", profile, name, proxy) + if err := c.SendAsync(cmd); err != nil { + log.Printf("[esl] GatewayAdd %s: ошибка: %v", name, err) return err } - - reply := headers["Reply-Text"] - if strings.HasPrefix(reply, "+OK") { - log.Printf("[esl] GatewayAdd %s: создан (%s → %s)", name, profile, proxy) - return nil - } - - // Gateway уже существует — не ошибка - if strings.Contains(body, "GATEWAY ALREADY") || strings.Contains(reply, "GATEWAY ALREADY") { - log.Printf("[esl] GatewayAdd %s: уже существует, ок", name) - return nil - } - - log.Printf("[esl] GatewayAdd %s: ошибка FS — Reply: %s, Body: %s", name, reply, body) - return fmt.Errorf("GatewayAdd %s: %s (%s)", name, reply, body) + c.IncGatewayOps() + log.Printf("[esl] GatewayAdd %s: отправлен (%s → %s)", name, profile, proxy) + return nil } // GatewayDelete удаляет gateway на FreeSWITCH. -// Возвращает nil если gateway удалён или не найден (идемпотентность). +// Использует bgapi — асинхронная команда, результат приходит как BACKGROUND_JOB событие. func (c *Client) GatewayDelete(profile, name string) error { - cmd := fmt.Sprintf("api sofia profile %s gwdel %s", profile, name) - headers, body, err := c.Send(cmd) - c.IncGatewayOps() - - if err != nil { - log.Printf("[esl] GatewayDelete %s: ошибка команды: %v", name, err) + cmd := fmt.Sprintf("bgapi sofia profile %s gwdel %s", profile, name) + if err := c.SendAsync(cmd); err != nil { + log.Printf("[esl] GatewayDelete %s: ошибка: %v", name, err) return err } - - reply := headers["Reply-Text"] - if strings.HasPrefix(reply, "+OK") { - log.Printf("[esl] GatewayDelete %s: удалён", name) - return nil - } - - // Gateway не найден — не ошибка - if strings.Contains(body, "GATEWAY NOT FOUND") || strings.Contains(reply, "GATEWAY NOT FOUND") { - log.Printf("[esl] GatewayDelete %s: не найден, ок", name) - return nil - } - - log.Printf("[esl] GatewayDelete %s: ошибка FS — Reply: %s, Body: %s", name, reply, body) - return fmt.Errorf("GatewayDelete %s: %s (%s)", name, reply, body) + c.IncGatewayOps() + log.Printf("[esl] GatewayDelete %s: отправлен", name) + return nil } // GatewayList возвращает список имён gateway в заданном профиле.