Merge pull request #49 from dj-oyu/fix/duplicate-http-listener
fix: eliminate duplicate HTTP listener on gateway port
This commit is contained in:
commit
8cae962d2a
2 changed files with 16 additions and 46 deletions
|
|
@ -11,7 +11,6 @@ import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"math"
|
"math"
|
||||||
"net/http"
|
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
|
@ -84,8 +83,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()
|
||||||
|
|
@ -342,41 +339,31 @@ 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 onto
|
||||||
// It registers health endpoints from the health server and discovers channels
|
// the health server's mux so everything is served by a single HTTP listener.
|
||||||
// that implement WebhookHandler and/or HealthChecker to register their handlers.
|
|
||||||
func (m *Manager) SetupHTTPServer(addr string, healthServer *health.Server) {
|
func (m *Manager) SetupHTTPServer(addr 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 {
|
||||||
|
|
@ -417,20 +404,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.FatalCF("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
|
||||||
}
|
}
|
||||||
|
|
@ -441,18 +414,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()
|
||||||
|
|
|
||||||
|
|
@ -356,6 +356,8 @@ func setupAndStartServices(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// HealthServer is the single HTTP listener. Channel webhooks and health
|
||||||
|
// checkers were registered on its mux via SetupHTTPServer.
|
||||||
go func() {
|
go func() {
|
||||||
var serverErr error
|
var serverErr error
|
||||||
if useTLS {
|
if useTLS {
|
||||||
|
|
@ -571,6 +573,13 @@ func restartServices(
|
||||||
if err = runningServices.ChannelManager.StartAll(context.Background()); err != nil {
|
if err = runningServices.ChannelManager.StartAll(context.Background()); err != nil {
|
||||||
return fmt.Errorf("error restarting channels: %w", err)
|
return fmt.Errorf("error restarting channels: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Start the HealthServer listener (single HTTP server for all routes)
|
||||||
|
go func() {
|
||||||
|
if sErr := runningServices.HealthServer.Start(); sErr != nil && sErr != http.ErrServerClosed {
|
||||||
|
logger.ErrorCF("health", "Health server error", map[string]any{"error": sErr.Error()})
|
||||||
|
}
|
||||||
|
}()
|
||||||
fmt.Printf(
|
fmt.Printf(
|
||||||
" ✓ Channels restarted, health endpoints at http://%s:%d/health and ready\n",
|
" ✓ Channels restarted, health endpoints at http://%s:%d/health and ready\n",
|
||||||
cfg.Gateway.Host,
|
cfg.Gateway.Host,
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue