Merge pull request #158 from jwetzell/module-context-handling

rework context handling throughout router and modules

cmd/showbridge/.gitignore

@@ -1,2 +0,0 @@
-# Added by goreleaser init:
-dist/

cmd/showbridge/.goreleaser.yaml

@@ -1,33 +0,0 @@
-version: 2
-project_name: showbridge
-before:
-  hooks:
-    - go mod tidy
-
-builds:
-  - env:
-      - CGO_ENABLED=0
-    goos:
-      - linux
-      - windows
-      - darwin
-    goarch:
-      - "amd64"
-      - "arm64"
-    ldflags:
-      - "-s -w -X main.version={{.RawVersion}}-{{.ShortCommit}}"
-
-archives:
-  - formats: [tar.gz]
-    name_template: "{{ .ProjectName }}_{{ .Version }}_{{ .Os }}-{{ .Arch }}"
-    format_overrides:
-      - goos: windows
-        formats: [zip]
-
-changelog:
-  sort: asc
-  use: github-native
-
-release:
-  draft: true
-  replace_existing_draft: true

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
 }
 

config.smoke.yaml

@@ -0,0 +1,85 @@
+api:
+  enabled: true
+  port: 8080
+modules:
+  - id: sqlite
+    type: db.sqlite
+    params:
+      dsn: ":memory:"
+  - id: httpin
+    type: http.server
+    params:
+      port: 3000
+  - id: tcpout
+    type: net.tcp.client
+    params:
+      host: "localhost"
+      port: 9000
+      framing: "RAW"
+  - id: tcpin
+    type: net.tcp.server
+    params:
+      port: 9000
+      framing: "RAW"
+  - id: ticker
+    type: time.interval
+    params:
+      duration: 1000
+  - id: udpout
+    type: net.udp.client
+    params:
+      host: "localhost"
+      port: 9001
+  - id: udpin
+    type: net.udp.server
+    params:
+      port: 9001
+  - id: wsecho
+    type: websocket.client
+    params:
+      url: "wss://echo.websocket.org"
+routes:
+  - input: ticker
+    processors:
+      - type: string.create
+        params:
+          template: "hello"
+      - type: router.output
+        params:
+          module: wsecho
+      - type: string.encode
+      - type: router.output
+        params:
+          module: tcpout
+      - type: router.output
+        params:
+          module: udpout
+  - input: ticker
+    processors:
+      - type: http.request.do
+        params:
+          url: "http://localhost:3000/echo"
+          method: "GET"
+      - type: debug.log
+  - input: tcpin
+    processors:
+      - type: string.decode
+      - type: debug.log
+  - input: httpin
+    processors:
+      - type: debug.log
+  - input: udpin
+    processors:
+      - type: string.decode
+      - type: debug.log
+  - input: wsecho
+    processors:
+      - type: debug.log
+  - input: ticker
+    processors:
+      - type: db.query
+        params:
+          module: "sqlite"
+          query: "SELECT 1;"
+      - type: debug.log
+  

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) {

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")
 	}

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")
 }

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")
 }

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")
 }

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")
 }

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")
 }

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")
 }

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")
 }

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) {

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")
 }

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")
 }

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")
 }

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")
+
 }

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")
 }

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")
 }

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")
 }

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")
 }

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")
 }

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")
 }

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")
 }

internal/processor/db-query.go

@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"text/template"
 
+	"github.com/google/jsonschema-go/jsonschema"
 	"github.com/jwetzell/showbridge-go/internal/common"
 	"github.com/jwetzell/showbridge-go/internal/config"
 )
@@ -116,6 +117,23 @@ func init() {
 	RegisterProcessor(ProcessorRegistration{
 		Type:  "db.query",
 		Title: "Query Database",
+		ParamsSchema: &jsonschema.Schema{
+			Type: "object",
+			Properties: map[string]*jsonschema.Schema{
+				"module": {
+					Title:       "Module ID",
+					Type:        "string",
+					Description: "ID of the database module to query",
+				},
+				"query": {
+					Title:       "Query",
+					Type:        "string",
+					Description: "SQL query to execute",
+				},
+			},
+			Required:             []string{"module", "query"},
+			AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
+		},
 		New: func(config config.ProcessorConfig) (Processor, error) {
 
 			params := config.Params

internal/processor/debug-log.go

@@ -9,6 +9,8 @@ import (
 	"github.com/jwetzell/showbridge-go/internal/config"
 )
 
+// TODO(jwetzell): maybe make a more useful logging processor
+
 type DebugLog struct {
 	config config.ProcessorConfig
 	logger *slog.Logger

internal/processor/sip-response-dtmf-create.go

@@ -27,6 +27,8 @@ type SipDTMFResponse struct {
 	Digits   string
 }
 
+var validDTMFRegex = regexp.MustCompile(`^[0-9*#A-Da-d]+$`)
+
 func (srdc *SipResponseDTMFCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
 
 	templateData := wrappedPayload
@@ -103,7 +105,7 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
-			return &SipResponseDTMFCreate{config: config, Digits: digitsTemplate, PreWait: int(preWaitNum), PostWait: int(postWaitNum), validDTMF: regexp.MustCompile(`^[0-9*#A-Da-d]+$`)}, nil
+			return &SipResponseDTMFCreate{config: config, Digits: digitsTemplate, PreWait: int(preWaitNum), PostWait: int(postWaitNum), validDTMF: validDTMFRegex}, nil
 		},
 	})
 }

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)
+		}
+	}
+}

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)