[config] configurable SSE keep-alive period

This commit is contained in:
Aleksandr Soloshenko 2025-08-01 09:34:59 +07:00 committed by Aleksandr
parent 38d1681772
commit ae42e2a16d
4 changed files with 43 additions and 20 deletions

View File

@ -13,6 +13,7 @@ type Config struct {
Database Database `yaml:"database"` // database config Database Database `yaml:"database"` // database config
FCM FCMConfig `yaml:"fcm"` // firebase cloud messaging config FCM FCMConfig `yaml:"fcm"` // firebase cloud messaging config
Tasks Tasks `yaml:"tasks"` // tasks config Tasks Tasks `yaml:"tasks"` // tasks config
SSE SSE `yaml:"sse"` // server-sent events config
} }
type Gateway struct { type Gateway struct {
@ -53,6 +54,10 @@ type HashingTask struct {
IntervalSeconds uint16 `yaml:"interval_seconds" envconfig:"TASKS__HASHING__INTERVAL_SECONDS"` // hashing interval in seconds IntervalSeconds uint16 `yaml:"interval_seconds" envconfig:"TASKS__HASHING__INTERVAL_SECONDS"` // hashing interval in seconds
} }
type SSE struct {
KeepAlivePeriodSeconds uint16 `yaml:"keep_alive_period_seconds" envconfig:"SSE__KEEP_ALIVE_PERIOD_SECONDS"` // keep alive period in seconds, 0 for no keep alive
}
var defaultConfig = Config{ var defaultConfig = Config{
Gateway: Gateway{Mode: GatewayModePublic}, Gateway: Gateway{Mode: GatewayModePublic},
HTTP: HTTP{ HTTP: HTTP{
@ -75,4 +80,7 @@ var defaultConfig = Config{
IntervalSeconds: uint16(15 * 60), IntervalSeconds: uint16(15 * 60),
}, },
}, },
SSE: SSE{
KeepAlivePeriodSeconds: 15,
},
} }

View File

@ -93,6 +93,8 @@ var Module = fx.Module(
} }
}), }),
fx.Provide(func(cfg Config) sse.Config { fx.Provide(func(cfg Config) sse.Config {
return sse.NewConfig() return sse.NewConfig(
sse.WithKeepAlivePeriod(time.Duration(cfg.SSE.KeepAlivePeriodSeconds) * time.Second),
)
}), }),
) )

View File

@ -2,28 +2,38 @@ package sse
import "time" import "time"
const defaultKeepAlivePeriod = 15 * time.Second type configOption func(*Config)
type Config struct { type Config struct {
keepAlivePeriod time.Duration keepAlivePeriod time.Duration
} }
func NewConfig() Config { const defaultKeepAlivePeriod = 15 * time.Second
return Config{
keepAlivePeriod: defaultKeepAlivePeriod, var defaultConfig = Config{
keepAlivePeriod: defaultKeepAlivePeriod,
}
func NewConfig(opts ...configOption) Config {
c := defaultConfig
for _, opt := range opts {
opt(&c)
} }
return c
} }
func (c *Config) KeepAlivePeriod() time.Duration { func (c *Config) KeepAlivePeriod() time.Duration {
return c.keepAlivePeriod return c.keepAlivePeriod
} }
func (c *Config) SetKeepAlivePeriod(d time.Duration) *Config { func WithKeepAlivePeriod(d time.Duration) configOption {
if d <= 0 { if d < 0 {
d = defaultKeepAlivePeriod d = defaultKeepAlivePeriod
} }
c.keepAlivePeriod = d return func(c *Config) {
c.keepAlivePeriod = d
return c }
} }

View File

@ -101,17 +101,13 @@ func (s *Service) Handler(deviceID string, c *fiber.Ctx) error {
conn := s.registerConnection(deviceID) conn := s.registerConnection(deviceID)
defer s.removeConnection(deviceID, conn.id) defer s.removeConnection(deviceID, conn.id)
if err := s.writeToStream(w, ":keepalive"); err != nil { // Conditionally create ticker
s.logger.Warn("Failed to write keepalive", var ticker *time.Ticker
zap.String("device_id", deviceID), if s.config.keepAlivePeriod > 0 {
zap.String("connection_id", conn.id), ticker = time.NewTicker(s.config.keepAlivePeriod)
zap.Error(err)) defer ticker.Stop()
return
} }
ticker := time.NewTicker(s.config.keepAlivePeriod)
defer ticker.Stop()
for { for {
select { select {
case event := <-conn.channel: case event := <-conn.channel:
@ -122,7 +118,14 @@ func (s *Service) Handler(deviceID string, c *fiber.Ctx) error {
zap.Error(err)) zap.Error(err))
return return
} }
case <-ticker.C: // Conditionally handle ticker events
case <-func() <-chan time.Time {
if ticker != nil {
return ticker.C
}
// Return nil channel that never fires when disabled
return make(chan time.Time)
}():
if err := s.writeToStream(w, ":keepalive"); err != nil { if err := s.writeToStream(w, ":keepalive"); err != nil {
s.logger.Warn("Failed to write keepalive", s.logger.Warn("Failed to write keepalive",
zap.String("device_id", deviceID), zap.String("device_id", deviceID),