Merge pull request #6 from dj-oyu/fix/dual-http-server-bind
fix: eliminate dual HTTP server bind on gateway port
This commit is contained in:
commit
a6f5bdbd0a
2 changed files with 11 additions and 49 deletions
|
|
@ -11,7 +11,6 @@ import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"math"
|
"math"
|
||||||
"net/http"
|
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
|
@ -80,8 +79,6 @@ type Manager struct {
|
||||||
config *config.Config
|
config *config.Config
|
||||||
mediaStore media.MediaStore
|
mediaStore media.MediaStore
|
||||||
dispatchTask *asyncTask
|
dispatchTask *asyncTask
|
||||||
mux *http.ServeMux
|
|
||||||
httpServer *http.Server
|
|
||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
placeholders sync.Map // "channel:chatID" → placeholderID (string)
|
placeholders sync.Map // "channel:chatID" → placeholderID (string)
|
||||||
typingStops sync.Map // "channel:chatID" → func()
|
typingStops sync.Map // "channel:chatID" → func()
|
||||||
|
|
@ -270,41 +267,32 @@ func (m *Manager) initChannels() error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetupHTTPServer creates a shared HTTP server with the given listen address.
|
// SetupHTTPServer registers channel webhook handlers and health checkers
|
||||||
// It registers health endpoints from the health server and discovers channels
|
// onto the provided health server's mux. The health server owns the HTTP
|
||||||
// that implement WebhookHandler and/or HealthChecker to register their handlers.
|
// listener; the channel manager no longer creates its own http.Server.
|
||||||
func (m *Manager) SetupHTTPServer(addr string, healthServer *health.Server) {
|
func (m *Manager) SetupHTTPServer(_ string, healthServer *health.Server) {
|
||||||
m.mux = http.NewServeMux()
|
if healthServer == nil {
|
||||||
|
return
|
||||||
// Register health endpoints
|
|
||||||
if healthServer != nil {
|
|
||||||
healthServer.RegisterOnMux(m.mux)
|
|
||||||
}
|
}
|
||||||
|
mux := healthServer.Mux()
|
||||||
|
|
||||||
// Discover and register webhook handlers and health checkers
|
// Discover and register webhook handlers and health checkers
|
||||||
for name, ch := range m.channels {
|
for name, ch := range m.channels {
|
||||||
if wh, ok := ch.(WebhookHandler); ok {
|
if wh, ok := ch.(WebhookHandler); ok {
|
||||||
m.mux.Handle(wh.WebhookPath(), wh)
|
mux.Handle(wh.WebhookPath(), wh)
|
||||||
logger.InfoCF("channels", "Webhook handler registered", map[string]any{
|
logger.InfoCF("channels", "Webhook handler registered", map[string]any{
|
||||||
"channel": name,
|
"channel": name,
|
||||||
"path": wh.WebhookPath(),
|
"path": wh.WebhookPath(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
if hc, ok := ch.(HealthChecker); ok {
|
if hc, ok := ch.(HealthChecker); ok {
|
||||||
m.mux.HandleFunc(hc.HealthPath(), hc.HealthHandler)
|
mux.HandleFunc(hc.HealthPath(), hc.HealthHandler)
|
||||||
logger.InfoCF("channels", "Health endpoint registered", map[string]any{
|
logger.InfoCF("channels", "Health endpoint registered", map[string]any{
|
||||||
"channel": name,
|
"channel": name,
|
||||||
"path": hc.HealthPath(),
|
"path": hc.HealthPath(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
m.httpServer = &http.Server{
|
|
||||||
Addr: addr,
|
|
||||||
Handler: m.mux,
|
|
||||||
ReadTimeout: 30 * time.Second,
|
|
||||||
WriteTimeout: 30 * time.Second,
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *Manager) StartAll(ctx context.Context) error {
|
func (m *Manager) StartAll(ctx context.Context) error {
|
||||||
|
|
@ -346,20 +334,6 @@ func (m *Manager) StartAll(ctx context.Context) error {
|
||||||
// Start the TTL janitor that cleans up stale typing/placeholder entries
|
// Start the TTL janitor that cleans up stale typing/placeholder entries
|
||||||
go m.runTTLJanitor(dispatchCtx)
|
go m.runTTLJanitor(dispatchCtx)
|
||||||
|
|
||||||
// Start shared HTTP server if configured
|
|
||||||
if m.httpServer != nil {
|
|
||||||
go func() {
|
|
||||||
logger.InfoCF("channels", "Shared HTTP server listening", map[string]any{
|
|
||||||
"addr": m.httpServer.Addr,
|
|
||||||
})
|
|
||||||
if err := m.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
|
||||||
logger.ErrorCF("channels", "Shared HTTP server error", map[string]any{
|
|
||||||
"error": err.Error(),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
|
|
||||||
logger.InfoC("channels", "All channels started")
|
logger.InfoC("channels", "All channels started")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -370,18 +344,6 @@ func (m *Manager) StopAll(ctx context.Context) error {
|
||||||
|
|
||||||
logger.InfoC("channels", "Stopping all channels")
|
logger.InfoC("channels", "Stopping all channels")
|
||||||
|
|
||||||
// Shutdown shared HTTP server first
|
|
||||||
if m.httpServer != nil {
|
|
||||||
shutdownCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
||||||
defer cancel()
|
|
||||||
if err := m.httpServer.Shutdown(shutdownCtx); err != nil {
|
|
||||||
logger.ErrorCF("channels", "Shared HTTP server shutdown error", map[string]any{
|
|
||||||
"error": err.Error(),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
m.httpServer = nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Cancel dispatcher
|
// Cancel dispatcher
|
||||||
if m.dispatchTask != nil {
|
if m.dispatchTask != nil {
|
||||||
m.dispatchTask.cancel()
|
m.dispatchTask.cancel()
|
||||||
|
|
|
||||||
|
|
@ -48,8 +48,8 @@ func NewServer(host string, port int) *Server {
|
||||||
s.server = &http.Server{
|
s.server = &http.Server{
|
||||||
Addr: addr,
|
Addr: addr,
|
||||||
Handler: mux,
|
Handler: mux,
|
||||||
ReadTimeout: 5 * time.Second,
|
ReadTimeout: 30 * time.Second,
|
||||||
WriteTimeout: 5 * time.Second,
|
WriteTimeout: 30 * time.Second,
|
||||||
}
|
}
|
||||||
|
|
||||||
return s
|
return s
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue