diff --git a/cmd/showbridge/main.go b/cmd/showbridge/main.go index 859c50f..d90665e 100644 --- a/cmd/showbridge/main.go +++ b/cmd/showbridge/main.go @@ -90,12 +90,11 @@ func main() { } type showbridgeApp struct { - ctx context.Context - configPath string - logger *slog.Logger - router *showbridge.Router - routerRunner *sync.WaitGroup - routerMutex sync.Mutex + ctx context.Context + configPath string + logger *slog.Logger + router *showbridge.Router + routerMutex sync.Mutex } func readConfig(configPath string) (config.Config, error) { @@ -211,10 +210,9 @@ func run(ctx context.Context, c *cli.Command) error { } showbridgeApp := &showbridgeApp{ - ctx: ctx, - configPath: configPath, - logger: slog.Default().With("component", "cmd"), - routerRunner: &sync.WaitGroup{}, + ctx: ctx, + configPath: configPath, + logger: slog.Default().With("component", "cmd"), } config, err := readConfig(showbridgeApp.configPath) @@ -236,9 +234,7 @@ func run(ctx context.Context, c *cli.Command) error { showbridgeApp.routerMutex.Lock() showbridgeApp.router = router - showbridgeApp.routerRunner.Go(func() { - router.Start(context.Background()) - }) + router.Start(context.Background()) showbridgeApp.routerMutex.Unlock() go showbridgeApp.handleChannels() @@ -246,8 +242,6 @@ func run(ctx context.Context, c *cli.Command) error { <-showbridgeApp.ctx.Done() showbridgeApp.logger.Debug("shutting down router") showbridgeApp.router.Stop() - showbridgeApp.logger.Debug("waiting for router to exit") - showbridgeApp.routerRunner.Wait() return nil } diff --git a/internal/api/api.go b/internal/api/api.go index d78b4e8..e61f85f 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -20,7 +20,6 @@ type ApiServer struct { config config.ApiConfig serverMu sync.Mutex server *http.Server - shutdown context.CancelFunc logger *slog.Logger configurableRouter config.Configurable eventRouter common.EventRouter @@ -63,7 +62,6 @@ func (as *ApiServer) Start(config config.ApiConfig) { if err != nil && err != http.ErrServerClosed { as.logger.Error("server error", "error", err) } - as.shutdown() }() } @@ -71,16 +69,18 @@ func (as *ApiServer) Stop() { if as.server == nil { return } - as.logger.Debug("stopping") as.serverMu.Lock() defer as.serverMu.Unlock() if as.server != nil { apiShutdownCtx, apiShutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) - as.shutdown = apiShutdownCancel - as.server.Shutdown(apiShutdownCtx) - <-apiShutdownCtx.Done() + defer apiShutdownCancel() + err := as.server.Shutdown(apiShutdownCtx) + if err != nil { + as.logger.Error("error shutting down server", "error", err) + } as.server = nil } + as.logger.Debug("done") } func (as *ApiServer) handleHealthHTTP(w http.ResponseWriter, req *http.Request) { diff --git a/internal/module/db-sqlite.go b/internal/module/db-sqlite.go index fe0a83d..d0b4d46 100644 --- a/internal/module/db-sqlite.go +++ b/internal/module/db-sqlite.go @@ -5,6 +5,7 @@ import ( "database/sql" "fmt" "log/slog" + "sync" "github.com/google/jsonschema-go/jsonschema" "github.com/jwetzell/showbridge-go/internal/common" @@ -20,6 +21,8 @@ type DbSqlite struct { router common.RouteIO db *sql.DB logger *slog.Logger + dbMu sync.Mutex + cancel context.CancelFunc } func init() { @@ -61,25 +64,38 @@ func (t *DbSqlite) Type() string { func (t *DbSqlite) Start(ctx context.Context, router common.RouteIO) error { t.logger.Debug("running") t.router = router - t.ctx = ctx + moduleContext, cancel := context.WithCancel(ctx) + t.ctx = moduleContext + t.cancel = cancel db, err := sql.Open("sqlite", t.Dsn) if err != nil { return fmt.Errorf("db.sqlite error opening database: %w", err) } + t.dbMu.Lock() t.db = db - defer t.db.Close() - <-ctx.Done() + t.dbMu.Unlock() + <-t.ctx.Done() return nil } func (t *DbSqlite) Stop() { + if t.cancel != nil { + t.cancel() + } + t.dbMu.Lock() + defer t.dbMu.Unlock() if t.db != nil { t.db.Close() + t.db = nil } + t.logger.Debug("done") } +// TODO(jwetzell): get a database module layout that doesn't require handing the DB over func (t *DbSqlite) Database() (*sql.DB, error) { + t.dbMu.Lock() + defer t.dbMu.Unlock() if t.db == nil { return nil, fmt.Errorf("database not initialized") } diff --git a/internal/module/http-server.go b/internal/module/http-server.go index 942930e..5f23861 100644 --- a/internal/module/http-server.go +++ b/internal/module/http-server.go @@ -7,6 +7,7 @@ import ( "fmt" "log/slog" "net/http" + "sync" "time" "github.com/google/jsonschema-go/jsonschema" @@ -16,12 +17,14 @@ import ( ) type HTTPServer struct { - config config.ModuleConfig - Port uint16 - ctx context.Context - router common.RouteIO - logger *slog.Logger - cancel context.CancelFunc + config config.ModuleConfig + Port uint16 + ctx context.Context + router common.RouteIO + logger *slog.Logger + cancel context.CancelFunc + server *http.Server + serverMu sync.Mutex } type ResponseIOError struct { @@ -169,10 +172,9 @@ func (hs *HTTPServer) Start(ctx context.Context, router common.RouteIO) error { Handler: hs, } - go func() { - <-hs.ctx.Done() - httpServer.Close() - }() + hs.serverMu.Lock() + hs.server = httpServer + hs.serverMu.Unlock() err := httpServer.ListenAndServe() // TODO(jwetzell): handle server closed error differently @@ -183,7 +185,6 @@ func (hs *HTTPServer) Start(ctx context.Context, router common.RouteIO) error { } <-hs.ctx.Done() - hs.logger.Debug("done") return nil } @@ -210,5 +211,17 @@ func (hs *HTTPServer) Output(ctx context.Context, payload any) error { } func (hs *HTTPServer) Stop() { - hs.cancel() + if hs.cancel != nil { + hs.cancel() + } + hs.serverMu.Lock() + defer hs.serverMu.Unlock() + if hs.server != nil { + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + hs.server.Shutdown(shutdownCtx) + shutdownCancel() + <-shutdownCtx.Done() + hs.server = nil + } + hs.logger.Debug("done") } diff --git a/internal/module/midi-input.go b/internal/module/midi-input.go index d028eb3..86f7e99 100644 --- a/internal/module/midi-input.go +++ b/internal/module/midi-input.go @@ -15,13 +15,13 @@ import ( ) type MIDIInput struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - Port string - SendFunc func(midi.Message) error - logger *slog.Logger - cancel context.CancelFunc + config config.ModuleConfig + ctx context.Context + router common.RouteIO + Port string + logger *slog.Logger + cancel context.CancelFunc + stop func() } func init() { @@ -61,7 +61,6 @@ func (mi *MIDIInput) Type() string { func (mi *MIDIInput) Start(ctx context.Context, router common.RouteIO) error { mi.logger.Debug("running") - defer midi.CloseDriver() mi.router = router moduleContext, cancel := context.WithCancel(ctx) mi.ctx = moduleContext @@ -81,14 +80,20 @@ func (mi *MIDIInput) Start(ctx context.Context, router common.RouteIO) error { if err != nil { return err } - - defer stop() + mi.stop = stop <-mi.ctx.Done() - mi.logger.Debug("done") return nil } func (mi *MIDIInput) Stop() { - mi.cancel() + if mi.cancel != nil { + mi.cancel() + } + if mi.stop != nil { + mi.stop() + mi.stop = nil + } + midi.CloseDriver() + mi.logger.Debug("done") } diff --git a/internal/module/midi-output.go b/internal/module/midi-output.go index b8a323f..5eb7d97 100644 --- a/internal/module/midi-output.go +++ b/internal/module/midi-output.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "log/slog" + "sync" "github.com/google/jsonschema-go/jsonschema" "github.com/jwetzell/showbridge-go/internal/common" @@ -16,13 +17,14 @@ import ( ) type MIDIOutput struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - Port string - SendFunc func(midi.Message) error - logger *slog.Logger - cancel context.CancelFunc + config config.ModuleConfig + ctx context.Context + router common.RouteIO + Port string + sendFunc func(midi.Message) error + logger *slog.Logger + cancel context.CancelFunc + sendFuncMu sync.Mutex } func init() { @@ -63,7 +65,6 @@ func (mo *MIDIOutput) Type() string { func (mo *MIDIOutput) Start(ctx context.Context, router common.RouteIO) error { mo.logger.Debug("running") - defer midi.CloseDriver() mo.router = router moduleContext, cancel := context.WithCancel(ctx) mo.ctx = moduleContext @@ -80,15 +81,18 @@ func (mo *MIDIOutput) Start(ctx context.Context, router common.RouteIO) error { return err } - mo.SendFunc = send + mo.sendFuncMu.Lock() + mo.sendFunc = send + mo.sendFuncMu.Unlock() <-mo.ctx.Done() - mo.logger.Debug("done") return nil } func (mo *MIDIOutput) Output(ctx context.Context, payload any) error { - if mo.SendFunc == nil { + mo.sendFuncMu.Lock() + defer mo.sendFuncMu.Unlock() + if mo.sendFunc == nil { return errors.New("midi.output output is not setup") } @@ -98,9 +102,13 @@ func (mo *MIDIOutput) Output(ctx context.Context, payload any) error { return errors.New("midi.output can only output midi.Message") } - return mo.SendFunc(payloadMessage) + return mo.sendFunc(payloadMessage) } func (mo *MIDIOutput) Stop() { - mo.cancel() + if mo.cancel != nil { + mo.cancel() + } + midi.CloseDriver() + mo.logger.Debug("done") } diff --git a/internal/module/mqtt-client.go b/internal/module/mqtt-client.go index 4c1357c..ad5bcb4 100644 --- a/internal/module/mqtt-client.go +++ b/internal/module/mqtt-client.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log/slog" + "sync" mqtt "github.com/eclipse/paho.mqtt.golang" "github.com/google/jsonschema-go/jsonschema" @@ -22,6 +23,7 @@ type MQTTClient struct { client mqtt.Client logger *slog.Logger cancel context.CancelFunc + clientMu sync.Mutex } func init() { @@ -100,8 +102,8 @@ func (mc *MQTTClient) Start(ctx context.Context, router common.RouteIO) error { token.Wait() } + mc.clientMu.Lock() mc.client = mqtt.NewClient(opts) - defer mc.client.Disconnect(250) token := mc.client.Connect() @@ -110,9 +112,9 @@ func (mc *MQTTClient) Start(ctx context.Context, router common.RouteIO) error { if err != nil { return err } + mc.clientMu.Unlock() <-mc.ctx.Done() - mc.logger.Debug("done") return nil } @@ -139,5 +141,14 @@ func (mc *MQTTClient) Output(ctx context.Context, payload any) error { } func (mc *MQTTClient) Stop() { - mc.cancel() + if mc.cancel != nil { + mc.cancel() + } + mc.clientMu.Lock() + defer mc.clientMu.Unlock() + if mc.client != nil { + mc.client.Disconnect(250) + mc.client = nil + } + mc.logger.Debug("done") } diff --git a/internal/module/nats-client.go b/internal/module/nats-client.go index e133b09..5724115 100644 --- a/internal/module/nats-client.go +++ b/internal/module/nats-client.go @@ -4,6 +4,7 @@ import ( "context" "errors" "log/slog" + "sync" "github.com/google/jsonschema-go/jsonschema" "github.com/jwetzell/showbridge-go/internal/common" @@ -13,14 +14,17 @@ import ( ) type NATSClient struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - URL string - Subject string - client *nats.Conn - logger *slog.Logger - cancel context.CancelFunc + config config.ModuleConfig + ctx context.Context + router common.RouteIO + URL string + Subject string + client *nats.Conn + logger *slog.Logger + cancel context.CancelFunc + sub *nats.Subscription + subMu sync.Mutex + clientMu sync.Mutex } func init() { @@ -81,10 +85,9 @@ func (nc *NATSClient) Start(ctx context.Context, router common.RouteIO) error { return err } + nc.clientMu.Lock() nc.client = client - - defer client.Drain() - defer client.Close() + nc.clientMu.Unlock() sub, err := nc.client.Subscribe(nc.Subject, func(msg *nats.Msg) { if nc.router != nil { @@ -95,11 +98,11 @@ func (nc *NATSClient) Start(ctx context.Context, router common.RouteIO) error { if err != nil { return err } - - defer sub.Unsubscribe() + nc.subMu.Lock() + nc.sub = sub + nc.subMu.Unlock() <-nc.ctx.Done() - nc.logger.Debug("done") return nil } @@ -111,6 +114,9 @@ func (nc *NATSClient) Output(ctx context.Context, payload any) error { return errors.New("nats.client is only able to output NATSMessage") } + nc.clientMu.Lock() + defer nc.clientMu.Unlock() + if nc.client == nil { return errors.New("nats.client client is not setup") } @@ -125,5 +131,23 @@ func (nc *NATSClient) Output(ctx context.Context, payload any) error { } func (nc *NATSClient) Stop() { - nc.cancel() + if nc.cancel != nil { + nc.cancel() + } + nc.subMu.Lock() + defer nc.subMu.Unlock() + if nc.sub != nil { + nc.sub.Unsubscribe() + nc.sub = nil + } + + nc.clientMu.Lock() + defer nc.clientMu.Unlock() + if nc.client != nil { + nc.client.Drain() + // TODO(jwetzell): setup closed callback to get when client is fully closed + nc.client.Close() + nc.client = nil + } + nc.logger.Debug("done") } diff --git a/internal/module/nats-server.go b/internal/module/nats-server.go index dd420bd..c7657a8 100644 --- a/internal/module/nats-server.go +++ b/internal/module/nats-server.go @@ -7,6 +7,7 @@ import ( "fmt" "log/slog" "net" + "sync" "time" "github.com/google/jsonschema-go/jsonschema" @@ -16,14 +17,15 @@ import ( ) type NATSServer struct { - config config.ModuleConfig - ctx context.Context - Ip string - Port int - router common.RouteIO - server *server.Server - logger *slog.Logger - cancel context.CancelFunc + config config.ModuleConfig + ctx context.Context + Ip string + Port int + router common.RouteIO + server *server.Server + logger *slog.Logger + cancel context.CancelFunc + serverMu sync.Mutex } func init() { @@ -103,9 +105,10 @@ func (ns *NATSServer) Start(ctx context.Context, router common.RouteIO) error { return err } + ns.serverMu.Lock() ns.server = natsServer + defer ns.serverMu.Unlock() natsServer.Start() - defer natsServer.Shutdown() if !natsServer.ReadyForConnections(5 * time.Second) { return errors.New("nats.server failed to start") @@ -114,13 +117,17 @@ func (ns *NATSServer) Start(ctx context.Context, router common.RouteIO) error { <-ns.ctx.Done() - ns.logger.Debug("done") return nil } func (ns *NATSServer) Stop() { - ns.cancel() + if ns.cancel != nil { + ns.cancel() + } + ns.serverMu.Lock() + defer ns.serverMu.Unlock() if ns.server != nil { ns.server.Shutdown() } + ns.logger.Debug("done") } diff --git a/internal/module/psn-client.go b/internal/module/psn-client.go index fbdbc7b..3f87d1f 100644 --- a/internal/module/psn-client.go +++ b/internal/module/psn-client.go @@ -4,6 +4,7 @@ import ( "context" "log/slog" "net" + "sync" "time" "github.com/jwetzell/psn-go" @@ -19,6 +20,7 @@ type PSNClient struct { decoder *psn.Decoder logger *slog.Logger cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -55,21 +57,22 @@ func (pc *PSNClient) Start(ctx context.Context, router common.RouteIO) error { if err != nil { return err } - defer client.Close() + pc.connMu.Lock() pc.conn = client + pc.connMu.Unlock() buffer := make([]byte, 2048) for { select { case <-pc.ctx.Done(): - // TODO(jwetzell): cleanup? - pc.logger.Debug("done") return nil default: + pc.connMu.Lock() pc.conn.SetDeadline(time.Now().Add(time.Millisecond * 200)) numBytes, _, err := pc.conn.ReadFromUDP(buffer) + pc.connMu.Unlock() if err != nil { //NOTE(jwetzell) we hit deadline if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() { @@ -99,5 +102,14 @@ func (pc *PSNClient) Start(ctx context.Context, router common.RouteIO) error { } func (pc *PSNClient) Stop() { - pc.cancel() + if pc.cancel != nil { + pc.cancel() + } + pc.connMu.Lock() + defer pc.connMu.Unlock() + if pc.conn != nil { + pc.conn.Close() + pc.conn = nil + } + pc.logger.Debug("done") } diff --git a/internal/module/redis-client.go b/internal/module/redis-client.go index 887a120..9212fa7 100644 --- a/internal/module/redis-client.go +++ b/internal/module/redis-client.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log/slog" + "sync" "github.com/google/jsonschema-go/jsonschema" "github.com/jwetzell/showbridge-go/internal/common" @@ -13,14 +14,15 @@ import ( ) type RedisClient struct { - config config.ModuleConfig - ctx context.Context - router common.RouteIO - Host string - Port uint16 - client *redis.Client - logger *slog.Logger - cancel context.CancelFunc + config config.ModuleConfig + ctx context.Context + router common.RouteIO + Host string + Port uint16 + client *redis.Client + logger *slog.Logger + cancel context.CancelFunc + clientMu sync.Mutex } func init() { @@ -87,17 +89,25 @@ func (rc *RedisClient) Start(ctx context.Context, router common.RouteIO) error { DB: 0, }) + rc.clientMu.Lock() rc.client = client - - defer client.Close() + rc.clientMu.Unlock() <-rc.ctx.Done() - rc.logger.Debug("done") return nil } func (rc *RedisClient) Stop() { - rc.cancel() + if rc.cancel != nil { + rc.cancel() + } + rc.clientMu.Lock() + defer rc.clientMu.Unlock() + if rc.client != nil { + rc.client.Close() + rc.client = nil + } + rc.logger.Debug("done") } func (rc *RedisClient) Get(key string) (any, error) { diff --git a/internal/module/serial-client.go b/internal/module/serial-client.go index ba53c3e..b788573 100644 --- a/internal/module/serial-client.go +++ b/internal/module/serial-client.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "log/slog" + "sync" "time" "github.com/google/jsonschema-go/jsonschema" @@ -26,6 +27,7 @@ type SerialClient struct { port serial.Port logger *slog.Logger cancel context.CancelFunc + portMu sync.Mutex } func init() { @@ -93,7 +95,8 @@ func (sc *SerialClient) Type() string { } func (sc *SerialClient) SetupPort() error { - + sc.portMu.Lock() + defer sc.portMu.Unlock() port, err := serial.Open(sc.Port, sc.Mode) if err != nil { return err @@ -111,23 +114,10 @@ func (sc *SerialClient) Start(ctx context.Context, router common.RouteIO) error sc.ctx = moduleContext sc.cancel = cancel - // TODO(jwetzell): shutdown with router.Context properly - go func() { - <-sc.ctx.Done() - sc.logger.Debug("done") - if sc.port != nil { - sc.port.Close() - } - }() - - for { + for sc.ctx.Err() == nil { err := sc.SetupPort() if err != nil { if sc.ctx.Err() != nil { - sc.logger.Debug("done") - if sc.port != nil { - sc.port.Close() - } return nil } sc.logger.Error("port setup error", "port", sc.Port, "error", err.Error()) @@ -138,17 +128,12 @@ func (sc *SerialClient) Start(ctx context.Context, router common.RouteIO) error buffer := make([]byte, 1024) select { case <-sc.ctx.Done(): - sc.logger.Debug("done") - if sc.port != nil { - sc.port.Close() - } return nil default: READ: - for { + for sc.ctx.Err() == nil { select { case <-sc.ctx.Done(): - sc.logger.Debug("done") return nil default: byteCount, err := sc.port.Read(buffer) @@ -174,6 +159,7 @@ func (sc *SerialClient) Start(ctx context.Context, router common.RouteIO) error } } } + return nil } func (sc *SerialClient) Output(ctx context.Context, payload any) error { @@ -189,5 +175,14 @@ func (sc *SerialClient) Output(ctx context.Context, payload any) error { } func (sc *SerialClient) Stop() { - sc.cancel() + if sc.cancel != nil { + sc.cancel() + } + sc.portMu.Lock() + defer sc.portMu.Unlock() + if sc.port != nil { + sc.port.Close() + sc.port = nil + } + sc.logger.Debug("done") } diff --git a/internal/module/sip-call-server.go b/internal/module/sip-call-server.go index 3442809..f407dc7 100644 --- a/internal/module/sip-call-server.go +++ b/internal/module/sip-call-server.go @@ -29,9 +29,10 @@ type SIPCallServer struct { Port int Transport string UserAgent string - dg *diago.Diago logger *slog.Logger cancel context.CancelFunc + ua *sipgo.UserAgent + uaMu sync.Mutex } type SIPCallMessage struct { @@ -145,7 +146,9 @@ func (scs *SIPCallServer) Start(ctx context.Context, router common.RouteIO) erro sipgo.WithUserAgentTransportLayerOptions(sip.WithTransportLayerLogger(diagoLogger)), sipgo.WithUserAgentTransactionLayerOptions(sip.WithTransactionLayerLogger(diagoLogger)), ) - defer ua.Close() + scs.uaMu.Lock() + scs.ua = ua + scs.uaMu.Unlock() sip.SetDefaultLogger(diagoLogger) media.SetDefaultLogger(diagoLogger) @@ -157,16 +160,14 @@ func (scs *SIPCallServer) Start(ctx context.Context, router common.RouteIO) erro }, )) - go func() { - dg.Serve(scs.ctx, func(inDialog *diago.DialogServerSession) { - scs.HandleCall(inDialog) - }) - }() - - scs.dg = dg + err := dg.Serve(scs.ctx, func(inDialog *diago.DialogServerSession) { + scs.HandleCall(inDialog) + }) + if err != nil { + scs.logger.Error("diago serve error", "error", err) + } <-scs.ctx.Done() - scs.logger.Debug("done") return nil } @@ -247,5 +248,13 @@ func (scs *SIPCallServer) Output(ctx context.Context, payload any) error { } func (scs *SIPCallServer) Stop() { - scs.cancel() + if scs.cancel != nil { + scs.cancel() + } + scs.uaMu.Lock() + defer scs.uaMu.Unlock() + if scs.ua != nil { + scs.ua.Close() + } + scs.logger.Debug("done") } diff --git a/internal/module/sip-dtmf-server.go b/internal/module/sip-dtmf-server.go index 3c30ab2..9ef4f95 100644 --- a/internal/module/sip-dtmf-server.go +++ b/internal/module/sip-dtmf-server.go @@ -33,6 +33,8 @@ type SIPDTMFServer struct { Separator string logger *slog.Logger cancel context.CancelFunc + ua *sipgo.UserAgent + uaMu sync.Mutex } type SIPDTMFMessage struct { @@ -164,7 +166,10 @@ func (sds *SIPDTMFServer) Start(ctx context.Context, router common.RouteIO) erro sipgo.WithUserAgentTransportLayerOptions(sip.WithTransportLayerLogger(diagoLogger)), sipgo.WithUserAgentTransactionLayerOptions(sip.WithTransactionLayerLogger(diagoLogger)), ) - defer ua.Close() + + sds.uaMu.Lock() + sds.ua = ua + sds.uaMu.Unlock() sip.SetDefaultLogger(diagoLogger) media.SetDefaultLogger(diagoLogger) @@ -185,7 +190,6 @@ func (sds *SIPDTMFServer) Start(ctx context.Context, router common.RouteIO) erro } <-sds.ctx.Done() - sds.logger.Debug("done") return nil } @@ -281,5 +285,13 @@ func (sds *SIPDTMFServer) Output(ctx context.Context, payload any) error { } func (sds *SIPDTMFServer) Stop() { - sds.cancel() + if sds.cancel != nil { + sds.cancel() + } + sds.uaMu.Lock() + defer sds.uaMu.Unlock() + if sds.ua != nil { + sds.ua.Close() + } + sds.logger.Debug("done") } diff --git a/internal/module/tcp-client.go b/internal/module/tcp-client.go index 1e82ce2..d1d71a9 100644 --- a/internal/module/tcp-client.go +++ b/internal/module/tcp-client.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net" + "sync" "time" "github.com/google/jsonschema-go/jsonschema" @@ -23,6 +24,7 @@ type TCPClient struct { Addr *net.TCPAddr logger *slog.Logger cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -98,20 +100,10 @@ func (tc *TCPClient) Start(ctx context.Context, router common.RouteIO) error { tc.ctx = moduleContext tc.cancel = cancel - // TODO(jwetzell): shutdown with router.Context properly - go func() { - <-tc.ctx.Done() - tc.logger.Debug("done") - if tc.conn != nil { - tc.conn.Close() - } - }() - - for { + for tc.ctx.Err() == nil { err := tc.SetupConn() if err != nil { if tc.ctx.Err() != nil { - tc.logger.Debug("done") return nil } tc.logger.Error("connection error", "error", err.Error()) @@ -122,14 +114,12 @@ func (tc *TCPClient) Start(ctx context.Context, router common.RouteIO) error { buffer := make([]byte, 1024) select { case <-tc.ctx.Done(): - tc.logger.Debug("done") return nil default: READ: for { select { case <-tc.ctx.Done(): - tc.logger.Debug("done") return nil default: byteCount, err := tc.conn.Read(buffer) @@ -155,21 +145,22 @@ func (tc *TCPClient) Start(ctx context.Context, router common.RouteIO) error { } } } + return nil } func (tc *TCPClient) SetupConn() error { + tc.connMu.Lock() + defer tc.connMu.Unlock() client, err := net.DialTCP("tcp", nil, tc.Addr) tc.conn = client return err } func (tc *TCPClient) Output(ctx context.Context, payload any) error { - // NOTE(jwetzell): not sure how this would occur but + tc.connMu.Lock() + defer tc.connMu.Unlock() if tc.conn == nil { - err := tc.SetupConn() - if err != nil { - return err - } + return errors.New("net.tcp.client client is not setup") } payloadBytes, ok := common.GetAnyAsByteSlice(payload) if !ok { @@ -180,5 +171,15 @@ func (tc *TCPClient) Output(ctx context.Context, payload any) error { } func (tc *TCPClient) Stop() { - tc.cancel() + if tc.cancel != nil { + tc.cancel() + } + tc.connMu.Lock() + defer tc.connMu.Unlock() + if tc.conn != nil { + tc.conn.Close() + tc.conn = nil + } + tc.logger.Debug("done") + } diff --git a/internal/module/tcp-server.go b/internal/module/tcp-server.go index dce1c62..5efc383 100644 --- a/internal/module/tcp-server.go +++ b/internal/module/tcp-server.go @@ -25,12 +25,13 @@ type TCPServer struct { Framer framer.Framer ctx context.Context router common.RouteIO - quit chan any wg sync.WaitGroup connections []*net.TCPConn connectionsMu sync.RWMutex logger *slog.Logger cancel context.CancelFunc + listener *net.TCPListener + listenerMu sync.Mutex } func init() { @@ -91,7 +92,7 @@ func init() { if err != nil { return nil, err } - return &TCPServer{Framer: framer, Addr: addr, config: moduleConfig, quit: make(chan any), logger: CreateLogger(moduleConfig)}, nil + return &TCPServer{Framer: framer, Addr: addr, config: moduleConfig, logger: CreateLogger(moduleConfig)}, nil }, }) } @@ -108,14 +109,14 @@ func (ts *TCPServer) handleClient(client *net.TCPConn) { ts.connectionsMu.Lock() ts.connections = append(ts.connections, client) ts.connectionsMu.Unlock() - ts.logger.Debug("net.tcp.server connection accepted", "remoteAddr", client.RemoteAddr().String()) + ts.logger.Debug("connection accepted", "remoteAddr", client.RemoteAddr().String()) defer client.Close() buffer := make([]byte, 1024) ClientRead: - for { + for ts.ctx.Err() == nil { select { - case <-ts.quit: + case <-ts.ctx.Done(): client.Close() ts.connectionsMu.Lock() for i := 0; i < len(ts.connections); i++ { @@ -157,7 +158,6 @@ ClientRead: break } } - ts.logger.Debug("stream ended", "remoteAddr", client.RemoteAddr().String()) ts.connectionsMu.Unlock() } return @@ -194,21 +194,17 @@ func (ts *TCPServer) Start(ctx context.Context, router common.RouteIO) error { if err != nil { return err } + ts.listenerMu.Lock() + ts.listener = listener + ts.listenerMu.Unlock() ts.wg.Add(1) - go func() { - <-ts.ctx.Done() - close(ts.quit) - listener.Close() - ts.logger.Debug("done") - }() - AcceptLoop: - for { + for ts.ctx.Err() == nil { conn, err := listener.AcceptTCP() if err != nil { select { - case <-ts.quit: + case <-ts.ctx.Done(): break AcceptLoop default: ts.logger.Debug("problem with listener", "error", err) @@ -220,7 +216,6 @@ AcceptLoop: } } ts.wg.Done() - ts.wg.Wait() return nil } @@ -248,6 +243,15 @@ func (ts *TCPServer) Output(ctx context.Context, payload any) error { } func (ts *TCPServer) Stop() { - ts.cancel() + if ts.cancel != nil { + ts.cancel() + } + ts.listenerMu.Lock() + defer ts.listenerMu.Unlock() + if ts.listener != nil { + ts.listener.Close() + ts.listener = nil + } ts.wg.Wait() + ts.logger.Debug("done") } diff --git a/internal/module/time-interval.go b/internal/module/time-interval.go index 149ec89..c9e7b8f 100644 --- a/internal/module/time-interval.go +++ b/internal/module/time-interval.go @@ -66,12 +66,10 @@ func (i *TimeInterval) Start(ctx context.Context, router common.RouteIO) error { ticker := time.NewTicker(time.Millisecond * time.Duration(i.Duration)) i.ticker = ticker - defer ticker.Stop() for { select { case <-i.ctx.Done(): - i.logger.Debug("done") return nil case <-ticker.C: if i.router != nil { @@ -79,9 +77,15 @@ func (i *TimeInterval) Start(ctx context.Context, router common.RouteIO) error { } } } - } func (i *TimeInterval) Stop() { - i.cancel() + if i.cancel != nil { + i.cancel() + } + if i.ticker != nil { + i.ticker.Stop() + i.ticker = nil + } + i.logger.Debug("done") } diff --git a/internal/module/time-timer.go b/internal/module/time-timer.go index 77bd641..1be1fa1 100644 --- a/internal/module/time-timer.go +++ b/internal/module/time-timer.go @@ -66,12 +66,9 @@ func (t *TimeTimer) Start(ctx context.Context, router common.RouteIO) error { t.cancel = cancel t.timer = time.NewTimer(time.Millisecond * time.Duration(t.Duration)) - defer t.timer.Stop() for { select { case <-t.ctx.Done(): - t.timer.Stop() - t.logger.Debug("done") return nil case time := <-t.timer.C: if t.router != nil { @@ -82,5 +79,12 @@ func (t *TimeTimer) Start(ctx context.Context, router common.RouteIO) error { } func (t *TimeTimer) Stop() { - t.cancel() + if t.cancel != nil { + t.cancel() + } + if t.timer != nil { + t.timer.Stop() + t.timer = nil + } + t.logger.Debug("done") } diff --git a/internal/module/udp-client.go b/internal/module/udp-client.go index 9a4916b..273dab9 100644 --- a/internal/module/udp-client.go +++ b/internal/module/udp-client.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net" + "sync" "github.com/google/jsonschema-go/jsonschema" "github.com/jwetzell/showbridge-go/internal/common" @@ -21,6 +22,7 @@ type UDPClient struct { router common.RouteIO logger *slog.Logger cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -74,6 +76,8 @@ func (uc *UDPClient) Type() string { } func (uc *UDPClient) SetupConn() error { + uc.connMu.Lock() + defer uc.connMu.Unlock() client, err := net.DialUDP("udp", nil, uc.Addr) uc.conn = client return err @@ -92,15 +96,12 @@ func (uc *UDPClient) Start(ctx context.Context, router common.RouteIO) error { } <-uc.ctx.Done() - uc.logger.Debug("done") - if uc.conn != nil { - uc.conn.Close() - } return nil } func (uc *UDPClient) Output(ctx context.Context, payload any) error { - + uc.connMu.Lock() + defer uc.connMu.Unlock() payloadBytes, ok := common.GetAnyAsByteSlice(payload) if !ok { return errors.New("net.udp.client is only able to output bytes") @@ -118,5 +119,15 @@ func (uc *UDPClient) Output(ctx context.Context, payload any) error { } func (uc *UDPClient) Stop() { - uc.cancel() + if uc.cancel != nil { + uc.cancel() + } + uc.connMu.Lock() + defer uc.connMu.Unlock() + if uc.conn != nil { + uc.conn.Close() + uc.conn = nil + } + + uc.logger.Debug("done") } diff --git a/internal/module/udp-multicast.go b/internal/module/udp-multicast.go index 9677656..b85554d 100644 --- a/internal/module/udp-multicast.go +++ b/internal/module/udp-multicast.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net" + "sync" "time" "github.com/google/jsonschema-go/jsonschema" @@ -21,6 +22,7 @@ type UDPMulticast struct { Addr *net.UDPAddr logger *slog.Logger cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -86,19 +88,21 @@ func (um *UDPMulticast) Start(ctx context.Context, router common.RouteIO) error } defer client.Close() + um.connMu.Lock() um.conn = client + um.connMu.Unlock() buffer := make([]byte, 2048) for { select { case <-um.ctx.Done(): - // TODO(jwetzell): cleanup? - um.logger.Debug("done") return nil default: + um.connMu.Lock() um.conn.SetDeadline(time.Now().Add(time.Millisecond * 200)) numBytes, _, err := um.conn.ReadFromUDP(buffer) + um.connMu.Unlock() if err != nil { //NOTE(jwetzell) we hit deadline if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() { @@ -136,5 +140,14 @@ func (um *UDPMulticast) Output(ctx context.Context, payload any) error { } func (um *UDPMulticast) Stop() { - um.cancel() + if um.cancel != nil { + um.cancel() + } + um.connMu.Lock() + defer um.connMu.Unlock() + if um.conn != nil { + um.conn.Close() + um.conn = nil + } + um.logger.Debug("done") } diff --git a/internal/module/udp-server.go b/internal/module/udp-server.go index 9acde51..2623473 100644 --- a/internal/module/udp-server.go +++ b/internal/module/udp-server.go @@ -7,6 +7,7 @@ import ( "fmt" "log/slog" "net" + "sync" "time" "github.com/google/jsonschema-go/jsonschema" @@ -22,6 +23,8 @@ type UDPServer struct { router common.RouteIO logger *slog.Logger cancel context.CancelFunc + listener *net.UDPConn + listenerMu sync.Mutex } func init() { @@ -106,15 +109,13 @@ func (us *UDPServer) Start(ctx context.Context, router common.RouteIO) error { if err != nil { return err } - - defer listener.Close() + us.listenerMu.Lock() + us.listener = listener buffer := make([]byte, us.BufferSize) - for { + for us.ctx.Err() == nil { select { case <-us.ctx.Done(): - // TODO(jwetzell): cleanup? - us.logger.Debug("done") return nil default: listener.SetDeadline(time.Now().Add(time.Millisecond * 200)) @@ -135,7 +136,8 @@ func (us *UDPServer) Start(ctx context.Context, router common.RouteIO) error { } } } - + us.listenerMu.Unlock() + return nil } func (us *UDPServer) Output(ctx context.Context, payload any) error { @@ -143,5 +145,14 @@ func (us *UDPServer) Output(ctx context.Context, payload any) error { } func (us *UDPServer) Stop() { - us.cancel() + if us.cancel != nil { + us.cancel() + } + us.listenerMu.Lock() + defer us.listenerMu.Unlock() + if us.listener != nil { + us.listener.Close() + us.listener = nil + } + us.logger.Debug("done") } diff --git a/internal/module/websocket-client.go b/internal/module/websocket-client.go index 356b5f7..2f0488d 100644 --- a/internal/module/websocket-client.go +++ b/internal/module/websocket-client.go @@ -5,7 +5,9 @@ import ( "errors" "fmt" "log/slog" + "net" "net/url" + "sync" "time" "github.com/google/jsonschema-go/jsonschema" @@ -22,6 +24,7 @@ type WebSocketClient struct { router common.RouteIO logger *slog.Logger cancel context.CancelFunc + connMu sync.Mutex } func init() { @@ -69,6 +72,8 @@ func (wc *WebSocketClient) Type() string { } func (wc *WebSocketClient) SetupConn() error { + wc.connMu.Lock() + defer wc.connMu.Unlock() conn, _, err := websocket.DefaultDialer.Dial(wc.URL.String(), nil) wc.conn = conn return err @@ -87,17 +92,13 @@ func (wc *WebSocketClient) Start(ctx context.Context, router common.RouteIO) err wc.logger.Error("connection error", "error", err) } else { // NOTE(jwetzell): enter read loop until an error occurs + wc.logger.Debug("websocket connection established entering read loop") wc.readLoop() } // NOTE(jwetzell): if connection is lost or read error wait before trying again time.Sleep(2 * time.Second) } - <-wc.ctx.Done() - wc.logger.Debug("done") - if wc.conn != nil { - wc.conn.Close() - } return nil } @@ -107,10 +108,20 @@ func (wc *WebSocketClient) readLoop() { wc.logger.Error("websocket connection is not established") return } - + wc.conn.SetReadDeadline(time.Now().Add(5 * time.Second)) messageType, message, err := wc.conn.ReadMessage() if err != nil { - wc.logger.Error("read error", "error", err) + if opErr, ok := err.(*net.OpError); ok { + // NOTE(jwetzell) we hit deadline + if opErr.Timeout() { + continue + } + // NOTE(jwetzell) connection was closed + if errors.Is(opErr, net.ErrClosed) { + continue + } + } + wc.logger.Error("websocket read error", "error", err) return } if wc.router != nil { @@ -154,7 +165,8 @@ func (wc *WebSocketClient) outputString(ctx context.Context, payload string) err } func (wc *WebSocketClient) Output(ctx context.Context, payload any) error { - + wc.connMu.Lock() + defer wc.connMu.Unlock() payloadBytes, ok := common.GetAnyAsByteSlice(payload) if ok { return wc.outputBytes(ctx, payloadBytes) @@ -169,5 +181,14 @@ func (wc *WebSocketClient) Output(ctx context.Context, payload any) error { } func (wc *WebSocketClient) Stop() { - wc.cancel() + if wc.cancel != nil { + wc.cancel() + } + wc.connMu.Lock() + defer wc.connMu.Unlock() + if wc.conn != nil { + wc.conn.Close() + wc.conn = nil + } + wc.logger.Debug("done") } diff --git a/router.go b/router.go index af6cf3c..20b98b2 100644 --- a/router.go +++ b/router.go @@ -5,6 +5,7 @@ import ( "errors" "log/slog" "sync" + "sync/atomic" "github.com/jwetzell/showbridge-go/internal/api" "github.com/jwetzell/showbridge-go/internal/common" @@ -170,17 +171,19 @@ func (r *Router) Start(ctx context.Context) { r.contextCancel = cancel r.startModules() r.apiServer.Start(r.GetRunningConfig().Api) - <-r.Context.Done() - r.logger.Debug("shutting down api server") - r.apiServer.Stop() - r.logger.Debug("waiting for modules to exit") - r.moduleWait.Wait() - r.logger.Info("done") } func (r *Router) Stop() { r.logger.Info("stopping") + r.logger.Debug("shutting down api server") + r.apiServer.Stop() + r.logger.Debug("stopping modules") + r.stopModules() + r.logger.Debug("waiting for modules to exit") + r.moduleWait.Wait() + r.logger.Debug("canceling router context") r.contextCancel() + r.logger.Info("done") } func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []common.RouteIOError) { @@ -190,7 +193,7 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) spanCtx, span := otel.Tracer("router").Start(ctx, "input", trace.WithAttributes(attribute.String("source.id", sourceId))) defer span.End() var routeIOErrors []common.RouteIOError - routeFound := false + var routeFound atomic.Bool r.broadcastEvent(common.Event{ Type: "input", @@ -209,7 +212,7 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) if routeInstance.Input() == sourceId { routeWaitGroup.Go(func() { - routeFound = true + routeFound.Store(true) routeCtx, routeSpan := otel.Tracer("router").Start(spanCtx, "route", trace.WithAttributes(attribute.Int("route.index", routeIndex), attribute.String("route.input", routeInstance.Input()))) _, err := routeInstance.ProcessPayload(routeCtx, common.WrappedPayload{ @@ -248,7 +251,7 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) } } routeWaitGroup.Wait() - return routeFound, routeIOErrors + return routeFound.Load(), routeIOErrors } func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload any) error { @@ -301,7 +304,6 @@ func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload } func (r *Router) startModules() { - for moduleId := range r.ModuleInstances { // TODO(jwetzell): handle module run errors err := r.startModule(r.Context, moduleId) @@ -310,3 +312,13 @@ func (r *Router) startModules() { } } } + +func (r *Router) stopModules() { + for moduleId := range r.ModuleInstances { + // TODO(jwetzell): handle module stop errors? + err := r.stopModule(moduleId) + if err != nil { + r.logger.Error("error stopping module", "moduleId", moduleId, "error", err) + } + } +} diff --git a/router_test.go b/router_test.go index a0aa937..a1cca98 100644 --- a/router_test.go +++ b/router_test.go @@ -5,7 +5,6 @@ import ( "fmt" "log/slog" "reflect" - "sync" "testing" "time" @@ -54,7 +53,9 @@ func (mcm *MockCounterModule) Type() string { } func (mcm *MockCounterModule) Stop() { - mcm.cancel() + if mcm.cancel != nil { + mcm.cancel() + } } func init() { @@ -222,12 +223,7 @@ func TestRouterInputUnknownDestinationModule(t *testing.T) { t.Fatalf("router should not have returned any route errors: %v", routeErrors) } - routerRunner := sync.WaitGroup{} - - routerRunner.Go(func() { - router.Start(t.Context()) - fmt.Println("router stopped") - }) + router.Start(t.Context()) time.Sleep(time.Second * 1) @@ -281,12 +277,7 @@ func TestRouterInputNoMatchingRoute(t *testing.T) { t.Fatalf("router should not have returned any route errors: %v", routeErrors) } - routerRunner := sync.WaitGroup{} - - routerRunner.Go(func() { - router.Start(t.Context()) - fmt.Println("router stopped") - }) + router.Start(t.Context()) time.Sleep(time.Second * 1) @@ -332,12 +323,7 @@ func TestRouterInputSingleRoute(t *testing.T) { t.Fatalf("router should not have returned any route errors: %v", routeErrors) } - routerRunner := sync.WaitGroup{} - - routerRunner.Go(func() { - router.Start(t.Context()) - fmt.Println("router stopped") - }) + router.Start(t.Context()) time.Sleep(time.Second * 1) @@ -425,11 +411,7 @@ func TestRouterInputMultipleRoutes(t *testing.T) { t.Fatalf("router should not have returned any route errors: %v", routeErrors) } - routerRunner := sync.WaitGroup{} - - routerRunner.Go(func() { - router.Start(t.Context()) - }) + router.Start(t.Context()) time.Sleep(time.Second * 1) defer router.Stop() @@ -510,11 +492,7 @@ func TestRouterInputMultipleModules(t *testing.T) { t.Fatalf("router should not have returned any route errors: %v", routeErrors) } - routerRunner := sync.WaitGroup{} - - routerRunner.Go(func() { - router.Start(t.Context()) - }) + router.Start(t.Context()) time.Sleep(time.Second * 1)