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

rework context handling throughout router and modules

.github/workflows/release-showbridge.yaml

@@ -4,13 +4,45 @@ on:
   push:
     tags:
       - "v*"
+  workflow_dispatch:
 
 permissions:
   contents: write
 
 jobs:
-  goreleaser:
+  setup_release:
     runs-on: ubuntu-latest
+    outputs:
+      version: ${{ steps.set_version.outputs.version }}
+      short_commit: ${{ steps.set_commit.outputs.short_commit }}
+    steps:
+      - name: checkout
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+      - name: set environment variables
+        run: echo "SHOWBRIDGE_VERSION=${GITHUB_REF_NAME}" >> $GITHUB_ENV
+      - name: clean tag
+        if: startsWith(github.ref, 'refs/tags/v')
+        run: echo "SHOWBRIDGE_VERSION=${SHOWBRIDGE_VERSION#v}" >> $GITHUB_ENV
+      - name: set version
+        id: set_version
+        run: echo "version=${SHOWBRIDGE_VERSION}" >> $GITHUB_OUTPUT
+      - name: set commit
+        id: set_commit
+        run: echo "short_commit=${GITHUB_SHA::7}" >> $GITHUB_OUTPUT
+      - name: create release
+        uses: softprops/action-gh-release@v3
+        if: github.ref_type == 'tag' && github.event_name != 'workflow_dispatch' && startsWith(github.ref, 'refs/tags/v')
+        with:
+          draft: true
+          generate_release_notes: true
+  linux_amd64:
+    runs-on: ubuntu-24.04
+    needs: setup_release
+    env:
+      SHOWBRIDGE_VERSION: ${{ needs.setup_release.outputs.version }}
+      SHORT_COMMIT: ${{ needs.setup_release.outputs.short_commit }}
     steps:
       - name: checkout
         uses: actions/checkout@v6
@@ -20,18 +52,166 @@ jobs:
         uses: actions/setup-go@v6
         with:
           go-version-file: "go.mod"
-      - name: release
-        uses: goreleaser/goreleaser-action@v7
+      - name: install alsa lib
+        run: sudo apt-get install -y libasound2-dev
+      - name: build
+        run: |
+          CGO_ENABLED=1 go build -o showbridge -ldflags="-s -w -X main.version=${SHOWBRIDGE_VERSION}-${SHORT_COMMIT}" ./cmd/showbridge
+          tar -czf showbridge_${SHOWBRIDGE_VERSION}_linux-amd64.tar.gz showbridge
+      - name: upload linux artifacts
+        uses: actions/upload-artifact@v7
+        if: github.event_name == 'workflow_dispatch'
         with:
-          workdir: cmd/showbridge
-          distribution: goreleaser
-          version: "v2.13.0"
-          args: release --clean
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          name: showbridge-linux-amd64
+          path: |
+            showbridge_*.tar.gz
+      - name: upload artifacts to release
+        uses: softprops/action-gh-release@v3
+        if: github.ref_type == 'tag' && github.event_name != 'workflow_dispatch' && startsWith(github.ref, 'refs/tags/v')
+        with:
+          draft: true
+          files: showbridge_*.tar.gz
+  linux_arm64:
+    runs-on: ubuntu-24.04-arm
+    needs: setup_release
+    env:
+      SHOWBRIDGE_VERSION: ${{ needs.setup_release.outputs.version }}
+      SHORT_COMMIT: ${{ needs.setup_release.outputs.short_commit }}
+    steps:
+      - name: checkout
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+      - name: setup go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: "go.mod"
+      - name: install alsa lib
+        run: sudo apt-get install -y libasound2-dev
+      - name: build
+        run: |
+          CGO_ENABLED=1 go build -o showbridge -ldflags="-s -w -X main.version=${SHOWBRIDGE_VERSION}-${SHORT_COMMIT}" ./cmd/showbridge
+          tar -czf showbridge_${SHOWBRIDGE_VERSION}_linux-arm64.tar.gz showbridge
+      - name: upload linux artifacts
+        uses: actions/upload-artifact@v7
+        if: github.event_name == 'workflow_dispatch'
+        with:
+          name: showbridge-linux-arm64
+          path: |
+            showbridge_*.tar.gz
+      - name: upload artifacts to release
+        uses: softprops/action-gh-release@v3
+        if: github.ref_type == 'tag' && github.event_name != 'workflow_dispatch' && startsWith(github.ref, 'refs/tags/v')
+        with:
+          draft: true
+          files: showbridge_*.tar.gz
+  windows:
+    runs-on: ubuntu-latest
+    needs: setup_release
+    env:
+      SHOWBRIDGE_VERSION: ${{ needs.setup_release.outputs.version }}
+      SHORT_COMMIT: ${{ needs.setup_release.outputs.short_commit }}
+    steps:
+      - name: checkout
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+      - name: setup go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: "go.mod"
+      - name: setup zig
+        uses: mlugg/setup-zig@v2
+        with:
+          version: 0.16.0
+      - name: build amd64
+        run: |
+          CGO_ENABLED=1 GOOS=windows GOARCH=amd64 CC="zig cc -target x86_64-windows-gnu" CXX="zig c++ -target x86_64-windows-gnu" go build -o showbridge.exe -ldflags="-s -w -X main.version=${SHOWBRIDGE_VERSION}-${SHORT_COMMIT}" ./cmd/showbridge
+          zip showbridge_${SHOWBRIDGE_VERSION}_windows-amd64.zip showbridge.exe
+      - name: build arm64
+        run: |
+          CGO_ENABLED=1 GOOS=windows GOARCH=arm64 CC="zig cc -target aarch64-windows-gnu" CXX="zig c++ -target aarch64-windows-gnu" go build -o showbridge.exe -ldflags="-s -w -X main.version=${SHOWBRIDGE_VERSION}-${SHORT_COMMIT}" ./cmd/showbridge
+          zip showbridge_${SHOWBRIDGE_VERSION}_windows-arm64.zip showbridge.exe
+      - name: upload windows artifacts
+        uses: actions/upload-artifact@v7
+        if: github.event_name == 'workflow_dispatch'
+        with:
+          name: showbridge-windows
+          path: |
+            showbridge_*.zip
+      - name: upload artifacts to release
+        uses: softprops/action-gh-release@v3
+        if: github.ref_type == 'tag' && github.event_name != 'workflow_dispatch' && startsWith(github.ref, 'refs/tags/v')
+        with:
+          draft: true
+          files: showbridge_*.zip
+  macos_arm64:
+    runs-on: macos-26
+    needs: setup_release
+    env:
+      SHOWBRIDGE_VERSION: ${{ needs.setup_release.outputs.version }}
+      SHORT_COMMIT: ${{ needs.setup_release.outputs.short_commit }}
+    steps:
+      - name: checkout
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+      - name: setup go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: "go.mod"
+      - name: build
+        run: |
+          CGO_ENABLED=1 GOOS=darwin GOARCH=arm64 go build -o showbridge -ldflags="-s -w -X main.version=${SHOWBRIDGE_VERSION}-${SHORT_COMMIT}" ./cmd/showbridge
+          tar -czf showbridge_${SHOWBRIDGE_VERSION}_macOS-arm64.tar.gz showbridge
+      - name: upload macos artifacts
+        uses: actions/upload-artifact@v7
+        if: github.event_name == 'workflow_dispatch'
+        with:
+          name: showbridge-macos-arm64
+          path: |
+            showbridge_*.tar.gz
+      - name: upload artifacts to release
+        uses: softprops/action-gh-release@v3
+        if: github.ref_type == 'tag' && github.event_name != 'workflow_dispatch' && startsWith(github.ref, 'refs/tags/v')
+        with:
+          draft: true
+          files: showbridge_*.tar.gz
+  macos_amd64:
+    runs-on: macos-26-intel
+    needs: setup_release
+    env:
+      SHOWBRIDGE_VERSION: ${{ needs.setup_release.outputs.version }}
+      SHORT_COMMIT: ${{ needs.setup_release.outputs.short_commit }}
+    steps:
+      - name: checkout
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+      - name: setup go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: "go.mod"
+      - name: build
+        run: |
+          CGO_ENABLED=1 GOOS=darwin GOARCH=amd64 go build -o showbridge -ldflags="-s -w -X main.version=${SHOWBRIDGE_VERSION}-${SHORT_COMMIT}" ./cmd/showbridge
+          tar -czf showbridge_${SHOWBRIDGE_VERSION}_macOS-amd64.tar.gz showbridge
+      - name: upload macos artifacts
+        uses: actions/upload-artifact@v7
+        if: github.event_name == 'workflow_dispatch'
+        with:
+          name: showbridge-macos-amd64
+          path: |
+            showbridge_*.tar.gz
+      - name: upload artifacts to release
+        uses: softprops/action-gh-release@v3
+        if: github.ref_type == 'tag' && github.event_name != 'workflow_dispatch' && startsWith(github.ref, 'refs/tags/v')
+        with:
+          draft: true
+          files: showbridge_*.tar.gz
   docker:
     runs-on: ubuntu-latest
-    needs: goreleaser
+    needs: linux_amd64
     steps:
       - name: Check out repository
         uses: actions/checkout@1af3b93b6815bc44a9784bd300feb67ff0d1eeb3 # v6.0.0
@@ -57,6 +237,7 @@ jobs:
 
       - name: Build and push
         uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        if: github.ref_type == 'tag'
         with:
           push: true
           context: ./

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) {
@@ -127,6 +126,9 @@ func readConfig(configPath string) (config.Config, error) {
 	}
 
 	validatedConfigBytes, err := json.Marshal(yamlMap)
+	if err != nil {
+		return config.Config{}, err
+	}
 
 	err = json.Unmarshal(validatedConfigBytes, &cfg)
 	if err != nil {
@@ -142,7 +144,7 @@ func writeConfig(configPath string, newConfig config.Config) error {
 		return err
 	}
 
-	err = os.WriteFile(configPath, configBytes, 0644)
+	err = os.WriteFile(configPath, configBytes, 0600)
 	if err != nil {
 		return err
 	}
@@ -156,7 +158,7 @@ func run(ctx context.Context, c *cli.Command) error {
 		return errors.New("config path cannot be empty")
 	}
 
-	logLevel := slog.LevelInfo
+	var logLevel slog.Level
 
 	logLevelFromFlag := c.String("log-level")
 
@@ -208,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)
@@ -233,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()
@@ -243,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
 }
 
@@ -260,7 +257,7 @@ func (app *showbridgeApp) handleChannels() {
 				app.routerMutex.Unlock()
 				continue
 			}
-			err, moduleErrors, routeErrors := app.router.UpdateConfig(config, false)
+			moduleErrors, routeErrors, err := app.router.UpdateConfig(config, false)
 			if err != nil {
 				app.logger.Error("failed to update router config", "error", err)
 				app.routerMutex.Unlock()

config.go

@@ -15,9 +15,9 @@ func (r *Router) GetRunningConfig() config.Config {
 	return r.runningConfig
 }
 
-func (r *Router) UpdateConfig(newConfig config.Config, triggerChangeChan bool) (error, []config.ModuleError, []config.RouteError) {
+func (r *Router) UpdateConfig(newConfig config.Config, triggerChangeChan bool) ([]config.ModuleError, []config.RouteError, error) {
 	if !r.runningConfigMu.TryLock() {
-		return errors.New("config update in progress"), nil, nil
+		return nil, nil, errors.New("config update in progress")
 	}
 	defer r.runningConfigMu.Unlock()
 	oldConfig := r.runningConfig
@@ -81,5 +81,5 @@ func (r *Router) UpdateConfig(newConfig config.Config, triggerChangeChan bool) (
 		r.ConfigChange <- newConfig
 	}
 
-	return nil, moduleErrors, routeErrors
+	return moduleErrors, routeErrors, 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
+  

go.mod

@@ -10,8 +10,8 @@ require (
 	github.com/extism/go-sdk v1.7.1
 	github.com/google/jsonschema-go v0.4.3
 	github.com/gorilla/websocket v1.5.3
-	github.com/jwetzell/artnet-go v0.2.1
-	github.com/jwetzell/free-d-go v0.1.0
+	github.com/jwetzell/artnet-go v0.2.2
+	github.com/jwetzell/free-d-go v0.1.1
 	github.com/jwetzell/osc-go v0.3.0
 	github.com/jwetzell/psn-go v0.3.0
 	github.com/nats-io/nats-server/v2 v2.14.0

go.sum

@@ -65,10 +65,10 @@ github.com/ianlancetaylor/demangle v0.0.0-20260505044615-1ff4bf46051f h1:NW3E2QS
 github.com/ianlancetaylor/demangle v0.0.0-20260505044615-1ff4bf46051f/go.mod h1:gx7rwoVhcfuVKG5uya9Hs3Sxj7EIvldVofAWIUtGouw=
 github.com/icholy/digest v1.1.0 h1:HfGg9Irj7i+IX1o1QAmPfIBNu/Q5A5Tu3n/MED9k9H4=
 github.com/icholy/digest v1.1.0/go.mod h1:QNrsSGQ5v7v9cReDI0+eyjsXGUoRSUZQHeQ5C4XLa0Y=
-github.com/jwetzell/artnet-go v0.2.1 h1:iYTKWcwYrF5kBkYfkw2UbWvoueeA23iKEn7fR27mWZE=
-github.com/jwetzell/artnet-go v0.2.1/go.mod h1:gli97Z32a0kMkZ6taoTiK7/lqHcF/dhiGjGJdx/PxqA=
-github.com/jwetzell/free-d-go v0.1.0 h1:xHt6dvyit98X+OC3jVzV0aLidxbyzi3vI9QiYkteEtA=
-github.com/jwetzell/free-d-go v0.1.0/go.mod h1:KmrkooRARRaxJTBSPvwt/6IMAIaHH1R8bSA8cwbbELw=
+github.com/jwetzell/artnet-go v0.2.2 h1:vaXSmRrSlEdETPZ/gvw6aAmVqzQm0i+gfAwyIQkVsB0=
+github.com/jwetzell/artnet-go v0.2.2/go.mod h1:gli97Z32a0kMkZ6taoTiK7/lqHcF/dhiGjGJdx/PxqA=
+github.com/jwetzell/free-d-go v0.1.1 h1:pzY2c4qRxKFBZ2jO3z512ysvUI0BUmgd4mSlWPiFVRM=
+github.com/jwetzell/free-d-go v0.1.1/go.mod h1:KmrkooRARRaxJTBSPvwt/6IMAIaHH1R8bSA8cwbbELw=
 github.com/jwetzell/osc-go v0.3.0 h1:z75TxuQSEmdcmZ56OAepkDa3m88SdZh//3m4nBb/XZI=
 github.com/jwetzell/osc-go v0.3.0/go.mod h1:kCs329JxY6Qjga08tRQ/Gl0PqhgQzLIMpOhm6uszvIc=
 github.com/jwetzell/psn-go v0.3.0 h1:WVpCEmExYE8a+I5hQak5jNJJp2x35VdGX/VuMUKPmhY=

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
@@ -53,13 +52,16 @@ func (as *ApiServer) Start(config config.ApiConfig) {
 	as.serverMu.Lock()
 	defer as.serverMu.Unlock()
 	as.server = &http.Server{
-		Addr:    fmt.Sprintf(":%d", as.config.Port),
-		Handler: mux,
+		Addr:              fmt.Sprintf(":%d", as.config.Port),
+		ReadHeaderTimeout: 5 * time.Second,
+		Handler:           mux,
 	}
 
 	go func() {
-		as.server.ListenAndServe()
-		as.shutdown()
+		err := as.server.ListenAndServe()
+		if err != nil && err != http.ErrServerClosed {
+			as.logger.Error("server error", "error", err)
+		}
 	}()
 }
 
@@ -67,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) {
@@ -148,7 +152,7 @@ func (as *ApiServer) handleConfigHTTP(w http.ResponseWriter, req *http.Request)
 			http.Error(w, "Bad request", http.StatusBadRequest)
 			return
 		}
-		err, moduleErrors, routeErrors := as.configurableRouter.UpdateConfig(newConfig, true)
+		moduleErrors, routeErrors, err := as.configurableRouter.UpdateConfig(newConfig, true)
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusConflict)
 			return

internal/common/common.go

@@ -81,11 +81,19 @@ func GetAnyAsByte(value any) (byte, bool) {
 }
 
 func GetAnyAsByteSlice(value any) ([]byte, bool) {
+	// already a []byte
+	byteSlice, ok := value.([]byte)
+	if ok {
+		return byteSlice, true
+	}
+
+	// check for a slice
 	v := reflect.ValueOf(value)
 	if v.Kind() != reflect.Slice {
 		return nil, false
 	}
 
+	// try to convert each element to a byte
 	result := make([]byte, v.Len())
 	for i := 0; i < v.Len(); i++ {
 		elem := v.Index(i).Interface()
@@ -99,6 +107,14 @@ func GetAnyAsByteSlice(value any) ([]byte, bool) {
 }
 
 func GetAnyAsIntSlice(value any) ([]int, bool) {
+
+	// already a []byte
+	intSlice, ok := value.([]int)
+	if ok {
+		return intSlice, true
+	}
+
+	// check for a slice
 	v := reflect.ValueOf(value)
 	if v.Kind() != reflect.Slice {
 		return nil, false

internal/common/module.go

@@ -22,5 +22,5 @@ type KeyValueModule interface {
 }
 
 type DatabaseModule interface {
-	Database() *sql.DB
+	Database() (*sql.DB, error)
 }

internal/config/config.go

@@ -7,6 +7,6 @@ type Config struct {
 }
 
 type Configurable interface {
-	UpdateConfig(newConfig Config, triggerChangeChannel bool) (error, []ModuleError, []RouteError)
+	UpdateConfig(newConfig Config, triggerChangeChannel bool) ([]ModuleError, []RouteError, error)
 	GetRunningConfig() Config
 }

internal/framer/slip.go

@@ -25,9 +25,10 @@ func (sf *SlipFramer) Decode(data []byte) [][]byte {
 		}
 
 		if escapeNext {
-			if packetByte == ESC_END {
+			switch packetByte {
+			case ESC_END:
 				sf.buffer = append(sf.buffer, END)
-			} else if packetByte == ESC_ESC {
+			case ESC_ESC:
 				sf.buffer = append(sf.buffer, ESC)
 			}
 			escapeNext = false

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,24 +64,40 @@ 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")
 }
 
-func (t *DbSqlite) Database() *sql.DB {
-	return t.db
+// 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")
+	}
+	return t.db, nil
 }

internal/module/http-server.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"log/slog"
 	"net/http"
+	"sync"
+	"time"
 
 	"github.com/google/jsonschema-go/jsonschema"
 	"github.com/jwetzell/showbridge-go/internal/common"
@@ -15,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 {
@@ -163,14 +167,14 @@ func (hs *HTTPServer) Start(ctx context.Context, router common.RouteIO) error {
 	hs.cancel = cancel
 
 	httpServer := &http.Server{
-		Addr:    fmt.Sprintf(":%d", hs.Port),
-		Handler: hs,
+		Addr:              fmt.Sprintf(":%d", hs.Port),
+		ReadHeaderTimeout: 5 * time.Second,
+		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
@@ -181,7 +185,6 @@ func (hs *HTTPServer) Start(ctx context.Context, router common.RouteIO) error {
 	}
 
 	<-hs.ctx.Done()
-	hs.logger.Debug("done")
 	return nil
 }
 
@@ -208,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() {
@@ -43,8 +45,13 @@ func init() {
 					Title: "Baud Rate",
 					Type:  "integer",
 				},
+				"framing": {
+					Title: "Framing Method",
+					Type:  "string",
+					Enum:  []any{"LF", "CR", "CRLF", "SLIP", "RAW"},
+				},
 			},
-			Required:             []string{"port", "baudRate"},
+			Required:             []string{"port", "baudRate", "framing"},
 			AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
 		},
 		New: func(config config.ModuleConfig) (common.Module, error) {
@@ -88,10 +95,11 @@ 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 fmt.Errorf("serial.client can't open input port: %s", sc.Port)
+		return err
 	}
 
 	sc.port = port
@@ -106,23 +114,13 @@ 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")
 				return nil
 			}
-			sc.logger.Error("port setup error", "error", err.Error())
+			sc.logger.Error("port setup error", "port", sc.Port, "error", err.Error())
 			time.Sleep(time.Second * 2)
 			continue
 		}
@@ -130,14 +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")
 			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)
@@ -163,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 {
@@ -178,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
 }
 
@@ -236,7 +231,7 @@ func (ts *TCPServer) Output(ctx context.Context, payload any) error {
 	for _, connection := range ts.connections {
 		_, err := connection.Write(payloadBytes)
 		if err != nil {
-			errorString.WriteString(fmt.Sprintf("%s\n", err.Error()))
+			fmt.Fprintf(&errorString, "%s\n", err.Error())
 		}
 	}
 	ts.connectionsMu.Unlock()
@@ -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"
 )
@@ -17,34 +18,43 @@ type DbQuery struct {
 	ModuleId string
 	Query    *template.Template
 	logger   *slog.Logger
+	module   common.DatabaseModule
 }
 
 func (dq *DbQuery) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
-	if wrappedPayload.Modules == nil {
-		wrappedPayload.End = true
-		return wrappedPayload, errors.New("db.query wrapped payload has no modules")
+	if dq.module == nil {
+		if wrappedPayload.Modules == nil {
+			wrappedPayload.End = true
+			return wrappedPayload, errors.New("db.query wrapped payload has no modules")
+		}
+
+		module, ok := wrappedPayload.Modules[dq.ModuleId]
+		if !ok {
+			wrappedPayload.End = true
+			return wrappedPayload, fmt.Errorf("db.query unable to find module with id: %s", dq.ModuleId)
+		}
+
+		dbModule, ok := module.(common.DatabaseModule)
+		if !ok {
+			wrappedPayload.End = true
+			return wrappedPayload, fmt.Errorf("db.query module with id %s is not a DatabaseModule", dq.ModuleId)
+		}
+		dq.module = dbModule
 	}
 
-	module, ok := wrappedPayload.Modules[dq.ModuleId]
-	if !ok {
+	db, err := dq.module.Database()
+	if err != nil {
 		wrappedPayload.End = true
-		return wrappedPayload, fmt.Errorf("db.query unable to find module with id: %s", dq.ModuleId)
+		return wrappedPayload, fmt.Errorf("db.query error getting database from module: %w", err)
 	}
 
-	dbModule, ok := module.(common.DatabaseModule)
-	if !ok {
-		wrappedPayload.End = true
-		return wrappedPayload, fmt.Errorf("db.query module with id %s is not a DatabaseModule", dq.ModuleId)
-	}
-
-	db := dbModule.Database()
 	if db == nil {
 		wrappedPayload.End = true
 		return wrappedPayload, fmt.Errorf("db.query module with id %s returned nil database", dq.ModuleId)
 	}
 
 	var queryBuffer bytes.Buffer
-	err := dq.Query.Execute(&queryBuffer, wrappedPayload)
+	err = dq.Query.Execute(&queryBuffer, wrappedPayload)
 
 	if err != nil {
 		wrappedPayload.End = true
@@ -65,6 +75,7 @@ func (dq *DbQuery) Process(ctx context.Context, wrappedPayload common.WrappedPay
 		return wrappedPayload, fmt.Errorf("db.query error getting columns: %w", err)
 	}
 
+	// TODO(jwetzell): optimize this
 	results := make([]map[string]any, 0)
 
 	for rows.Next() {
@@ -106,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/kv-get.go

@@ -16,27 +16,31 @@ type KVGet struct {
 	ModuleId string
 	Key      string
 	logger   *slog.Logger
+	module   common.KeyValueModule
 }
 
 func (kvg *KVGet) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
-	if wrappedPayload.Modules == nil {
-		wrappedPayload.End = true
-		return wrappedPayload, errors.New("kv.get wrapped payload has no modules")
+	if kvg.module == nil {
+		if wrappedPayload.Modules == nil {
+			wrappedPayload.End = true
+			return wrappedPayload, errors.New("kv.get wrapped payload has no modules")
+		}
+
+		module, ok := wrappedPayload.Modules[kvg.ModuleId]
+		if !ok {
+			wrappedPayload.End = true
+			return wrappedPayload, fmt.Errorf("kv.get unable to find module with id: %s", kvg.ModuleId)
+		}
+
+		kvModule, ok := module.(common.KeyValueModule)
+		if !ok {
+			wrappedPayload.End = true
+			return wrappedPayload, fmt.Errorf("kv.get module with id %s is not a KeyValueModule", kvg.ModuleId)
+		}
+		kvg.module = kvModule
 	}
 
-	module, ok := wrappedPayload.Modules[kvg.ModuleId]
-	if !ok {
-		wrappedPayload.End = true
-		return wrappedPayload, fmt.Errorf("kv.get unable to find module with id: %s", kvg.ModuleId)
-	}
-
-	kvModule, ok := module.(common.KeyValueModule)
-	if !ok {
-		wrappedPayload.End = true
-		return wrappedPayload, fmt.Errorf("kv.get module with id %s is not a KeyValueModule", kvg.ModuleId)
-	}
-
-	value, err := kvModule.Get(kvg.Key)
+	value, err := kvg.module.Get(kvg.Key)
 	if err != nil {
 		wrappedPayload.End = true
 		return wrappedPayload, fmt.Errorf("kv.get error getting key: %w", err)

internal/processor/kv-set.go

@@ -19,25 +19,28 @@ type KVSet struct {
 	Key      string
 	Value    *template.Template
 	logger   *slog.Logger
+	module   common.KeyValueModule
 }
 
 func (kvs *KVSet) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
+	if kvs.module == nil {
+		if wrappedPayload.Modules == nil {
+			wrappedPayload.End = true
+			return wrappedPayload, errors.New("kv.set wrapped payload has no modules")
+		}
 
-	if wrappedPayload.Modules == nil {
-		wrappedPayload.End = true
-		return wrappedPayload, errors.New("kv.set wrapped payload has no modules")
-	}
+		module, ok := wrappedPayload.Modules[kvs.ModuleId]
+		if !ok {
+			wrappedPayload.End = true
+			return wrappedPayload, fmt.Errorf("kv.set unable to find module with id: %s", kvs.ModuleId)
+		}
 
-	module, ok := wrappedPayload.Modules[kvs.ModuleId]
-	if !ok {
-		wrappedPayload.End = true
-		return wrappedPayload, fmt.Errorf("kv.set unable to find module with id: %s", kvs.ModuleId)
-	}
-
-	kvModule, ok := module.(common.KeyValueModule)
-	if !ok {
-		wrappedPayload.End = true
-		return wrappedPayload, fmt.Errorf("kv.set module with id %s is not a KeyValueModule", kvs.ModuleId)
+		kvModule, ok := module.(common.KeyValueModule)
+		if !ok {
+			wrappedPayload.End = true
+			return wrappedPayload, fmt.Errorf("kv.set module with id %s is not a KeyValueModule", kvs.ModuleId)
+		}
+		kvs.module = kvModule
 	}
 
 	var valueBuffer bytes.Buffer
@@ -48,7 +51,7 @@ func (kvs *KVSet) Process(ctx context.Context, wrappedPayload common.WrappedPayl
 		return wrappedPayload, err
 	}
 
-	err = kvModule.Set(kvs.Key, valueBuffer.String())
+	err = kvs.module.Set(kvs.Key, valueBuffer.String())
 	if err != nil {
 		wrappedPayload.End = true
 		return wrappedPayload, fmt.Errorf("kv.set error setting key: %w", err)

internal/processor/midi-control_change-create.go

@@ -35,6 +35,11 @@ func (mccc *MIDIControlChangeCreate) Process(ctx context.Context, wrappedPayload
 
 	channelValue, err := strconv.ParseUint(channelBuffer.String(), 10, 8)
 
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	var controlBuffer bytes.Buffer
 	err = mccc.Control.Execute(&controlBuffer, templateData)
 
@@ -45,6 +50,11 @@ func (mccc *MIDIControlChangeCreate) Process(ctx context.Context, wrappedPayload
 
 	controlValue, err := strconv.ParseUint(controlBuffer.String(), 10, 8)
 
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	var valueBuffer bytes.Buffer
 	err = mccc.Value.Execute(&valueBuffer, templateData)
 
@@ -55,6 +65,11 @@ func (mccc *MIDIControlChangeCreate) Process(ctx context.Context, wrappedPayload
 
 	valueValue, err := strconv.ParseUint(valueBuffer.String(), 10, 8)
 
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	payloadMessage := midi.ControlChange(uint8(channelValue), uint8(controlValue), uint8(valueValue))
 	wrappedPayload.Payload = payloadMessage
 	return wrappedPayload, nil

internal/processor/midi-note_off-create.go

@@ -35,6 +35,11 @@ func (mnoc *MIDINoteOffCreate) Process(ctx context.Context, wrappedPayload commo
 
 	channelValue, err := strconv.ParseUint(channelBuffer.String(), 10, 8)
 
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	var noteBuffer bytes.Buffer
 	err = mnoc.Note.Execute(&noteBuffer, templateData)
 
@@ -45,6 +50,11 @@ func (mnoc *MIDINoteOffCreate) Process(ctx context.Context, wrappedPayload commo
 
 	noteValue, err := strconv.ParseUint(noteBuffer.String(), 10, 8)
 
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	var velocityBuffer bytes.Buffer
 	err = mnoc.Velocity.Execute(&velocityBuffer, templateData)
 
@@ -54,6 +64,12 @@ func (mnoc *MIDINoteOffCreate) Process(ctx context.Context, wrappedPayload commo
 	}
 
 	velocityValue, err := strconv.ParseUint(velocityBuffer.String(), 10, 8)
+
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	payloadMessage := midi.NoteOffVelocity(uint8(channelValue), uint8(noteValue), uint8(velocityValue))
 	wrappedPayload.Payload = payloadMessage
 	return wrappedPayload, nil

internal/processor/midi-note_on-create.go

@@ -35,6 +35,11 @@ func (mnoc *MIDINoteOnCreate) Process(ctx context.Context, wrappedPayload common
 
 	channelValue, err := strconv.ParseUint(channelBuffer.String(), 10, 8)
 
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	var noteBuffer bytes.Buffer
 	err = mnoc.Note.Execute(&noteBuffer, templateData)
 
@@ -45,6 +50,11 @@ func (mnoc *MIDINoteOnCreate) Process(ctx context.Context, wrappedPayload common
 
 	noteValue, err := strconv.ParseUint(noteBuffer.String(), 10, 8)
 
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	var velocityBuffer bytes.Buffer
 	err = mnoc.Velocity.Execute(&velocityBuffer, templateData)
 
@@ -54,6 +64,11 @@ func (mnoc *MIDINoteOnCreate) Process(ctx context.Context, wrappedPayload common
 	}
 
 	velocityValue, err := strconv.ParseUint(velocityBuffer.String(), 10, 8)
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	payloadMessage := midi.NoteOn(uint8(channelValue), uint8(noteValue), uint8(velocityValue))
 	wrappedPayload.Payload = payloadMessage
 	return wrappedPayload, nil

internal/processor/midi-program_change-create.go

@@ -34,6 +34,11 @@ func (mpcc *MIDIProgramChangeCreate) Process(ctx context.Context, wrappedPayload
 
 	channelValue, err := strconv.ParseUint(channelBuffer.String(), 10, 8)
 
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	var programBuffer bytes.Buffer
 	err = mpcc.Program.Execute(&programBuffer, templateData)
 
@@ -44,6 +49,11 @@ func (mpcc *MIDIProgramChangeCreate) Process(ctx context.Context, wrappedPayload
 
 	programValue, err := strconv.ParseUint(programBuffer.String(), 10, 8)
 
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+
 	payloadMessage := midi.ProgramChange(uint8(channelValue), uint8(programValue))
 	wrappedPayload.Payload = payloadMessage
 	return wrappedPayload, nil

internal/processor/script-js.go

@@ -32,9 +32,13 @@ func (sj *ScriptJS) Process(ctx context.Context, wrappedPayload common.WrappedPa
 		}
 	}
 
-	sj.vm.SetProperty(sj.vm.GlobalObject(), sj.payloadAtom, wrappedPayload.Payload)
+	err := sj.vm.SetProperty(sj.vm.GlobalObject(), sj.payloadAtom, wrappedPayload.Payload)
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
 
-	_, err := sj.vm.Eval(sj.Program, quickjs.EvalGlobal)
+	_, err = sj.vm.Eval(sj.Program, quickjs.EvalGlobal)
 
 	if err != nil {
 		wrappedPayload.End = true

internal/processor/script-wasm.go

@@ -14,7 +14,7 @@ import (
 
 type ScriptWASM struct {
 	config   config.ProcessorConfig
-	Program  *extism.CompiledPlugin
+	Program  *extism.Plugin
 	Function string
 }
 
@@ -28,14 +28,7 @@ func (sw *ScriptWASM) Process(ctx context.Context, wrappedPayload common.Wrapped
 		return wrappedPayload, fmt.Errorf("script.wasm can only process a byte array")
 	}
 
-	program, err := sw.Program.Instance(ctx, extism.PluginInstanceConfig{})
-
-	if err != nil {
-		wrappedPayload.End = true
-		return wrappedPayload, err
-	}
-
-	_, output, err := program.Call(sw.Function, payloadBytes)
+	_, output, err := sw.Program.Call(sw.Function, payloadBytes)
 
 	if err != nil {
 		wrappedPayload.End = true
@@ -117,7 +110,13 @@ func init() {
 				return nil, err
 			}
 
-			return &ScriptWASM{config: processorConfig, Program: program, Function: functionString}, nil
+			programInstance, err := program.Instance(context.Background(), extism.PluginInstanceConfig{})
+
+			if err != nil {
+				return nil, err
+			}
+
+			return &ScriptWASM{config: processorConfig, Program: programInstance, Function: functionString}, nil
 		},
 	})
 }

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

internal/processor/test/artnet-packet-decode_test.go

@@ -41,7 +41,7 @@ func TestGoodArtnetPacketDecode(t *testing.T) {
 			name:    "number",
 			payload: []byte{65, 114, 116, 45, 78, 101, 116, 0, 0, 80, 0, 14, 237, 0, 1, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
 			expected: &artnet.ArtDmx{
-				ID:        []byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
+				ID:        [8]byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
 				OpCode:    artnet.OpDmx,
 				ProtVerHi: 0,
 				ProtVerLo: 14,
@@ -49,7 +49,6 @@ func TestGoodArtnetPacketDecode(t *testing.T) {
 				Physical:  0,
 				SubUni:    1,
 				Net:       0,
-				Length:    512,
 				Data:      make([]uint8, 512),
 			},
 		},
@@ -106,3 +105,15 @@ func TestBadArtnetPacketDecode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkArtnetPacketDecode(b *testing.B) {
+	processorInstance := processor.ArtNetPacketDecode{}
+	payload := []byte{65, 114, 116, 45, 78, 101, 116, 0, 0, 80, 0, 14, 237, 0, 1, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: payload})
+		if err != nil {
+			b.Fatalf("artnet.packet.decode processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/artnet-packet-encode_test.go

@@ -41,7 +41,7 @@ func TestGoodArtnetPacketEncode(t *testing.T) {
 			name:     "number",
 			expected: []byte{65, 114, 116, 45, 78, 101, 116, 0, 0, 80, 0, 14, 237, 0, 1, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
 			payload: &artnet.ArtDmx{
-				ID:        []byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
+				ID:        [8]byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
 				OpCode:    artnet.OpDmx,
 				ProtVerHi: 0,
 				ProtVerLo: 14,
@@ -49,7 +49,6 @@ func TestGoodArtnetPacketEncode(t *testing.T) {
 				Physical:  0,
 				SubUni:    1,
 				Net:       0,
-				Length:    512,
 				Data:      make([]uint8, 512),
 			},
 		},
@@ -101,3 +100,25 @@ func TestBadArtnetPacketEncode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkArtnetPacketEncode(b *testing.B) {
+	processorInstance := processor.ArtNetPacketEncode{}
+	payload := &artnet.ArtDmx{
+		ID:        [8]byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
+		OpCode:    artnet.OpDmx,
+		ProtVerHi: 0,
+		ProtVerLo: 14,
+		Sequence:  237,
+		Physical:  0,
+		SubUni:    1,
+		Net:       0,
+		Data:      make([]uint8, 512),
+	}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: payload})
+		if err != nil {
+			b.Fatalf("artnet.packet.encode processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/db-query_test.go

@@ -301,3 +301,32 @@ func TestBadDbQuery(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkDbQuery(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["db.query"]
+	if !ok {
+		b.Fatalf("db.query processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "db.query",
+		Params: map[string]any{
+			"module": "test",
+			"query":  "SELECT * FROM test;",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("db.query failed to create processor: %s", err)
+	}
+	modules := map[string]common.Module{
+		"test": test.NewTestDBModule("test"),
+	}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: nil, Modules: modules})
+		if err != nil {
+			b.Fatalf("db.query processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/debug-log_test.go

@@ -118,3 +118,29 @@ func TestBadDebugLog(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkDebugLog(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["debug.log"]
+	if !ok {
+		b.Fatalf("debug.log processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type:   "debug.log",
+		Params: nil,
+	})
+
+	if err != nil {
+		b.Fatalf("debug.log failed to create processor: %s", err)
+	}
+	modules := map[string]common.Module{}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count, Modules: modules})
+		if err != nil {
+			b.Fatalf("debug.log processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/filter-change_test.go

@@ -124,3 +124,29 @@ func TestBadFilterChange(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkFilterChange(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["filter.change"]
+	if !ok {
+		b.Fatalf("filter.change processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type:   "filter.change",
+		Params: nil,
+	})
+
+	if err != nil {
+		b.Fatalf("filter.change failed to create processor: %s", err)
+	}
+	modules := map[string]common.Module{}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count, Modules: modules})
+		if err != nil {
+			b.Fatalf("filter.change processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/filter-expr_test.go

@@ -171,3 +171,30 @@ func TestBadFilterExpr(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkFilterExpr(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["filter.expr"]
+	if !ok {
+		b.Fatalf("filter.expr processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "filter.expr",
+		Params: map[string]any{
+			"expression": "Payload % 2 == 0",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("filter.expr failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("filter.expr processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/filter-regex_test.go

@@ -1,6 +1,7 @@
 package processor_test
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/jwetzell/showbridge-go/internal/common"
@@ -173,3 +174,30 @@ func TestBadFilterRegex(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkFilterRegex(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["filter.regex"]
+	if !ok {
+		b.Fatalf("filter.regex processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "filter.regex",
+		Params: map[string]any{
+			"pattern": ".*",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("filter.regex failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: fmt.Sprintf("%d", count)})
+		if err != nil {
+			b.Fatalf("filter.regex processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/float-parse_test.go

@@ -1,6 +1,7 @@
 package processor_test
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/jwetzell/showbridge-go/internal/common"
@@ -164,3 +165,28 @@ func TestBadFloatParse(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkFloatParse(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["float.parse"]
+	if !ok {
+		b.Fatalf("float.parse processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type:   "float.parse",
+		Params: nil,
+	})
+
+	if err != nil {
+		b.Fatalf("float.parse failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: fmt.Sprintf("%d", count)})
+		if err != nil {
+			b.Fatalf("float.parse processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/float-random_test.go

@@ -184,3 +184,29 @@ func TestBadFloatRandom(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkFloatRandom(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["float.random"]
+	if !ok {
+		b.Fatalf("float.random processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "float.random",
+		Params: map[string]any{
+			"min": 0.0,
+			"max": 1.0,
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("float.random failed to create processor: %s", err)
+	}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: nil})
+		if err != nil {
+			b.Fatalf("float.random processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/free-d-create_test.go

@@ -877,3 +877,38 @@ func TestBadFreeDCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkFreeDCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["freed.create"]
+	if !ok {
+		b.Fatalf("freed.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "freed.create",
+		Params: map[string]any{
+			"id":    "0",
+			"pan":   "0",
+			"tilt":  "0",
+			"roll":  "0",
+			"posX":  "0",
+			"posY":  "0",
+			"posZ":  "0",
+			"zoom":  "0",
+			"focus": "0",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("freed.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("freed.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/free-d-decode_test.go

@@ -102,3 +102,17 @@ func TestBadFreeDDecode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkFreeDDecode(b *testing.B) {
+	processorInstance := processor.FreeDDecode{}
+	payload := []byte{0xd1, 0x01, 0x5a, 0x00, 0x00, 0x2d, 0x00, 0x00, 0xa6, 0x00, 0x00, 0x7f, 0xff, 0x40, 0x7f, 0xff, 0x80, 0x7f, 0xff,
+		0xc0, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x00, 0x00, 50,
+	}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: payload})
+		if err != nil {
+			b.Fatalf("freed.encode processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/free-d-encode_test.go

@@ -102,3 +102,25 @@ func TestBadFreeDEncode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkFreeDEncode(b *testing.B) {
+	processorInstance := processor.FreeDEncode{}
+	payload := freeD.FreeDPosition{
+		ID:    1,
+		Pan:   180,
+		Tilt:  90,
+		Roll:  -180,
+		PosX:  131069,
+		PosY:  131070,
+		PosZ:  131071,
+		Zoom:  66051,
+		Focus: 263430,
+	}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: payload})
+		if err != nil {
+			b.Fatalf("freed.encode processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/http-response-create_test.go

@@ -157,3 +157,31 @@ func TestBadHTTPResponseCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkHTTPResponseCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["http.response.create"]
+	if !ok {
+		b.Fatalf("http.response.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "http.response.create",
+		Params: map[string]any{
+			"status":       200,
+			"bodyTemplate": "Hello, {{.Payload}}!",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("http.response.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("http.response.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/int-parse_test.go

@@ -1,6 +1,7 @@
 package processor_test
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/jwetzell/showbridge-go/internal/common"
@@ -198,3 +199,28 @@ func TestBadIntParse(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkIntParse(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["int.parse"]
+	if !ok {
+		b.Fatalf("int.parse processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type:   "int.parse",
+		Params: nil,
+	})
+
+	if err != nil {
+		b.Fatalf("int.parse failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: fmt.Sprintf("%d", count)})
+		if err != nil {
+			b.Fatalf("int.parse processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/int-random_test.go

@@ -195,3 +195,29 @@ func TestBadIntRandom(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkIntRandom(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["int.random"]
+	if !ok {
+		b.Fatalf("int.random processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "int.random",
+		Params: map[string]any{
+			"min": 0,
+			"max": 100,
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("int.random failed to create processor: %s", err)
+	}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: nil})
+		if err != nil {
+			b.Fatalf("int.random processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/int-scale_test.go

@@ -169,3 +169,33 @@ func TestBadIntScale(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkIntScale(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["int.scale"]
+	if !ok {
+		b.Fatalf("int.scale processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "int.scale",
+		Params: map[string]any{
+			"inMin":  0,
+			"inMax":  100,
+			"outMin": 0,
+			"outMax": 127,
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("int.scale failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count % 100})
+		if err != nil {
+			b.Fatalf("int.scale processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/json-decode_test.go

@@ -1,6 +1,7 @@
 package processor_test
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 
@@ -124,3 +125,27 @@ func TestBadJsonDecode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkJsonDecode(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["json.decode"]
+	if !ok {
+		b.Fatalf("json.decode processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "json.decode",
+	})
+
+	if err != nil {
+		b.Fatalf("json.decode failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: []byte(fmt.Sprintf("{\"key\":%d}", count))})
+		if err != nil {
+			b.Fatalf("json.decode processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/json-encode_test.go

@@ -1,6 +1,7 @@
 package processor_test
 
 import (
+	"fmt"
 	"slices"
 	"testing"
 
@@ -113,3 +114,27 @@ func TestBadJsonEncode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkJsonEncode(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["json.encode"]
+	if !ok {
+		b.Fatalf("json.encode processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "json.encode",
+	})
+
+	if err != nil {
+		b.Fatalf("json.encode failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: fmt.Sprintf("{\"key\":%d}", count)})
+		if err != nil {
+			b.Fatalf("json.encode processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/kv-get_test.go

@@ -58,21 +58,10 @@ func TestGoodKvGet(t *testing.T) {
 		expected any
 	}{
 		{
-			name: "basic value",
+			name: "basic key",
 			params: map[string]any{
 				"module": "test",
 				"key":    "test",
-				"value":  "hello",
-			},
-			payload:  "hello",
-			expected: "test",
-		},
-		{
-			name: "template value",
-			params: map[string]any{
-				"module": "test",
-				"key":    "test",
-				"value":  "{{.Payload}}",
 			},
 			payload:  "hello",
 			expected: "test",
@@ -232,3 +221,32 @@ func TestBadKvGet(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkKvGet(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["kv.get"]
+	if !ok {
+		b.Fatalf("kv.get processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "kv.get",
+		Params: map[string]any{
+			"module": "test",
+			"key":    "test",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("kv.get failed to create processor: %s", err)
+	}
+	modules := map[string]common.Module{
+		"test": test.NewTestKVModule("test"),
+	}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: nil, Modules: modules})
+		if err != nil {
+			b.Fatalf("kv.get processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/kv-set_test.go

@@ -281,3 +281,35 @@ func TestBadKvSet(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkKvSet(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["kv.set"]
+	if !ok {
+		b.Fatalf("kv.set processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "kv.set",
+		Params: map[string]any{
+			"module": "test",
+			"key":    "test",
+			"value":  "{{.Payload}}",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("kv.set failed to create processor: %s", err)
+	}
+	modules := map[string]common.Module{
+		"test": test.NewTestKVModule("test"),
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count, Modules: modules})
+		if err != nil {
+			b.Fatalf("kv.set processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/midi-control_change-create_test.go

@@ -159,3 +159,32 @@ func TestBadMIDIControlChangeCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMIDIControlChangeCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["midi.control_change.create"]
+	if !ok {
+		b.Fatalf("midi.control_change.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "midi.control_change.create",
+		Params: map[string]any{
+			"channel": "1",
+			"control": "64",
+			"value":   "127",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("midi.control_change.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("midi.control_change.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/midi-message-decode_test.go

@@ -91,3 +91,15 @@ func TestBadMIDIMessageDecode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMIDIMessageDecode(b *testing.B) {
+	processorInstance := processor.MIDIMessageDecode{}
+	payload := []byte{0x90, 0x40, 0x7F}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: payload})
+		if err != nil {
+			b.Fatalf("midi.message.decode processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/midi-message-encode_test.go

@@ -89,3 +89,15 @@ func TestBadMIDIMessageEncode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMIDIMessageEncode(b *testing.B) {
+	processorInstance := processor.MIDIMessageEncode{}
+	payload := midi.NoteOn(1, 60, 127)
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: payload})
+		if err != nil {
+			b.Fatalf("midi.message.encode processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/midi-message-unpack_test.go

@@ -150,3 +150,15 @@ func TestBadMIDIMessageUnpack(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMIDIMessageUnpack(b *testing.B) {
+	processorInstance := processor.MIDIMessageUnpack{}
+	payload := midi.NoteOn(1, 60, 127)
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: payload})
+		if err != nil {
+			b.Fatalf("midi.message.unpack processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/midi-note_off-create_test.go

@@ -158,3 +158,32 @@ func TestBadMIDINoteOffCretea(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMIDINoteOffCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["midi.note_off.create"]
+	if !ok {
+		b.Fatalf("midi.note_off.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "midi.note_off.create",
+		Params: map[string]any{
+			"channel":  "1",
+			"note":     "60",
+			"velocity": "100",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("midi.note_off.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("midi.note_off.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/midi-note_on-create_test.go

@@ -155,3 +155,32 @@ func TestBadMIDINoteOnCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMIDINoteOnCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["midi.note_on.create"]
+	if !ok {
+		b.Fatalf("midi.note_on.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "midi.note_on.create",
+		Params: map[string]any{
+			"channel":  "1",
+			"note":     "60",
+			"velocity": "100",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("midi.note_on.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("midi.note_off.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/midi-program_change-create_test.go

@@ -145,3 +145,31 @@ func TestBadMIDIProgramChangeCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMIDIProgramChangeCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["midi.program_change.create"]
+	if !ok {
+		b.Fatalf("midi.program_change.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "midi.program_change.create",
+		Params: map[string]any{
+			"channel": "1",
+			"program": "60",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("midi.program_change.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("midi.program_change.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/mqtt-message-create_test.go

@@ -240,3 +240,33 @@ func TestBadMQTTMessageCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMQTTMessageCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["mqtt.message.create"]
+	if !ok {
+		b.Fatalf("mqtt.message.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "mqtt.message.create",
+		Params: map[string]any{
+			"topic":    "test/topic",
+			"qos":      1,
+			"retained": true,
+			"payload":  "{{.Payload}}",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("mqtt.message.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("mqtt.message.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/nats-message-create_test.go

@@ -208,3 +208,31 @@ func TestBadNATSMessageCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkNATSMessageCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["nats.message.create"]
+	if !ok {
+		b.Fatalf("nats.message.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "nats.message.create",
+		Params: map[string]any{
+			"subject": "test.subject",
+			"payload": "{{.Payload}}",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("nats.message.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("nats.message.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/osc-message-create_test.go

@@ -400,3 +400,32 @@ func TestBadOSCMessageCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkOSCMessageCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["osc.message.create"]
+	if !ok {
+		b.Fatalf("osc.message.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "osc.message.create",
+		Params: map[string]any{
+			"address": "/hello",
+			"args":    []any{"{{.Payload}}"},
+			"types":   "i",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("osc.message.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("osc.message.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/osc-message-decode_test.go

@@ -121,3 +121,15 @@ func TestBadOSCMessageDecode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkOSCMessageDecode(b *testing.B) {
+	processorInstance := processor.OSCMessageDecode{}
+	payload := []byte{47, 116, 101, 115, 116, 0, 0, 0, 44, 105, 0, 0, 0, 0, 0, 42}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: payload})
+		if err != nil {
+			b.Fatalf("osc.message.decode processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/osc-message-encode_test.go

@@ -112,3 +112,23 @@ func TestBadOSCMessageEncode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkOSCMessageEncode(b *testing.B) {
+	processorInstance := processor.OSCMessageEncode{}
+	payload := &osc.OSCMessage{
+		Address: "/test",
+		Args: []osc.OSCArg{
+			{
+				Type:  "i",
+				Value: int32(42),
+			},
+		},
+	}
+
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: payload})
+		if err != nil {
+			b.Fatalf("osc.message.encode processing failed: %s", err)
+		}
+	}
+}

internal/processor/test/script-expr_test.go

@@ -146,3 +146,30 @@ func TestBadScriptExpr(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkScriptExpr(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["script.expr"]
+	if !ok {
+		b.Fatalf("script.expr processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "script.expr",
+		Params: map[string]any{
+			"expression": "Payload % 2",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("script.expr failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("script.expr processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/script-js_test.go

@@ -209,6 +209,13 @@ func TestBadScriptJS(t *testing.T) {
 				Params: test.params,
 			})
 
+			if err != nil {
+				if test.errorString != err.Error() {
+					t.Fatalf("script.js got error '%s', expected '%s'", err.Error(), test.errorString)
+				}
+				return
+			}
+
 			got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Payload: test.payload})
 
 			if err == nil {
@@ -221,3 +228,30 @@ func TestBadScriptJS(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkScriptJS(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["script.js"]
+	if !ok {
+		b.Fatalf("script.js processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "script.js",
+		Params: map[string]any{
+			"program": "payload = payload + 1",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("script.js failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("script.js processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/script-wasm_test.go

@@ -1,6 +1,7 @@
 package processor_test
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 
@@ -188,3 +189,30 @@ func TestBadScriptWASM(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkScriptWASM(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["script.wasm"]
+	if !ok {
+		b.Fatalf("script.wasm processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "script.wasm",
+		Params: map[string]any{
+			"path": "good.wasm",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("script.wasm failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: []byte(fmt.Sprintf("%d", count))})
+		if err != nil {
+			b.Fatalf("script.wasm processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/sip-response-audio-create_test.go

@@ -212,3 +212,32 @@ func TestBadSipResponseAudioCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkSipResponseAudioCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["sip.response.audio.create"]
+	if !ok {
+		b.Fatalf("sip.response.audio.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "sip.response.audio.create",
+		Params: map[string]any{
+			"preWait":   0,
+			"audioFile": "good.wav",
+			"postWait":  0,
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("sip.response.audio.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("sip.response.audio.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/sip-response-dtmf-create_test.go

@@ -220,3 +220,32 @@ func TestBadSipResponseDTMFCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkSipResponseDTMFCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["sip.response.dtmf.create"]
+	if !ok {
+		b.Fatalf("sip.response.dtmf.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "sip.response.dtmf.create",
+		Params: map[string]any{
+			"preWait":  0,
+			"digits":   "{{.Payload}}",
+			"postWait": 0,
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("sip.response.dtmf.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("sip.response.dtmf.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/string-create_test.go

@@ -186,3 +186,30 @@ func TestBadStringCreate(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkStringCreate(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["string.create"]
+	if !ok {
+		b.Fatalf("string.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "string.create",
+		Params: map[string]any{
+			"template": "{{.Payload}}",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("string.create failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count})
+		if err != nil {
+			b.Fatalf("string.create processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/string-decode_test.go

@@ -1,6 +1,7 @@
 package processor_test
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/jwetzell/showbridge-go/internal/common"
@@ -98,3 +99,27 @@ func TestBadStringDecode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkStringDecode(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["string.decode"]
+	if !ok {
+		b.Fatalf("string.decode processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "string.decode",
+	})
+
+	if err != nil {
+		b.Fatalf("string.decode failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: []byte(fmt.Sprintf("%d", count))})
+		if err != nil {
+			b.Fatalf("string.decode processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/string-encode_test.go

@@ -1,6 +1,7 @@
 package processor_test
 
 import (
+	"fmt"
 	"slices"
 	"testing"
 
@@ -104,3 +105,27 @@ func TestBadStringEncode(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkStringEncode(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["string.encode"]
+	if !ok {
+		b.Fatalf("string.encode processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "string.encode",
+	})
+
+	if err != nil {
+		b.Fatalf("string.encode failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: fmt.Sprintf("%d", count)})
+		if err != nil {
+			b.Fatalf("string.encode processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/string-split_test.go

@@ -1,6 +1,7 @@
 package processor_test
 
 import (
+	"fmt"
 	"slices"
 	"testing"
 
@@ -153,3 +154,30 @@ func TestBadStringSplit(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkStringSplit(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["string.split"]
+	if !ok {
+		b.Fatalf("string.split processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "string.split",
+		Params: map[string]any{
+			"separator": ",",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("string.split failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: fmt.Sprintf("%d,%d,%d", count, count, count)})
+		if err != nil {
+			b.Fatalf("string.split processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/struct-field-get_test.go

@@ -191,3 +191,30 @@ func TestBadStructFieldGet(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkStructFieldGet(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["struct.field.get"]
+	if !ok {
+		b.Fatalf("struct.field.get processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "struct.field.get",
+		Params: map[string]any{
+			"name": "Data",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("struct.field.get failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: test.TestStruct{Data: count}})
+		if err != nil {
+			b.Fatalf("struct.field.get processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/processor/test/struct-method-get_test.go

@@ -216,3 +216,30 @@ func TestBadStructMethodGet(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkStructMethodGet(b *testing.B) {
+	registration, ok := processor.ProcessorRegistry["struct.method.get"]
+	if !ok {
+		b.Fatalf("struct.method.get processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "struct.method.get",
+		Params: map[string]any{
+			"name": "GetData",
+		},
+	})
+
+	if err != nil {
+		b.Fatalf("struct.method.get failed to create processor: %s", err)
+	}
+
+	count := 0
+	for b.Loop() {
+		_, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: test.TestStruct{Data: count}})
+		if err != nil {
+			b.Fatalf("struct.method.get processing failed: %s", err)
+		}
+		count++
+	}
+}

internal/test/module.go

@@ -79,11 +79,14 @@ func (m *TestDBModule) Start(ctx context.Context, router common.RouteIO) error {
 	return nil
 }
 
-func (m *TestDBModule) Database() *sql.DB {
+func (m *TestDBModule) Database() (*sql.DB, error) {
 	if m.db == nil {
-		db, _ := sql.Open("sqlite", ":memory:")
+		db, err := sql.Open("sqlite", ":memory:")
+		if err != nil {
+			return nil, err
+		}
 
-		db.Exec(`
+		_, err = db.Exec(`
 		CREATE TABLE test (
 			id INTEGER PRIMARY KEY,
 			value TEXT
@@ -91,9 +94,12 @@ func (m *TestDBModule) Database() *sql.DB {
 		INSERT INTO test (id, value) VALUES (1, 'test-1'), (2, 'test-2');
 		
 		`)
+		if err != nil {
+			return nil, err
+		}
 		m.db = db
 	}
-	return m.db
+	return m.db, nil
 }
 
 func (m *TestDBModule) Stop() {}

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)