533a1fba49
- Raw TCP ESL client with reconnect - GatewayAdd/GatewayDelete/GatewaySyncAll - SOFIA::gateway_* event processing - SendAsync for deadlock-free bgapi commands
443 lines
13 KiB
Go
443 lines
13 KiB
Go
// Package esl реализует клиент FreeSWITCH Event Socket Library (ESL).
|
||
// Использует сырой TCP — протокол ESL прост (HTTP-подобные заголовки +
|
||
// Content-Length тело), не нужна внешняя библиотека.
|
||
//
|
||
// Клиент обеспечивает:
|
||
// - Подключение и аутентификацию
|
||
// - Синхронные команды (api)
|
||
// - Подписку на события
|
||
// - Автоматический реконнект с exponential backoff
|
||
// - Потокобезопасность (mutex на отправку команд)
|
||
package esl
|
||
|
||
import (
|
||
"bufio"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"net"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
)
|
||
|
||
// Event — событие, полученное от FreeSWITCH ESL.
|
||
type Event struct {
|
||
Name string
|
||
Headers map[string]string
|
||
Body string
|
||
}
|
||
|
||
// Client — ESL-клиент для управления FreeSWITCH.
|
||
type Client struct {
|
||
host string
|
||
port int
|
||
password string
|
||
|
||
conn net.Conn
|
||
reader *bufio.Reader
|
||
mu sync.Mutex // защищает отправку синхронных команд (auth, subscribe)
|
||
sendMu sync.Mutex // защищает запись в сокет (async gateway commands)
|
||
|
||
connected atomic.Bool
|
||
eventsCh chan *Event
|
||
closeCh chan struct{}
|
||
|
||
// Переподключение
|
||
shouldReconn atomic.Bool
|
||
backoff time.Duration
|
||
|
||
// Статистика
|
||
reconnects atomic.Int64
|
||
gatewayOps atomic.Int64
|
||
eventsRecv atomic.Int64
|
||
startTime time.Time
|
||
}
|
||
|
||
// NewClient создаёт нового ESL-клиента.
|
||
func NewClient(host string, port int, password string) *Client {
|
||
return &Client{
|
||
host: host,
|
||
port: port,
|
||
password: password,
|
||
eventsCh: make(chan *Event, 100),
|
||
closeCh: make(chan struct{}),
|
||
backoff: 1 * time.Second,
|
||
}
|
||
}
|
||
|
||
// Connect подключается к FreeSWITCH и запускает фоновый реконнект.
|
||
// Блокируется до первой успешной аутентификации.
|
||
func (c *Client) Connect() error {
|
||
c.shouldReconn.Store(true)
|
||
if err := c.tryConnect(); err != nil {
|
||
return fmt.Errorf("esl connect: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ConnectWithRetry подключается в фоне с бесконечными попытками.
|
||
func (c *Client) ConnectWithRetry() {
|
||
c.shouldReconn.Store(true)
|
||
for {
|
||
err := c.tryConnect()
|
||
if err == nil {
|
||
return // успех
|
||
}
|
||
if !c.shouldReconn.Load() {
|
||
return // остановка
|
||
}
|
||
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
|
||
}
|
||
}
|
||
}
|
||
|
||
// tryConnect выполняет одну попытку подключения и аутентификации.
|
||
func (c *Client) tryConnect() error {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
// Закрываем предыдущее соединение если было
|
||
if c.conn != nil {
|
||
c.conn.Close()
|
||
c.closeCh = make(chan struct{})
|
||
}
|
||
|
||
addr := fmt.Sprintf("%s:%d", c.host, c.port)
|
||
dialer := net.Dialer{Timeout: 5 * time.Second}
|
||
conn, err := dialer.Dial("tcp", addr)
|
||
if err != nil {
|
||
return fmt.Errorf("dial %s: %w", addr, err)
|
||
}
|
||
c.conn = conn
|
||
c.reader = bufio.NewReader(conn)
|
||
|
||
// 1. Читаем auth/request
|
||
headers, _, err := c.readMessageLocked()
|
||
if err != nil {
|
||
conn.Close()
|
||
return fmt.Errorf("чтение auth/request: %w", err)
|
||
}
|
||
if headers["Content-Type"] != "auth/request" {
|
||
conn.Close()
|
||
return fmt.Errorf("ожидался auth/request, получен %s", headers["Content-Type"])
|
||
}
|
||
|
||
// 2. Отправляем пароль
|
||
authCmd := fmt.Sprintf("auth %s\n\n", c.password)
|
||
if _, err := conn.Write([]byte(authCmd)); err != nil {
|
||
conn.Close()
|
||
return fmt.Errorf("отправка auth: %w", err)
|
||
}
|
||
|
||
// 3. Читаем auth response
|
||
headers, _, err = c.readMessageLocked()
|
||
if err != nil {
|
||
conn.Close()
|
||
return fmt.Errorf("чтение auth response: %w", err)
|
||
}
|
||
if reply := headers["Reply-Text"]; !strings.HasPrefix(reply, "+OK") {
|
||
conn.Close()
|
||
return fmt.Errorf("auth rejected: %s", reply)
|
||
}
|
||
|
||
c.connected.Store(true)
|
||
c.startTime = time.Now()
|
||
c.backoff = 1 * time.Second // сброс backoff после успешного коннекта
|
||
|
||
endpoint := fmt.Sprintf("%s:%d", c.host, c.port)
|
||
log.Printf("[esl] подключён к %s", endpoint)
|
||
|
||
return nil
|
||
}
|
||
|
||
// Disconnect отключает клиент и останавливает реконнект.
|
||
func (c *Client) Disconnect() {
|
||
c.shouldReconn.Store(false)
|
||
c.mu.Lock()
|
||
c.connected.Store(false)
|
||
if c.conn != nil {
|
||
c.conn.Close()
|
||
c.conn = nil
|
||
}
|
||
c.mu.Unlock()
|
||
// Безопасное закрытие closeCh (может быть уже закрыт)
|
||
select {
|
||
case <-c.closeCh:
|
||
// уже закрыт
|
||
default:
|
||
close(c.closeCh)
|
||
}
|
||
}
|
||
|
||
// IsConnected возвращает статус подключения.
|
||
func (c *Client) IsConnected() bool {
|
||
return c.connected.Load()
|
||
}
|
||
|
||
// Send отправляет синхронную команду и возвращает тело ответа.
|
||
// Пример: Send("api sofia profile external gwlist") → список gateway.
|
||
func (c *Client) Send(command string) (map[string]string, string, error) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
if !c.connected.Load() || c.conn == nil {
|
||
return nil, "", fmt.Errorf("esl не подключён")
|
||
}
|
||
|
||
// Отправляем команду
|
||
fullCmd := command + "\n\n"
|
||
if _, err := c.conn.Write([]byte(fullCmd)); err != nil {
|
||
c.connected.Store(false)
|
||
return nil, "", fmt.Errorf("отправка команды: %w", err)
|
||
}
|
||
|
||
// Читаем ответ
|
||
return c.readMessageLocked()
|
||
}
|
||
|
||
// Subscribe подписывается на события ESL.
|
||
// events — имена событий для фильтра: "SOFIA::gateway_register SOFIA::gateway_unregister".
|
||
func (c *Client) Subscribe(events string) error {
|
||
cmd := fmt.Sprintf("event plain %s", events)
|
||
_, _, err := c.Send(cmd)
|
||
if err != nil {
|
||
return fmt.Errorf("подписка на события: %w", err)
|
||
}
|
||
log.Printf("[esl] подписан на события: %s", events)
|
||
return nil
|
||
}
|
||
|
||
// Events возвращает канал с входящими событиями.
|
||
func (c *Client) Events() <-chan *Event {
|
||
return c.eventsCh
|
||
}
|
||
|
||
// StartReadingEvents запускает горутину чтения асинхронных событий FS.
|
||
// Должна вызываться после Connect().
|
||
func (c *Client) StartReadingEvents() {
|
||
if !c.shouldReconn.Load() {
|
||
return
|
||
}
|
||
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 {
|
||
case <-c.closeCh:
|
||
return
|
||
default:
|
||
}
|
||
|
||
if !c.connected.Load() || c.conn == nil {
|
||
time.Sleep(500 * time.Millisecond)
|
||
continue
|
||
}
|
||
|
||
headers, body, err := c.readMessageUnlocked()
|
||
|
||
if err != nil {
|
||
if err == io.EOF || strings.Contains(err.Error(), "use of closed network connection") {
|
||
c.connected.Store(false)
|
||
log.Printf("[esl] соединение разорвано, реконнект...")
|
||
c.reconnects.Add(1)
|
||
c.reconnect()
|
||
continue
|
||
}
|
||
log.Printf("[esl] ошибка чтения события: %v", err)
|
||
time.Sleep(100 * time.Millisecond)
|
||
continue
|
||
}
|
||
|
||
eventName := headers["Event-Name"]
|
||
if eventName == "" {
|
||
continue // не событие (например, ответ на команду)
|
||
}
|
||
|
||
evt := &Event{
|
||
Name: eventName,
|
||
Headers: headers,
|
||
Body: body,
|
||
}
|
||
c.eventsRecv.Add(1)
|
||
|
||
select {
|
||
case c.eventsCh <- evt:
|
||
default:
|
||
log.Printf("[esl] канал событий переполнен, пропускаем %s", eventName)
|
||
}
|
||
}
|
||
}
|
||
|
||
// reconnect выполняет реконнект с backoff.
|
||
func (c *Client) reconnect() {
|
||
if !c.shouldReconn.Load() {
|
||
return
|
||
}
|
||
log.Printf("[esl] реконнект через %v...", c.backoff)
|
||
time.Sleep(c.backoff)
|
||
|
||
c.mu.Lock()
|
||
if c.conn != nil {
|
||
c.conn.Close()
|
||
c.conn = nil
|
||
}
|
||
c.mu.Unlock()
|
||
|
||
if err := c.tryConnect(); err != nil {
|
||
log.Printf("[esl] реконнект не удался: %v", err)
|
||
c.backoff *= 2
|
||
if c.backoff > 60*time.Second {
|
||
c.backoff = 60 * time.Second
|
||
}
|
||
return
|
||
}
|
||
|
||
// Успешный реконнект
|
||
c.backoff = 1 * time.Second
|
||
log.Printf("[esl] реконнект успешен, переподписываемся на события")
|
||
// TODO: gatewaySyncAll() и resubscribe на события вызываются снаружи
|
||
}
|
||
|
||
// --- Приватные методы ---
|
||
|
||
// 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 (для синхронных команд Send).
|
||
func (c *Client) readMessageLocked() (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
|
||
}
|
||
|
||
// --- Метрики ---
|
||
|
||
// Stats возвращает статистику ESL-клиента для health endpoint.
|
||
type Stats struct {
|
||
Status string `json:"status"`
|
||
Host string `json:"host"`
|
||
Reconnects int64 `json:"reconnects"`
|
||
GatewayOps int64 `json:"gateway_ops_total"`
|
||
EventsRecv int64 `json:"events_received"`
|
||
UptimeSec int64 `json:"uptime_seconds"`
|
||
}
|
||
|
||
// GetStats возвращает текущую статистику.
|
||
func (c *Client) GetStats() Stats {
|
||
st := "connected"
|
||
if !c.connected.Load() {
|
||
st = "disconnected"
|
||
}
|
||
return Stats{
|
||
Status: st,
|
||
Host: fmt.Sprintf("%s:%d", c.host, c.port),
|
||
Reconnects: c.reconnects.Load(),
|
||
GatewayOps: c.gatewayOps.Load(),
|
||
EventsRecv: c.eventsRecv.Load(),
|
||
UptimeSec: int64(time.Since(c.startTime).Seconds()),
|
||
}
|
||
}
|
||
|
||
// IncGatewayOps увеличивает счётчик операций gateway.
|
||
func (c *Client) IncGatewayOps() {
|
||
c.gatewayOps.Add(1)
|
||
}
|