Merge pull request #52 from jwetzell/fix/debug-log-payload-string

more compatible printing of payload in debug.log

.github/labeler.yml

@@ -14,10 +14,14 @@ processor:
   - changed-files:
       - any-glob-to-any-file: 'internal/processor/**'
 
+router:
+  - changed-files:
+      - any-glob-to-any-file: 'router*'
+      
 route:
   - changed-files:
       - any-glob-to-any-file: 'internal/route/**'
-
+      
 cli:
   - changed-files:
       - any-glob-to-any-file: 'cmd/showbridge/**'

.github/release.yml

@@ -17,12 +17,13 @@ changelog:
     - title: Processor 🏭
       labels:
         - processor
-    - title: Route 🛣️
+    - title: Routing 🛣️
       labels:
+        - router
         - route
     - title: CLI ⌨️
       labels:
-        - cmd
+        - cli
     - title: Other Changes
       labels:
         - '*'

.github/workflows/release-showbridge.yaml

@@ -40,10 +40,10 @@ jobs:
         uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v3.7.0
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v3.11.1
+        uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3.12.0
 
       - name: Login to Docker Hub
-        uses: docker/login-action@5e57cd118135c172c3672efd75eb46360885c0ef # v3.6.0
+        uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_TOKEN }}

.vscode/launch.json

@@ -11,7 +11,7 @@
             "request": "launch",
             "mode": "auto",
             "program": "cmd/showbridge",
-            "args": ["--debug"],
+            "args": ["--log-level", "debug"],
             "cwd": "./"
         }
     ]

README.md

@@ -34,9 +34,9 @@ USAGE:
    showbridge [global options]
 
 GLOBAL OPTIONS:
-   --config string  path to config file (default: "./config.yaml")
+   --config string      path to config file (default: "./config.yaml")
-   --debug          set log level to DEBUG
+   --log-level string   set log level (default: "info")
-   --json           log using JSON
+   --log-format string  log format to use (default: "text")
-   --help, -h       show help
+   --help, -h           show help
-   --version, -v    print the version
+   --version, -v        print the version
 ```

cmd/showbridge/main.go

@@ -3,9 +3,11 @@ package main
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"os/signal"
+	"slices"
 	"sync"
 
 	"github.com/jwetzell/showbridge-go"
@@ -29,15 +31,29 @@ func main() {
 				Value: "./config.yaml",
 				Usage: "path to config file",
 			},
-			&cli.BoolFlag{
+			&cli.StringFlag{
-				Name:  "debug",
+				Name:  "log-level",
-				Value: false,
+				Value: "info",
-				Usage: "set log level to DEBUG",
+				Usage: "set log level",
+				Validator: func(level string) error {
+					levels := []string{"debug", "info", "warn", "error"}
+					if !slices.Contains(levels, level) {
+						return fmt.Errorf("unknown log level: %s", level)
+					}
+					return nil
+				},
 			},
-			&cli.BoolFlag{
+			&cli.StringFlag{
-				Name:  "json",
+				Name:  "log-format",
-				Value: false,
+				Value: "text",
-				Usage: "log using JSON",
+				Usage: "log format to use",
+				Validator: func(format string) error {
+					formats := []string{"text", "json"}
+					if !slices.Contains(formats, format) {
+						return fmt.Errorf("unknown log format: %s", format)
+					}
+					return nil
+				},
 			},
 		},
 		Action: run,
@@ -83,8 +99,19 @@ func run(ctx context.Context, c *cli.Command) error {
 
 	logLevel := slog.LevelInfo
 
-	if c.Bool("debug") {
+	logLevelFromFlag := c.String("log-level")
+
+	switch logLevelFromFlag {
+	case "debug":
 		logLevel = slog.LevelDebug
+	case "info":
+		logLevel = slog.LevelInfo
+	case "warn":
+		logLevel = slog.LevelWarn
+	case "error":
+		logLevel = slog.LevelError
+	default:
+		logLevel = slog.LevelInfo
 	}
 
 	logHandlerOptions := &slog.HandlerOptions{
@@ -93,17 +120,24 @@ func run(ctx context.Context, c *cli.Command) error {
 
 	logOutput := os.Stderr
 
-	var logHandler slog.Handler = slog.NewTextHandler(logOutput, logHandlerOptions)
+	var logHandler slog.Handler
 
-	if c.Bool("json") {
+	logFormat := c.String("log-format")
+
+	switch logFormat {
+	case "json":
 		logHandler = slog.NewJSONHandler(logOutput, logHandlerOptions)
+	case "text":
+		logHandler = slog.NewTextHandler(logOutput, logHandlerOptions)
+	default:
+		logHandler = slog.NewTextHandler(logOutput, logHandlerOptions)
 	}
 
 	slog.SetDefault(slog.New(logHandler))
 
 	commandLogger := slog.Default().With("component", "cmd")
 
-	router, moduleErrors, routeErrors := showbridge.NewRouter(context.Background(), config)
+	router, moduleErrors, routeErrors := showbridge.NewRouter(config)
 
 	for _, moduleError := range moduleErrors {
 		commandLogger.Error("problem initializing module", "index", moduleError.Index, "error", moduleError.Error)
@@ -116,7 +150,7 @@ func run(ctx context.Context, c *cli.Command) error {
 	routerRunner := sync.WaitGroup{}
 
 	routerRunner.Go(func() {
-		router.Run()
+		router.Run(context.Background())
 	})
 
 	<-ctx.Done()

go.mod

@@ -4,53 +4,63 @@ go 1.25.5
 
 require (
 	github.com/eclipse/paho.mqtt.golang v1.5.1
-	github.com/emiago/diago v0.23.1-0.20251211215055-e1d875617111
+	github.com/emiago/diago v0.26.2
-	github.com/emiago/sipgo v1.0.1
+	github.com/emiago/sipgo v1.1.2
 	github.com/expr-lang/expr v1.17.7
-	github.com/jwetzell/artnet-go v0.0.0-20251223201031-5097901aa9db
+	github.com/extism/go-sdk v1.7.1
+	github.com/jwetzell/artnet-go v0.2.1
 	github.com/jwetzell/free-d-go v0.1.0
 	github.com/jwetzell/osc-go v0.1.0
 	github.com/jwetzell/psn-go v0.3.0
 	github.com/nats-io/nats.go v1.48.0
-	github.com/urfave/cli/v3 v3.6.1
+	github.com/urfave/cli/v3 v3.6.2
 	gitlab.com/gomidi/midi/v2 v2.3.18
 	go.bug.st/serial v1.6.4
-	modernc.org/quickjs v0.17.0
+	modernc.org/quickjs v0.17.1
 	sigs.k8s.io/yaml v1.6.0
 )
 
 require (
 	github.com/creack/goselect v0.1.2 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/dylibso/observe-sdk/go v0.0.0-20240819160327-2d926c5d788a // indirect
+	github.com/emiago/dtls/v3 v3.0.0-20260122183559-8b8d23e359c0 // indirect
 	github.com/go-audio/riff v1.0.0 // indirect
+	github.com/gobwas/glob v0.2.3 // indirect
 	github.com/gobwas/httphead v0.1.0 // indirect
 	github.com/gobwas/pool v0.2.1 // indirect
 	github.com/gobwas/ws v1.4.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/websocket v1.5.3 // indirect
+	github.com/ianlancetaylor/demangle v0.0.0-20240805132620-81f5be970eca // indirect
 	github.com/icholy/digest v1.1.0 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/nats-io/nkeys v0.4.11 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
-	github.com/ncruces/go-strftime v0.1.9 // indirect
+	github.com/ncruces/go-strftime v1.0.0 // indirect
 	github.com/pion/logging v0.2.4 // indirect
 	github.com/pion/randutil v0.1.0 // indirect
 	github.com/pion/rtcp v1.2.16 // indirect
 	github.com/pion/rtp v1.8.26 // indirect
 	github.com/pion/srtp/v3 v3.0.9 // indirect
 	github.com/pion/transport/v3 v3.1.1 // indirect
+	github.com/pion/transport/v4 v4.0.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
+	github.com/tetratelabs/wabin v0.0.0-20230304001439-f6f874872834 // indirect
+	github.com/tetratelabs/wazero v1.9.0 // indirect
 	github.com/zaf/g711 v1.4.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	golang.org/x/crypto v0.42.0 // indirect
-	golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b // indirect
+	golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
 	golang.org/x/net v0.44.0 // indirect
 	golang.org/x/sync v0.19.0 // indirect
 	golang.org/x/sys v0.39.0 // indirect
+	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/hraban/opus.v2 v2.0.0-20230925203106-0188a62cb302 // indirect
-	modernc.org/libc v1.66.10 // indirect
+	modernc.org/libc v1.67.1 // indirect
-	modernc.org/libquickjs v0.12.2 // indirect
+	modernc.org/libquickjs v0.12.3 // indirect
 	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.11.0 // indirect
 )

go.sum

@@ -4,16 +4,24 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/dylibso/observe-sdk/go v0.0.0-20240819160327-2d926c5d788a h1:UwSIFv5g5lIvbGgtf3tVwC7Ky9rmMFBp0RMs+6f6YqE=
+github.com/dylibso/observe-sdk/go v0.0.0-20240819160327-2d926c5d788a/go.mod h1:C8DzXehI4zAbrdlbtOByKX6pfivJTBiV9Jjqv56Yd9Q=
 github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
 github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
-github.com/emiago/diago v0.23.1-0.20251211215055-e1d875617111 h1:jqhOZbH40pf2jiUhGaBNk334wOtNYvAaXg/mHOXhy/Y=
+github.com/emiago/diago v0.26.2 h1:3QL03V0drX96eIBFBpfueNcywydRgYqffKihluGL0gA=
-github.com/emiago/diago v0.23.1-0.20251211215055-e1d875617111/go.mod h1:3vLCCq8/G/Ei5I64IHtrmBTag+nPLcgXcKeN1KkLtuc=
+github.com/emiago/diago v0.26.2/go.mod h1:jZ+7EnKcmgqKnLjCHPqfbP4Y/9Q/JLSLxMflDrp2J1M=
-github.com/emiago/sipgo v1.0.1 h1:8eCZ6L/VX3isyByyv1RrBoQ5GyBoRXBHkNMYjwacRfk=
+github.com/emiago/dtls/v3 v3.0.0-20260122183559-8b8d23e359c0 h1:o4LxpUnZ1zxiQ+Qjc9kLwXcjz31NGAHmnZ7xoJto3VM=
-github.com/emiago/sipgo v1.0.1/go.mod h1:DuwAxBZhKMqIzQFPGZb1MVAGU6Wuxj64oTOhd5dx/FY=
+github.com/emiago/dtls/v3 v3.0.0-20260122183559-8b8d23e359c0/go.mod h1:ydcZ977eS1I6uOWodzMuw30BwvNAzT9su/xcNYSJqjA=
+github.com/emiago/sipgo v1.1.2 h1:JvLqEvqNSQm2mBX40qZ7O0WC3Ee67Z0UrfmBI7y6Beo=
+github.com/emiago/sipgo v1.1.2/go.mod h1:DuwAxBZhKMqIzQFPGZb1MVAGU6Wuxj64oTOhd5dx/FY=
 github.com/expr-lang/expr v1.17.7 h1:Q0xY/e/2aCIp8g9s/LGvMDCC5PxYlvHgDZRQ4y16JX8=
 github.com/expr-lang/expr v1.17.7/go.mod h1:8/vRC7+7HBzESEqt5kKpYXxrxkr31SaO8r40VO/1IT4=
+github.com/extism/go-sdk v1.7.1 h1:lWJos6uY+tRFdlIHR+SJjwFDApY7OypS/2nMhiVQ9Sw=
+github.com/extism/go-sdk v1.7.1/go.mod h1:IT+Xdg5AZM9hVtpFUA+uZCJMge/hbvshl8bwzLtFyKA=
 github.com/go-audio/riff v1.0.0 h1:d8iCGbDvox9BfLagY94fBynxSPHO80LmZCaOsmKxokA=
 github.com/go-audio/riff v1.0.0/go.mod h1:l3cQwc85y79NQFCRB7TiPoNiaijp6q8Z0Uv38rVG498=
+github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y=
+github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
 github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU=
 github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM=
 github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og=
@@ -28,10 +36,12 @@ github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aN
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
 github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
+github.com/ianlancetaylor/demangle v0.0.0-20240805132620-81f5be970eca h1:T54Ema1DU8ngI+aef9ZhAhNGQhcRTrWxVeG07F+c/Rw=
+github.com/ianlancetaylor/demangle v0.0.0-20240805132620-81f5be970eca/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.0.0-20251223201031-5097901aa9db h1:PR9sdDEWBufi8/3yiQHRXncLNKUHBMKiEBDtdNWsK/E=
+github.com/jwetzell/artnet-go v0.2.1 h1:iYTKWcwYrF5kBkYfkw2UbWvoueeA23iKEn7fR27mWZE=
-github.com/jwetzell/artnet-go v0.0.0-20251223201031-5097901aa9db/go.mod h1:gli97Z32a0kMkZ6taoTiK7/lqHcF/dhiGjGJdx/PxqA=
+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/osc-go v0.1.0 h1:EXxup5VWBErHot2Ri4MFToPf6KCzLDTbCt2x6GLfw8I=
@@ -52,8 +62,8 @@ github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0=
 github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
+github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
-github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
+github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
 github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8=
 github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so=
 github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
@@ -66,6 +76,8 @@ github.com/pion/srtp/v3 v3.0.9 h1:lRGF4G61xxj+m/YluB3ZnBpiALSri2lTzba0kGZMrQY=
 github.com/pion/srtp/v3 v3.0.9/go.mod h1:E+AuWd7Ug2Fp5u38MKnhduvpVkveXJX6J4Lq4rxUYt8=
 github.com/pion/transport/v3 v3.1.1 h1:Tr684+fnnKlhPceU+ICdrw6KKkTms+5qHMgw6bIkYOM=
 github.com/pion/transport/v3 v3.1.1/go.mod h1:+c2eewC5WJQHiAA46fkMMzoYZSuGzA/7E2FPrOYHctQ=
+github.com/pion/transport/v4 v4.0.1 h1:sdROELU6BZ63Ab7FrOLn13M6YdJLY20wldXW2Cu2k8o=
+github.com/pion/transport/v4 v4.0.1/go.mod h1:nEuEA4AD5lPdcIegQDpVLgNoDGreqM/YqmEx3ovP4jM=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
@@ -74,24 +86,30 @@ github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR
 github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
-github.com/urfave/cli/v3 v3.6.1 h1:j8Qq8NyUawj/7rTYdBGrxcH7A/j7/G8Q5LhWEW4G3Mo=
+github.com/tetratelabs/wabin v0.0.0-20230304001439-f6f874872834 h1:ZF+QBjOI+tILZjBaFj3HgFonKXUcwgJ4djLb6i42S3Q=
-github.com/urfave/cli/v3 v3.6.1/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso=
+github.com/tetratelabs/wabin v0.0.0-20230304001439-f6f874872834/go.mod h1:m9ymHTgNSEjuxvw8E7WWe4Pl4hZQHXONY8wE6dMLaRk=
+github.com/tetratelabs/wazero v1.9.0 h1:IcZ56OuxrtaEz8UYNRHBrUa9bYeX9oVY93KspZZBf/I=
+github.com/tetratelabs/wazero v1.9.0/go.mod h1:TSbcXCfFP0L2FGkRPxHphadXPjo1T6W+CseNNY7EkjM=
+github.com/urfave/cli/v3 v3.6.2 h1:lQuqiPrZ1cIz8hz+HcrG0TNZFxU70dPZ3Yl+pSrH9A8=
+github.com/urfave/cli/v3 v3.6.2/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso=
 github.com/zaf/g711 v1.4.0 h1:XZYkjjiAg9QTBnHqEg37m2I9q3IIDv5JRYXs2N8ma7c=
 github.com/zaf/g711 v1.4.0/go.mod h1:eCDXt3dSp/kYYAoooba7ukD/Q75jvAaS4WOMr0l1Roo=
 gitlab.com/gomidi/midi/v2 v2.3.18 h1:sj2fOhtvOe+zI8YJe8qTxLw5zv0ntULLUDwcFOaZQbI=
 gitlab.com/gomidi/midi/v2 v2.3.18/go.mod h1:jDpP4O4skYi+7iVwt6Zyp18bd2M4hkjtMuw2cmgKgfw=
 go.bug.st/serial v1.6.4 h1:7FmqNPgVp3pu2Jz5PoPtbZ9jJO5gnEnZIvnI1lzve8A=
 go.bug.st/serial v1.6.4/go.mod h1:nofMJxTeNVny/m6+KaafC6vJGj3miwQZ6vW4BZUGJPI=
+go.opentelemetry.io/proto/otlp v1.3.1 h1:TrMUixzpM0yuc/znrFTP9MMRh8trP93mkCiDVeXrui0=
+go.opentelemetry.io/proto/otlp v1.3.1/go.mod h1:0X1WI4de4ZsLrrJNLAQbFeLCm3T7yBkR0XqQ7niQU+8=
 go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
 go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
 go.yaml.in/yaml/v3 v3.0.3 h1:bXOww4E/J3f66rav3pX3m8w6jDE4knZjGOw8b5Y6iNE=
 go.yaml.in/yaml/v3 v3.0.3/go.mod h1:tBHosrYAkRZjRAOREWbDnBXUf08JOwYq++0QNwQiWzI=
 golang.org/x/crypto v0.42.0 h1:chiH31gIWm57EkTXpwnqf8qeuMUi0yekh6mT2AvFlqI=
 golang.org/x/crypto v0.42.0/go.mod h1:4+rDnOTJhQCx2q7/j6rAN5XDw8kPjeaXEUR2eL94ix8=
-golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b h1:M2rDM6z3Fhozi9O7NWsxAkg/yqS/lQJ6PmkyIV3YP+o=
+golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 h1:mgKeJMpvi0yx/sU5GsxQ7p6s2wtOnGAHZWCHUM4KGzY=
-golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b/go.mod h1:3//PLf8L/X+8b4vuAfHzxeRUl04Adcb341+IGKfnqS8=
+golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546/go.mod h1:j/pmGrbnkbPtQfxEe5D0VQhZC6qKbfKifgD0oM7sR70=
-golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA=
+golang.org/x/mod v0.30.0 h1:fDEXFVZ/fmCKProc/yAXXUijritrDzahmwwefnjoPFk=
-golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w=
+golang.org/x/mod v0.30.0/go.mod h1:lAsf5O2EvJeSFMiBxXDki7sCgAxEUcZHXoXMKT4GJKc=
 golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I=
 golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY=
 golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
@@ -99,8 +117,10 @@ golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk=
 golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
-golang.org/x/tools v0.37.0 h1:DVSRzp7FwePZW356yEAChSdNcQo6Nsp+fex1SUW09lE=
+golang.org/x/tools v0.38.0 h1:Hx2Xv8hISq8Lm16jvBZ2VQf+RLmbd7wVUsALibYI/IQ=
-golang.org/x/tools v0.37.0/go.mod h1:MBN5QPQtLMHVdvsbtarmTNukZDdgwdwlO5qGacAzF0w=
+golang.org/x/tools v0.38.0/go.mod h1:yEsQ/d/YK8cjh0L6rZlY8tgtlKiBNTL14pGDJPJpYQs=
+google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
+google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
@@ -122,18 +142,18 @@ modernc.org/gc/v3 v3.1.1 h1:k8T3gkXWY9sEiytKhcgyiZ2L0DTyCQ/nvX+LoCljoRE=
 modernc.org/gc/v3 v3.1.1/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY=
 modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks=
 modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI=
-modernc.org/libc v1.66.10 h1:yZkb3YeLx4oynyR+iUsXsybsX4Ubx7MQlSYEw4yj59A=
+modernc.org/libc v1.67.1 h1:bFaqOaa5/zbWYJo8aW0tXPX21hXsngG2M7mckCnFSVk=
-modernc.org/libc v1.66.10/go.mod h1:8vGSEwvoUoltr4dlywvHqjtAqHBaw0j1jI7iFBTAr2I=
+modernc.org/libc v1.67.1/go.mod h1:QvvnnJ5P7aitu0ReNpVIEyesuhmDLQ8kaEoyMjIFZJA=
-modernc.org/libquickjs v0.12.2 h1:XtF6iD+aoN/pEz1MAQjg1wMZT412pzmOlO11UGLW/wQ=
+modernc.org/libquickjs v0.12.3 h1:2IU9B6njBmce2PuYttJDkXeoLRV9WnvgP+eU5HAC8YI=
-modernc.org/libquickjs v0.12.2/go.mod h1:n+vyuJ4mXh1pQYt1bJttvoCE+2t+MqCM8BFlcSDg/70=
+modernc.org/libquickjs v0.12.3/go.mod h1:iCsgVxnHTX3i0YPxxHBmJk0GLA5sVUHXWI/090UXgeE=
 modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
 modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
 modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
 modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
 modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8=
 modernc.org/opt v0.1.4/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
-modernc.org/quickjs v0.17.0 h1:gYZRSCKVokyOiHhZEKwran/NoBaGjAaWk+xJq52SWfQ=
+modernc.org/quickjs v0.17.1 h1:CbYnbTf7ksZk9YZ1rRM2Ab1Zfi+X6s50kXiOhpd2NIg=
-modernc.org/quickjs v0.17.0/go.mod h1:TE+wpAYX4V7Nvi/sHRUrd9l5L176ZIiS0as3MMObvC0=
+modernc.org/quickjs v0.17.1/go.mod h1:hATT7DIJc33I5Q/Fjffhm0tpUHNSqdKHma/ossibTA0=
 modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
 modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
 modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=

internal/module/http-client.go

@@ -22,15 +22,9 @@ type HTTPClient struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "http.client",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &HTTPClient{config: config, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("http.client unable to get router from context")
-			}
-
-			return &HTTPClient{config: config, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -43,7 +37,14 @@ func (hc *HTTPClient) Type() string {
 	return hc.config.Type
 }
 
-func (hc *HTTPClient) Run() error {
+func (hc *HTTPClient) Run(ctx context.Context) error {
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("http.client unable to get router from context")
+	}
+	hc.router = router
+	hc.ctx = ctx
 
 	hc.client = &http.Client{
 		Timeout: 10 * time.Second,
@@ -73,7 +74,7 @@ func (hc *HTTPClient) Output(ctx context.Context, payload any) error {
 	}
 
 	if hc.router != nil {
-		hc.router.HandleInput(hc.Id(), response)
+		hc.router.HandleInput(hc.ctx, hc.Id(), response)
 	}
 
 	return nil

internal/module/http-server.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 
 	"github.com/jwetzell/showbridge-go/internal/config"
+	"github.com/jwetzell/showbridge-go/internal/processor"
 	"github.com/jwetzell/showbridge-go/internal/route"
 )
 
@@ -20,15 +21,40 @@ type HTTPServer struct {
 	logger *slog.Logger
 }
 
-type ResponseData struct {
+type ResponseIOError struct {
-	Message string `json:"message"`
+	Index        int      `json:"index"`
-	Status  string `json:"status"`
+	OutputErrors []string `json:"outputErrors"`
+	ProcessError *string  `json:"processError"`
+	InputError   *string  `json:"inputError"`
+}
+
+type IOResponseData struct {
+	IOErrors []ResponseIOError `json:"ioErrors"`
+	Message  string            `json:"message"`
+	Status   string            `json:"status"`
+}
+
+type httpServerContextKey string
+
+type HTTPServerResponseWriter struct {
+	http.ResponseWriter
+	done bool
+}
+
+func (hsrw *HTTPServerResponseWriter) WriteHeader(status int) {
+	hsrw.done = true
+	hsrw.ResponseWriter.WriteHeader(status)
+}
+
+func (hsrw *HTTPServerResponseWriter) Write(data []byte) (int, error) {
+	hsrw.done = true
+	return hsrw.ResponseWriter.Write(data)
 }
 
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "http.server",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			port, ok := params["port"]
 			if !ok {
@@ -41,13 +67,7 @@ func init() {
 				return nil, errors.New("http.server port must be uint16")
 			}
 
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &HTTPServer{Port: uint16(portNum), config: config, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("http.server unable to get router from context")
-			}
-
-			return &HTTPServer{Port: uint16(portNum), config: config, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -61,33 +81,85 @@ func (hs *HTTPServer) Type() string {
 }
 
 func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	w.Header().Set("Content-Type", "application/json")
+	responseWriter := HTTPServerResponseWriter{ResponseWriter: w}
 
-	response := ResponseData{
+	response := IOResponseData{
 		Message: "routing successful",
 		Status:  "ok",
 	}
-
 	if hs.router != nil {
-		routingErrors := hs.router.HandleInput(hs.Id(), r)
+		inputContext := context.WithValue(hs.ctx, httpServerContextKey("responseWriter"), &responseWriter)
-		if routingErrors != nil {
+		aRouteFound, routingErrors := hs.router.HandleInput(inputContext, hs.Id(), r)
-			w.WriteHeader(http.StatusInternalServerError)
+		if !responseWriter.done {
-			response.Status = "error"
+			if aRouteFound {
-			response.Message = "routing failed"
+				if routingErrors != nil {
-		} else {
+					w.WriteHeader(http.StatusInternalServerError)
-			w.WriteHeader(http.StatusOK)
+					response.Status = "error"
-			response.Message = "routing successful"
+					response.Message = "routing failed"
+
+					response.IOErrors = []ResponseIOError{}
+					for _, responseIOError := range routingErrors {
+						errorToAdd := ResponseIOError{
+							Index: responseIOError.Index,
+						}
+
+						if responseIOError.InputError != nil {
+							errorMsg := responseIOError.InputError.Error()
+							errorToAdd.InputError = &errorMsg
+						}
+
+						if responseIOError.ProcessError != nil {
+							errorMsg := responseIOError.ProcessError.Error()
+							errorToAdd.ProcessError = &errorMsg
+						}
+
+						if responseIOError.OutputErrors != nil {
+							outputErrorMsgs := []string{}
+
+							for _, outputError := range responseIOError.OutputErrors {
+								outputErrorMsgs = append(outputErrorMsgs, outputError.Error())
+							}
+
+							errorToAdd.OutputErrors = outputErrorMsgs
+						}
+
+						response.IOErrors = append(response.IOErrors, errorToAdd)
+
+					}
+					json.NewEncoder(w).Encode(response)
+					return
+				} else {
+					w.WriteHeader(http.StatusOK)
+					response.Message = "routing successful"
+					json.NewEncoder(w).Encode(response)
+					return
+				}
+			} else {
+				w.WriteHeader(http.StatusNotFound)
+				response.Status = "error"
+				response.Message = "no matching routes found"
+				json.NewEncoder(w).Encode(response)
+				return
+			}
 		}
 	} else {
 		w.WriteHeader(http.StatusInternalServerError)
 		response.Message = "no router registered"
 		response.Status = "error"
+		json.NewEncoder(w).Encode(response)
+		return
 	}
-
-	json.NewEncoder(w).Encode(response)
 }
 
-func (hs *HTTPServer) Run() error {
+func (hs *HTTPServer) Run(ctx context.Context) error {
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("http.server unable to get router from context")
+	}
+	hs.router = router
+	hs.ctx = ctx
+
 	httpServer := &http.Server{
 		Addr:    fmt.Sprintf(":%d", hs.Port),
 		Handler: hs,
@@ -112,5 +184,23 @@ func (hs *HTTPServer) Run() error {
 }
 
 func (hs *HTTPServer) Output(ctx context.Context, payload any) error {
-	return errors.New("http.server output is not implemented")
+	responseWriter, ok := ctx.Value(httpServerContextKey("responseWriter")).(*HTTPServerResponseWriter)
+
+	if !ok {
+		return errors.New("http.server output must originate from an http.server input")
+	}
+
+	payloadResponse, ok := payload.(processor.HTTPResponse)
+
+	if !ok {
+		return errors.New("http.server is only able to output HTTPResponse")
+	}
+
+	if responseWriter.done {
+		return errors.New("http.server response writer has already been written to")
+	}
+
+	responseWriter.WriteHeader(payloadResponse.Status)
+	responseWriter.Write(payloadResponse.Body)
+	return nil
 }

internal/module/midi-input.go

@@ -26,7 +26,7 @@ type MIDIInput struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "midi.input",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			port, ok := params["port"]
 
@@ -40,13 +40,7 @@ func init() {
 				return nil, errors.New("midi.input port must be a string")
 			}
 
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &MIDIInput{config: config, Port: portString, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("midi.input unable to get router from context")
-			}
-
-			return &MIDIInput{config: config, Port: portString, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -59,8 +53,15 @@ func (mi *MIDIInput) Type() string {
 	return mi.config.Type
 }
 
-func (mi *MIDIInput) Run() error {
+func (mi *MIDIInput) Run(ctx context.Context) error {
 	defer midi.CloseDriver()
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("midi.input unable to get router from context")
+	}
+	mi.router = router
+	mi.ctx = ctx
 
 	in, err := midi.FindInPort(mi.Port)
 	if err != nil {
@@ -69,7 +70,7 @@ func (mi *MIDIInput) Run() error {
 
 	stop, err := midi.ListenTo(in, func(msg midi.Message, timestampms int32) {
 		if mi.router != nil {
-			mi.router.HandleInput(mi.Id(), msg)
+			mi.router.HandleInput(mi.ctx, mi.Id(), msg)
 		}
 	}, midi.UseSysEx())
 

internal/module/midi-output.go

@@ -26,7 +26,7 @@ type MIDIOutput struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "midi.output",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 
 			port, ok := params["port"]
@@ -41,13 +41,7 @@ func init() {
 				return nil, errors.New("midi.output port must be a string")
 			}
 
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &MIDIOutput{config: config, Port: portString, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("midi.output unable to get router from context")
-			}
-
-			return &MIDIOutput{config: config, Port: portString, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -60,8 +54,15 @@ func (mo *MIDIOutput) Type() string {
 	return mo.config.Type
 }
 
-func (mo *MIDIOutput) Run() error {
+func (mo *MIDIOutput) Run(ctx context.Context) error {
 	defer midi.CloseDriver()
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("midi.output unable to get router from context")
+	}
+	mo.router = router
+	mo.ctx = ctx
 
 	out, err := midi.FindOutPort(mo.Port)
 

internal/module/module.go

@@ -18,13 +18,13 @@ type ModuleError struct {
 type Module interface {
 	Id() string
 	Type() string
-	Run() error
+	Run(context.Context) error
 	Output(context.Context, any) error
 }
 
 type ModuleRegistration struct {
 	Type string `json:"type"`
-	New  func(context.Context, config.ModuleConfig) (Module, error)
+	New  func(config.ModuleConfig) (Module, error)
 }
 
 func RegisterModule(mod ModuleRegistration) {

internal/module/mqtt-client.go

@@ -24,7 +24,7 @@ type MQTTClient struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "mqtt.client",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			broker, ok := params["broker"]
 
@@ -62,13 +62,7 @@ func init() {
 				return nil, errors.New("mqtt.client clientId must be string")
 			}
 
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &MQTTClient{config: config, Broker: brokerString, Topic: topicString, ClientID: clientIdString, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("mqtt.client unable to get router from context")
-			}
-
-			return &MQTTClient{config: config, Broker: brokerString, Topic: topicString, ClientID: clientIdString, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -81,7 +75,15 @@ func (mc *MQTTClient) Type() string {
 	return mc.config.Type
 }
 
-func (mc *MQTTClient) Run() error {
+func (mc *MQTTClient) Run(ctx context.Context) error {
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("mqtt.client unable to get router from context")
+	}
+	mc.router = router
+	mc.ctx = ctx
+
 	opts := mqtt.NewClientOptions()
 	opts.AddBroker(mc.Broker)
 	opts.SetClientID(mc.ClientID)
@@ -90,7 +92,7 @@ func (mc *MQTTClient) Run() error {
 
 	opts.OnConnect = func(c mqtt.Client) {
 		token := mc.client.Subscribe(mc.Topic, 1, func(c mqtt.Client, m mqtt.Message) {
-			mc.router.HandleInput(mc.Id(), m)
+			mc.router.HandleInput(mc.ctx, mc.Id(), m)
 		})
 		token.Wait()
 	}

internal/module/nats-client.go

@@ -24,7 +24,7 @@ type NATSClient struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "nats.client",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			url, ok := params["url"]
 
@@ -50,13 +50,7 @@ func init() {
 				return nil, errors.New("nats.client subject must be string")
 			}
 
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &NATSClient{config: config, URL: urlString, Subject: subjectString, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("nats.client unable to get router from context")
-			}
-
-			return &NATSClient{config: config, URL: urlString, Subject: subjectString, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -69,7 +63,16 @@ func (nc *NATSClient) Type() string {
 	return nc.config.Type
 }
 
-func (nc *NATSClient) Run() error {
+func (nc *NATSClient) Run(ctx context.Context) error {
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("nats.client unable to get router from context")
+	}
+
+	nc.router = router
+	nc.ctx = ctx
+
 	client, err := nats.Connect(nc.URL, nats.RetryOnFailedConnect(true))
 
 	if err != nil {
@@ -83,7 +86,7 @@ func (nc *NATSClient) Run() error {
 
 	sub, err := nc.client.Subscribe(nc.Subject, func(msg *nats.Msg) {
 		if nc.router != nil {
-			nc.router.HandleInput(nc.Id(), msg)
+			nc.router.HandleInput(nc.ctx, nc.Id(), msg)
 		}
 	})
 

internal/module/psn-client.go

@@ -25,13 +25,9 @@ type PSNClient struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "psn.client",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
 
-			if !ok {
+			return &PSNClient{config: config, decoder: psn.NewDecoder(), logger: CreateLogger(config)}, nil
-				return nil, errors.New("psn.client unable to get router from context")
-			}
-			return &PSNClient{config: config, decoder: psn.NewDecoder(), ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -44,7 +40,14 @@ func (pc *PSNClient) Type() string {
 	return pc.config.Type
 }
 
-func (pc *PSNClient) Run() error {
+func (pc *PSNClient) Run(ctx context.Context) error {
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("psn.client unable to get router from context")
+	}
+	pc.router = router
+	pc.ctx = ctx
 
 	addr, err := net.ResolveUDPAddr("udp", "236.10.10.10:56565")
 	if err != nil {
@@ -86,8 +89,9 @@ func (pc *PSNClient) Run() error {
 				}
 
 				if pc.router != nil {
+					// TODO(jwetzell): better input handling
 					for _, tracker := range pc.decoder.Trackers {
-						pc.router.HandleInput(pc.Id(), tracker)
+						pc.router.HandleInput(pc.ctx, pc.Id(), tracker)
 					}
 				} else {
 					pc.logger.Error("has no router")

internal/module/serial-client.go

@@ -29,7 +29,7 @@ type SerialClient struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "serial.client",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			port, ok := params["port"]
 
@@ -43,20 +43,19 @@ func init() {
 				return nil, errors.New("serial.client port must be a string")
 			}
 
-			framingMethod := "RAW"
+			framingMethod, ok := params["framing"]
 
-			framingMethodRaw, ok := params["framing"]
+			if !ok {
-
+				return nil, errors.New("serial.client requires a framing parameter")
-			if ok {
-				framingMethodString, ok := framingMethodRaw.(string)
-
-				if !ok {
-					return nil, errors.New("serial.client framing method must be a string")
-				}
-				framingMethod = framingMethodString
 			}
 
-			framer := framer.GetFramer(framingMethod)
+			framingMethodString, ok := framingMethod.(string)
+
+			if !ok {
+				return nil, errors.New("serial.client framing method must be a string")
+			}
+
+			framer := framer.GetFramer(framingMethodString)
 
 			if framer == nil {
 				return nil, fmt.Errorf("serial.client unknown framing method: %s", framingMethod)
@@ -76,13 +75,7 @@ func init() {
 				BaudRate: int(baudRateNum),
 			}
 
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &SerialClient{config: config, Port: portString, Framer: framer, Mode: &mode, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("serial.client unable to get router from context")
-			}
-
-			return &SerialClient{config: config, Port: portString, Framer: framer, Mode: &mode, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -107,7 +100,15 @@ func (sc *SerialClient) SetupPort() error {
 	return nil
 }
 
-func (sc *SerialClient) Run() error {
+func (sc *SerialClient) Run(ctx context.Context) error {
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("serial.client unable to get router from context")
+	}
+
+	sc.router = router
+	sc.ctx = ctx
 
 	// TODO(jwetzell): shutdown with router.Context properly
 	go func() {
@@ -155,7 +156,7 @@ func (sc *SerialClient) Run() error {
 							messages := sc.Framer.Decode(buffer[0:byteCount])
 							for _, message := range messages {
 								if sc.router != nil {
-									sc.router.HandleInput(sc.Id(), message)
+									sc.router.HandleInput(sc.ctx, sc.Id(), message)
 								} else {
 									sc.logger.Error("input received but no router is configured")
 								}

internal/module/sip-call-server.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+	"os"
+	"sync"
 	"time"
 
 	"github.com/emiago/diago"
@@ -13,6 +15,7 @@ import (
 	"github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
 	"github.com/jwetzell/showbridge-go/internal/config"
+	"github.com/jwetzell/showbridge-go/internal/processor"
 	"github.com/jwetzell/showbridge-go/internal/route"
 )
 
@@ -32,10 +35,17 @@ type SIPCallMessage struct {
 	To string
 }
 
+type SIPCall struct {
+	inDialog *diago.DialogServerSession
+	lock     sync.Mutex
+}
+
+type sipCallContextKey string
+
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "sip.call.server",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			portNum := 5060
 
@@ -88,12 +98,7 @@ func init() {
 				userAgentString = specificTransportString
 			}
 
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &SIPCallServer{config: config, IP: ipString, Port: int(portNum), Transport: transportString, UserAgent: userAgentString, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("sip.call.server unable to get router from context")
-			}
-			return &SIPCallServer{config: config, ctx: ctx, router: router, IP: ipString, Port: int(portNum), Transport: transportString, UserAgent: userAgentString, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -106,7 +111,15 @@ func (scs *SIPCallServer) Type() string {
 	return scs.config.Type
 }
 
-func (scs *SIPCallServer) Run() error {
+func (scs *SIPCallServer) Run(ctx context.Context) error {
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("sip.call.server unable to get router from context")
+	}
+	scs.router = router
+	scs.ctx = ctx
+
 	diagoLogger := slog.New(slog.NewJSONHandler(io.Discard, nil))
 
 	ua, _ := sipgo.NewUA(
@@ -143,51 +156,75 @@ func (scs *SIPCallServer) HandleCall(inDialog *diago.DialogServerSession) {
 	inDialog.Trying()
 	inDialog.Ringing()
 	inDialog.Answer()
-	scs.router.HandleInput(scs.Id(), SIPCallMessage{
+
+	dialogContext := context.WithValue(scs.ctx, sipCallContextKey("call"), &SIPCall{
+		inDialog: inDialog,
+	})
+	scs.router.HandleInput(dialogContext, scs.Id(), SIPCallMessage{
 		To: inDialog.ToUser(),
 	})
-	<-inDialog.Context().Done()
 }
 
 func (scs *SIPCallServer) Output(ctx context.Context, payload any) error {
 
-	payloadMsg, ok := payload.(string)
+	call, ok := ctx.Value(sipCallContextKey("call")).(*SIPCall)
+
 	if !ok {
-		return errors.New("sip.call.server output payload must be of type string")
+		return errors.New("sip.call.server output must originate from sip.call.server input")
 	}
 
-	if scs.dg == nil {
+	gotLock := call.lock.TryLock()
-		return errors.New("sip.call.server diago is not initialized")
+
+	if !gotLock {
+		return errors.New("sip.call.server call is already locked")
 	}
 
-	var uri sip.Uri
+	if call.inDialog.LoadState() == sip.DialogStateEnded {
-	err := sip.ParseUri(payloadMsg, &uri)
+		return errors.New("sip.call.server inDialog already ended")
-	if err != nil {
-		return fmt.Errorf("sip.call.server output payload is not a valid SIP URI: %s", err)
-	}
-	outDialog, err := scs.dg.NewDialog(uri, diago.NewDialogOptions{
-		Transport: scs.Transport,
-	})
-
-	if err != nil {
-		return fmt.Errorf("sip.call.server failed to create new dialog: %s", err)
 	}
 
-	err = outDialog.Invite(scs.ctx, diago.InviteClientOptions{})
+	payloadDTMFResponse, ok := payload.(processor.SipDTMFResponse)
-	if err != nil {
+
-		return fmt.Errorf("sip.call.server failed to send invite: %s", err)
+	if ok {
+		dtmfWriter := call.inDialog.AudioWriterDTMF()
+
+		time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PreWait))
+		for i, dtmfRune := range payloadDTMFResponse.Digits {
+			err := dtmfWriter.WriteDTMF(dtmfRune)
+
+			if err != nil {
+				return fmt.Errorf("sip.dtmf.server error output dtmf digit at index %d", i)
+			}
+		}
+		time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PreWait))
+		return nil
 	}
 
-	err = outDialog.Ack(scs.ctx)
+	payloadAudioFileResponse, ok := payload.(processor.SipAudioFileResponse)
-	if err != nil {
+
-		return fmt.Errorf("sip.call.server failed to send ack: %s", err)
+	if ok {
+		audioFile, err := os.Open(payloadAudioFileResponse.AudioFile)
+		if err != nil {
+			return err
+		}
+		defer audioFile.Close()
+
+		playback, err := call.inDialog.PlaybackCreate()
+
+		if err != nil {
+			return err
+		}
+
+		time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PreWait))
+
+		_, err = playback.Play(audioFile, "audio/wav")
+
+		time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PostWait))
+
+		if err != nil {
+			return err
+		}
+		return nil
 	}
-	// TODO(jwetzell): make this configurable
+	return errors.New("sip.dtmf.server can only output SipDTMFResponse or SipAudioFileResponse")
-	// NOTE(jwetzell): wait 5 seconds before hanging up the call
-	time.Sleep(5 * time.Second)
-	err = outDialog.Hangup(scs.ctx)
-	if err != nil {
-		return fmt.Errorf("sip.call.server failed to hangup call: %s", err)
-	}
-	return nil
 }

internal/module/sip-dtmf-server.go

@@ -3,9 +3,12 @@ package module
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
+	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/emiago/diago"
@@ -13,6 +16,7 @@ import (
 	"github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
 	"github.com/jwetzell/showbridge-go/internal/config"
+	"github.com/jwetzell/showbridge-go/internal/processor"
 	"github.com/jwetzell/showbridge-go/internal/route"
 )
 
@@ -32,10 +36,15 @@ type SIPDTMFMessage struct {
 	Digits string
 }
 
+type SIPDTMFCall struct {
+	inDialog *diago.DialogServerSession
+	lock     sync.Mutex
+}
+
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "sip.dtmf.server",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			portNum := 5060
 
@@ -91,12 +100,7 @@ func init() {
 			if !strings.ContainsRune("0123456789*#ABCD", rune(separatorString[0])) {
 				return nil, errors.New("sip.dtmf.server separator must be a valid DTMF character")
 			}
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &SIPDTMFServer{config: config, IP: ipString, Port: int(portNum), Transport: transportString, Separator: separatorString, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("sip.dtmf.server unable to get router from context")
-			}
-			return &SIPDTMFServer{config: config, ctx: ctx, router: router, IP: ipString, Port: int(portNum), Transport: transportString, Separator: separatorString, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -109,7 +113,15 @@ func (sds *SIPDTMFServer) Type() string {
 	return sds.config.Type
 }
 
-func (sds *SIPDTMFServer) Run() error {
+func (sds *SIPDTMFServer) Run(ctx context.Context) error {
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("sip.dtmf.server unable to get router from context")
+	}
+	sds.router = router
+	sds.ctx = ctx
+
 	diagoLogger := slog.New(slog.NewJSONHandler(io.Discard, nil))
 
 	ua, _ := sipgo.NewUA(
@@ -148,10 +160,14 @@ func (sds *SIPDTMFServer) HandleCall(inDialog *diago.DialogServerSession) error
 
 	reader := inDialog.AudioReaderDTMF()
 	userString := ""
+
 	return reader.Listen(func(dtmf rune) error {
 		if dtmf == rune(sds.Separator[0]) {
 			if sds.router != nil {
-				sds.router.HandleInput(sds.Id(), SIPDTMFMessage{
+				dialogContext := context.WithValue(sds.ctx, sipCallContextKey("call"), &SIPDTMFCall{
+					inDialog: inDialog,
+				})
+				sds.router.HandleInput(dialogContext, sds.Id(), SIPDTMFMessage{
 					To:     inDialog.ToUser(),
 					Digits: userString,
 				})
@@ -165,5 +181,65 @@ func (sds *SIPDTMFServer) HandleCall(inDialog *diago.DialogServerSession) error
 }
 
 func (sds *SIPDTMFServer) Output(ctx context.Context, payload any) error {
-	return errors.New("sip.dtmf.server output is not implemented")
+	call, ok := ctx.Value(sipCallContextKey("call")).(*SIPDTMFCall)
+
+	if !ok {
+		return errors.New("sip.dtmf.server output must originate from sip.dtmf.server input")
+	}
+
+	gotLock := call.lock.TryLock()
+
+	if !gotLock {
+		return errors.New("sip.dtmf.server call is already locked")
+	}
+
+	if call.inDialog.LoadState() == sip.DialogStateEnded {
+		return errors.New("sip.dtmf.server inDialog already ended")
+	}
+
+	payloadDTMFResponse, ok := payload.(processor.SipDTMFResponse)
+
+	if ok {
+		dtmfWriter := call.inDialog.AudioWriterDTMF()
+
+		time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PreWait))
+		for i, dtmfRune := range payloadDTMFResponse.Digits {
+			err := dtmfWriter.WriteDTMF(dtmfRune)
+
+			if err != nil {
+				return fmt.Errorf("sip.dtmf.server error output dtmf digit at index %d", i)
+			}
+		}
+		time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PreWait))
+		return nil
+	}
+
+	payloadAudioFileResponse, ok := payload.(processor.SipAudioFileResponse)
+
+	if ok {
+		audioFile, err := os.Open(payloadAudioFileResponse.AudioFile)
+		if err != nil {
+			return err
+		}
+		defer audioFile.Close()
+
+		playback, err := call.inDialog.PlaybackCreate()
+
+		if err != nil {
+			return err
+		}
+
+		time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PreWait))
+
+		_, err = playback.Play(audioFile, "audio/wav")
+
+		time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PostWait))
+
+		if err != nil {
+			return err
+		}
+		return nil
+	}
+
+	return errors.New("sip.dtmf.server can only output SipDTMFResponse or SipAudioFileResponse")
 }

internal/module/tcp-client.go

@@ -26,7 +26,7 @@ type TCPClient struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "net.tcp.client",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			host, ok := params["host"]
 
@@ -56,32 +56,24 @@ func init() {
 				return nil, err
 			}
 
-			framingMethod := "RAW"
+			framingMethod, ok := params["framing"]
 
-			framingMethodRaw, ok := params["framing"]
+			if !ok {
-
+				return nil, errors.New("net.tcp.client requires a framing parameter")
-			if ok {
-				framingMethodString, ok := framingMethodRaw.(string)
-
-				if !ok {
-					return nil, errors.New("net.tcp.client framing method must be a string")
-				}
-				framingMethod = framingMethodString
 			}
 
-			framer := framer.GetFramer(framingMethod)
+			framingMethodString, ok := framingMethod.(string)
+
+			if !ok {
+				return nil, errors.New("net.tcp.client framing method must be a string")
+			}
+
+			framer := framer.GetFramer(framingMethodString)
 
 			if framer == nil {
 				return nil, fmt.Errorf("net.tcp.client unknown framing method: %s", framingMethod)
 			}
-
+			return &TCPClient{framer: framer, Addr: addr, config: config, logger: CreateLogger(config)}, nil
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
-
-			if !ok {
-				return nil, errors.New("net.tcp.client unable to get router from context")
-			}
-
-			return &TCPClient{framer: framer, Addr: addr, config: config, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -94,7 +86,15 @@ func (tc *TCPClient) Type() string {
 	return tc.config.Type
 }
 
-func (tc *TCPClient) Run() error {
+func (tc *TCPClient) Run(ctx context.Context) error {
+
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("net.tcp.client unable to get router from context")
+	}
+	tc.router = router
+	tc.ctx = ctx
 
 	// TODO(jwetzell): shutdown with router.Context properly
 	go func() {
@@ -142,7 +142,7 @@ func (tc *TCPClient) Run() error {
 							messages := tc.framer.Decode(buffer[0:byteCount])
 							for _, message := range messages {
 								if tc.router != nil {
-									tc.router.HandleInput(tc.Id(), message)
+									tc.router.HandleInput(tc.ctx, tc.Id(), message)
 								} else {
 									tc.logger.Error("input received but no router is configured")
 								}

internal/module/tcp-server.go

@@ -32,7 +32,7 @@ type TCPServer struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "net.tcp.server",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			port, ok := params["port"]
 			if !ok {
@@ -45,20 +45,19 @@ func init() {
 				return nil, errors.New("net.tcp.server port must be a number")
 			}
 
-			framingMethod := "RAW"
+			framingMethod, ok := params["framing"]
 
-			framingMethodRaw, ok := params["framing"]
+			if !ok {
-
+				return nil, errors.New("net.tcp.server requires a framing parameter")
-			if ok {
-				framingMethodString, ok := framingMethodRaw.(string)
-
-				if !ok {
-					return nil, errors.New("net.tcp.server framing method must be a string")
-				}
-				framingMethod = framingMethodString
 			}
 
-			framer := framer.GetFramer(framingMethod)
+			framingMethodString, ok := framingMethod.(string)
+
+			if !ok {
+				return nil, errors.New("net.tcp.server framing method must be a string")
+			}
+
+			framer := framer.GetFramer(framingMethodString)
 
 			if framer == nil {
 				return nil, fmt.Errorf("net.tcp.server unknown framing method: %s", framingMethod)
@@ -81,14 +80,7 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
-
+			return &TCPServer{Framer: framer, Addr: addr, config: config, quit: make(chan interface{}), logger: CreateLogger(config)}, nil
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
-
-			if !ok {
-				return nil, errors.New("net.tcp.server unable to get router from context")
-			}
-
-			return &TCPServer{Framer: framer, Addr: addr, config: config, quit: make(chan interface{}), ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -155,7 +147,7 @@ ClientRead:
 					messages := ts.Framer.Decode(buffer[0:byteCount])
 					for _, message := range messages {
 						if ts.router != nil {
-							ts.router.HandleInput(ts.Id(), message)
+							ts.router.HandleInput(ts.ctx, ts.Id(), message)
 						} else {
 							ts.logger.Error("input received but no router is configured")
 						}
@@ -166,7 +158,16 @@ ClientRead:
 	}
 }
 
-func (ts *TCPServer) Run() error {
+func (ts *TCPServer) Run(ctx context.Context) error {
+
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("net.tcp.server unable to get router from context")
+	}
+	ts.router = router
+	ts.ctx = ctx
+
 	listener, err := net.ListenTCP("tcp", ts.Addr)
 	if err != nil {
 		return err

internal/module/time-interval.go

@@ -22,7 +22,7 @@ type TimeInterval struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "time.interval",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 
 			duration, ok := params["duration"]
@@ -35,14 +35,7 @@ func init() {
 			if !ok {
 				return nil, errors.New("time.interval duration must be number")
 			}
-
+			return &TimeInterval{Duration: uint32(durationNum), config: config, logger: CreateLogger(config)}, nil
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
-
-			if !ok {
-				return nil, errors.New("time.interval unable to get router from context")
-			}
-
-			return &TimeInterval{Duration: uint32(durationNum), config: config, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -55,7 +48,16 @@ func (i *TimeInterval) Type() string {
 	return i.config.Type
 }
 
-func (i *TimeInterval) Run() error {
+func (i *TimeInterval) Run(ctx context.Context) error {
+
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("time.interval unable to get router from context")
+	}
+	i.router = router
+	i.ctx = ctx
+
 	ticker := time.NewTicker(time.Millisecond * time.Duration(i.Duration))
 	i.ticker = ticker
 	defer ticker.Stop()
@@ -67,7 +69,7 @@ func (i *TimeInterval) Run() error {
 			return nil
 		case <-ticker.C:
 			if i.router != nil {
-				i.router.HandleInput(i.Id(), time.Now())
+				i.router.HandleInput(i.ctx, i.Id(), time.Now())
 			}
 		}
 	}

internal/module/time-timer.go

@@ -22,7 +22,7 @@ type TimeTimer struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "time.timer",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 
 			duration, ok := params["duration"]
@@ -36,13 +36,7 @@ func init() {
 				return nil, errors.New("time.timer duration must be a number")
 			}
 
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+			return &TimeTimer{Duration: uint32(durationNum), config: config, logger: CreateLogger(config)}, nil
-
-			if !ok {
-				return nil, errors.New("net.tcp.client unable to get router from context")
-			}
-
-			return &TimeTimer{Duration: uint32(durationNum), config: config, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -55,7 +49,16 @@ func (t *TimeTimer) Type() string {
 	return t.config.Type
 }
 
-func (t *TimeTimer) Run() error {
+func (t *TimeTimer) Run(ctx context.Context) error {
+
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("net.tcp.client unable to get router from context")
+	}
+	t.router = router
+	t.ctx = ctx
+
 	t.timer = time.NewTimer(time.Millisecond * time.Duration(t.Duration))
 	defer t.timer.Stop()
 	for {
@@ -66,7 +69,7 @@ func (t *TimeTimer) Run() error {
 			return nil
 		case time := <-t.timer.C:
 			if t.router != nil {
-				t.router.HandleInput(t.Id(), time)
+				t.router.HandleInput(t.ctx, t.Id(), time)
 			}
 		}
 	}

internal/module/udp-client.go

@@ -24,7 +24,7 @@ type UDPClient struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "net.udp.client",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			host, ok := params["host"]
 
@@ -53,14 +53,7 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
-
+			return &UDPClient{Addr: addr, config: config, logger: CreateLogger(config)}, nil
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
-
-			if !ok {
-				return nil, errors.New("net.udp.client unable to get router from context")
-			}
-
-			return &UDPClient{Addr: addr, config: config, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -79,7 +72,15 @@ func (uc *UDPClient) SetupConn() error {
 	return err
 }
 
-func (uc *UDPClient) Run() error {
+func (uc *UDPClient) Run(ctx context.Context) error {
+
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("net.udp.client unable to get router from context")
+	}
+	uc.router = router
+	uc.ctx = ctx
 
 	err := uc.SetupConn()
 	if err != nil {

internal/module/udp-multicast.go

@@ -24,7 +24,7 @@ type UDPMulticast struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "net.udp.multicast",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			ip, ok := params["ip"]
 
@@ -53,13 +53,7 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
-
+			return &UDPMulticast{config: config, Addr: addr, logger: CreateLogger(config)}, nil
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
-
-			if !ok {
-				return nil, errors.New("net.udp.multicast unable to get router from context")
-			}
-			return &UDPMulticast{config: config, Addr: addr, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -72,7 +66,15 @@ func (um *UDPMulticast) Type() string {
 	return um.config.Type
 }
 
-func (um *UDPMulticast) Run() error {
+func (um *UDPMulticast) Run(ctx context.Context) error {
+
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("net.udp.multicast unable to get router from context")
+	}
+	um.router = router
+	um.ctx = ctx
 
 	client, err := net.ListenMulticastUDP("udp", nil, um.Addr)
 	if err != nil {
@@ -105,7 +107,7 @@ func (um *UDPMulticast) Run() error {
 				message := buffer[:numBytes]
 
 				if um.router != nil {
-					um.router.HandleInput(um.Id(), message)
+					um.router.HandleInput(um.ctx, um.Id(), message)
 				} else {
 					um.logger.Error("input received but no router is configured")
 				}

internal/module/udp-server.go

@@ -25,7 +25,7 @@ type UDPServer struct {
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "net.udp.server",
-		New: func(ctx context.Context, config config.ModuleConfig) (Module, error) {
+		New: func(config config.ModuleConfig) (Module, error) {
 			params := config.Params
 			port, ok := params["port"]
 			if !ok {
@@ -67,13 +67,7 @@ func init() {
 				}
 				bufferSizeNum = int(bufferSizeFloat)
 			}
-
+			return &UDPServer{Addr: addr, BufferSize: bufferSizeNum, config: config, logger: CreateLogger(config)}, nil
-			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
-
-			if !ok {
-				return nil, errors.New("net.udp.server unable to get router from context")
-			}
-			return &UDPServer{Addr: addr, BufferSize: bufferSizeNum, config: config, ctx: ctx, router: router, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -86,7 +80,15 @@ func (us *UDPServer) Type() string {
 	return us.config.Id
 }
 
-func (us *UDPServer) Run() error {
+func (us *UDPServer) Run(ctx context.Context) error {
+
+	router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
+
+	if !ok {
+		return errors.New("net.udp.server unable to get router from context")
+	}
+	us.router = router
+	us.ctx = ctx
 
 	listener, err := net.ListenUDP("udp", us.Addr)
 	if err != nil {
@@ -115,7 +117,7 @@ func (us *UDPServer) Run() error {
 			}
 			message := buffer[:numBytes]
 			if us.router != nil {
-				us.router.HandleInput(us.Id(), message)
+				us.router.HandleInput(us.ctx, us.Id(), message)
 			} else {
 				us.logger.Error("input received but no router is configured")
 			}

internal/processor/debug-log.go

@@ -14,7 +14,9 @@ type DebugLog struct {
 }
 
 func (dl *DebugLog) Process(ctx context.Context, payload any) (any, error) {
-	dl.logger.Debug("", "payload", payload, "payloadType", fmt.Sprintf("%T", payload))
+	payloadString := fmt.Sprintf("%+v", payload)
+	payloadType := fmt.Sprintf("%T", payload)
+	dl.logger.Debug("", "payload", payloadString, "payloadType", payloadType)
 	return payload, nil
 }
 

internal/processor/http-response-create.go

@@ -0,0 +1,81 @@
+package processor
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"text/template"
+
+	"github.com/jwetzell/showbridge-go/internal/config"
+)
+
+type HTTPResponseCreate struct {
+	Status   int
+	BodyTmpl *template.Template
+	config   config.ProcessorConfig
+}
+
+type HTTPResponse struct {
+	Status int
+	Body   []byte
+}
+
+func (hre *HTTPResponseCreate) Process(ctx context.Context, payload any) (any, error) {
+	var bodyBuffer bytes.Buffer
+	err := hre.BodyTmpl.Execute(&bodyBuffer, payload)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return HTTPResponse{
+		Status: hre.Status,
+		Body:   bodyBuffer.Bytes(),
+	}, nil
+}
+
+func (hre *HTTPResponseCreate) Type() string {
+	return hre.config.Type
+}
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type: "http.response.create",
+		New: func(config config.ProcessorConfig) (Processor, error) {
+			params := config.Params
+
+			status, ok := params["status"]
+
+			if !ok {
+				return nil, errors.New("http.response.create requires a status parameter")
+			}
+
+			statusNum, ok := status.(float64)
+
+			if !ok {
+				return nil, errors.New("http.response.create status must be a number")
+			}
+
+			bodyTmpl, ok := params["bodyTemplate"]
+
+			if !ok {
+				return nil, errors.New("http.response.create requires a bodyTemplate parameter")
+			}
+
+			bodyTemplateString, ok := bodyTmpl.(string)
+
+			if !ok {
+				return nil, errors.New("http.response.create bodyTemplate must be a string")
+			}
+
+			bodyTemplate, err := template.New("body").Parse(bodyTemplateString)
+
+			if err != nil {
+				return nil, err
+			}
+
+			// TODO(jwetzell): support other body kind (direct bytes from input, from file?)
+			return &HTTPResponseCreate{config: config, Status: int(statusNum), BodyTmpl: bodyTemplate}, nil
+		},
+	})
+}

internal/processor/json-encode.go

@@ -0,0 +1,42 @@
+package processor
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+
+	"github.com/jwetzell/showbridge-go/internal/config"
+)
+
+type JsonEncode struct {
+	config config.ProcessorConfig
+}
+
+func (je *JsonEncode) Process(ctx context.Context, payload any) (any, error) {
+	var payloadBuffer bytes.Buffer
+
+	err := json.NewEncoder(&payloadBuffer).Encode(payload)
+
+	if err != nil {
+		return nil, err
+	}
+
+	payloadBytes := payloadBuffer.Bytes()
+
+	payloadBytes = payloadBytes[0 : len(payloadBytes)-1]
+
+	return payloadBytes, nil
+}
+
+func (je *JsonEncode) Type() string {
+	return je.config.Type
+}
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type: "json.encode",
+		New: func(config config.ProcessorConfig) (Processor, error) {
+			return &JsonEncode{config: config}, nil
+		},
+	})
+}

internal/processor/json-encode_test.go

@@ -0,0 +1,45 @@
+package processor_test
+
+import (
+	"slices"
+	"testing"
+
+	"github.com/jwetzell/osc-go"
+	"github.com/jwetzell/showbridge-go/internal/processor"
+)
+
+func TestGoodJsonEncode(t *testing.T) {
+	stringEncoder := processor.JsonEncode{}
+	tests := []struct {
+		processor processor.Processor
+		name      string
+		payload   any
+		expected  []byte
+	}{
+		{
+			processor: &stringEncoder,
+			name:      "hello",
+			payload: osc.OSCMessage{
+				Address: "/hello",
+			},
+			expected: []byte("{\"address\":\"/hello\",\"args\":null}"),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got, err := test.processor.Process(t.Context(), test.payload)
+
+			gotBytes, ok := got.([]byte)
+			if !ok {
+				t.Fatalf("json.encode returned a %T payload: %s", got, got)
+			}
+			if err != nil {
+				t.Fatalf("json.encode failed: %s", err)
+			}
+			if !slices.Equal(gotBytes, test.expected) {
+				t.Fatalf("json.encode got %x, expected %s", got, test.expected)
+			}
+		})
+	}
+}

internal/processor/mqtt-message-create.go

@@ -51,6 +51,7 @@ func (mm MQTTMessage) Payload() []byte {
 func (mm MQTTMessage) Ack() {}
 
 func (mmc *MQTTMessageCreate) Process(ctx context.Context, payload any) (any, error) {
+	// TODO(jwetzell): support templating
 
 	message := MQTTMessage{
 		topic:    mmc.Topic,

internal/processor/nats-message-create.go

@@ -16,7 +16,7 @@ type NATSMessage struct {
 
 type NATSMessageCreate struct {
 	config  config.ProcessorConfig
-	Subject string
+	Subject *template.Template
 	Payload *template.Template
 }
 
@@ -31,8 +31,17 @@ func (nmc *NATSMessageCreate) Process(ctx context.Context, payload any) (any, er
 
 	payloadString := payloadBuffer.String()
 
+	var subjectBuffer bytes.Buffer
+	err = nmc.Subject.Execute(&subjectBuffer, payload)
+
+	if err != nil {
+		return nil, err
+	}
+
+	subjectString := subjectBuffer.String()
+
 	message := NATSMessage{
-		Subject: nmc.Subject,
+		Subject: subjectString,
 		Payload: []byte(payloadString),
 	}
 
@@ -48,7 +57,6 @@ func init() {
 		Type: "nats.message.create",
 		New: func(config config.ProcessorConfig) (Processor, error) {
 			params := config.Params
-			// TODO(jwetzell): support template for subject
 			subject, ok := params["subject"]
 
 			if !ok {
@@ -61,6 +69,12 @@ func init() {
 				return nil, errors.New("nats.message.create subject must be a string")
 			}
 
+			subjectTemplate, err := template.New("subject").Parse(subjectString)
+
+			if err != nil {
+				return nil, err
+			}
+
 			payload, ok := params["payload"]
 
 			if !ok {
@@ -79,7 +93,7 @@ func init() {
 				return nil, err
 			}
 
-			return &NATSMessageCreate{config: config, Subject: subjectString, Payload: payloadTemplate}, nil
+			return &NATSMessageCreate{config: config, Subject: subjectTemplate, Payload: payloadTemplate}, nil
 		},
 	})
 }

internal/processor/script-wasm.go

@@ -0,0 +1,92 @@
+package processor
+
+import (
+	"context"
+	"fmt"
+
+	extism "github.com/extism/go-sdk"
+	"github.com/jwetzell/showbridge-go/internal/config"
+)
+
+type ScriptWASM struct {
+	config   config.ProcessorConfig
+	Program  *extism.CompiledPlugin
+	Function string
+}
+
+func (se *ScriptWASM) Process(ctx context.Context, payload any) (any, error) {
+
+	payloadBytes, ok := payload.([]byte)
+
+	if !ok {
+		return nil, fmt.Errorf("script.wasm can only operator on byte array")
+	}
+
+	program, err := se.Program.Instance(ctx, extism.PluginInstanceConfig{})
+
+	if err != nil {
+		return nil, err
+	}
+
+	_, output, err := program.Call(se.Function, payloadBytes)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return output, nil
+}
+
+func (se *ScriptWASM) Type() string {
+	return se.config.Type
+}
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type: "script.wasm",
+		New: func(config config.ProcessorConfig) (Processor, error) {
+			params := config.Params
+
+			path, ok := params["path"]
+
+			if !ok {
+				return nil, fmt.Errorf("script.wasm requires a path parameter")
+			}
+
+			pathString, ok := path.(string)
+
+			if !ok {
+				return nil, fmt.Errorf("script.wasm path must be a string")
+			}
+
+			functionString := "process"
+
+			function, ok := params["function"]
+
+			if ok {
+				specificFunctionString, ok := function.(string)
+
+				if !ok {
+					return nil, fmt.Errorf("script.wasm function must be a string")
+				}
+				functionString = specificFunctionString
+			}
+
+			manifest := extism.Manifest{
+				Wasm: []extism.Wasm{
+					extism.WasmFile{
+						Path: pathString,
+					},
+				},
+			}
+
+			program, err := extism.NewCompiledPlugin(context.Background(), manifest, extism.PluginConfig{}, []extism.HostFunction{})
+
+			if err != nil {
+				return nil, err
+			}
+
+			return &ScriptWASM{config: config, Program: program, Function: functionString}, nil
+		},
+	})
+}

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

@@ -0,0 +1,97 @@
+package processor
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"text/template"
+
+	"github.com/jwetzell/showbridge-go/internal/config"
+)
+
+type SipResponseAudioCreate struct {
+	config    config.ProcessorConfig
+	PreWait   int
+	PostWait  int
+	AudioFile *template.Template
+}
+
+type SipAudioFileResponse struct {
+	PreWait   int
+	PostWait  int
+	AudioFile string
+}
+
+func (scc *SipResponseAudioCreate) Process(ctx context.Context, payload any) (any, error) {
+
+	var audioFileBuffer bytes.Buffer
+	err := scc.AudioFile.Execute(&audioFileBuffer, payload)
+
+	if err != nil {
+		return nil, err
+	}
+
+	audioFileString := audioFileBuffer.String()
+
+	return SipAudioFileResponse{
+		PreWait:   scc.PreWait,
+		PostWait:  scc.PostWait,
+		AudioFile: audioFileString,
+	}, nil
+}
+
+func (scc *SipResponseAudioCreate) Type() string {
+	return scc.config.Type
+}
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type: "sip.response.audio.create",
+		New: func(config config.ProcessorConfig) (Processor, error) {
+			params := config.Params
+
+			preWait, ok := params["preWait"]
+
+			if !ok {
+				return nil, errors.New("sip.response.audio.create requires a preWait parameter")
+			}
+
+			preWaitNum, ok := preWait.(float64)
+
+			if !ok {
+				return nil, errors.New("sip.response.audio.create preWait must be a number")
+			}
+
+			postWait, ok := params["postWait"]
+
+			if !ok {
+				return nil, errors.New("sip.response.audio.create requires a postWait parameter")
+			}
+
+			postWaitNum, ok := postWait.(float64)
+
+			if !ok {
+				return nil, errors.New("sip.response.audio.create postWait must be a number")
+			}
+
+			audioFile, ok := params["audioFile"]
+
+			if !ok {
+				return nil, errors.New("sip.response.audio.create requires a audioFile parameter")
+			}
+
+			audioFileString, ok := audioFile.(string)
+
+			if !ok {
+				return nil, errors.New("sip.response.audio.create audioFile must be a string")
+			}
+
+			audioFileTemplate, err := template.New("audioFile").Parse(audioFileString)
+
+			if err != nil {
+				return nil, err
+			}
+			return &SipResponseAudioCreate{config: config, AudioFile: audioFileTemplate, PreWait: int(preWaitNum), PostWait: int(postWaitNum)}, nil
+		},
+	})
+}

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

@@ -0,0 +1,103 @@
+package processor
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"regexp"
+	"text/template"
+
+	"github.com/jwetzell/showbridge-go/internal/config"
+)
+
+type SipResponseDTMFCreate struct {
+	config    config.ProcessorConfig
+	PreWait   int
+	PostWait  int
+	Digits    *template.Template
+	validDTMF *regexp.Regexp
+}
+
+type SipDTMFResponse struct {
+	PreWait  int
+	PostWait int
+	Digits   string
+}
+
+func (scc *SipResponseDTMFCreate) Process(ctx context.Context, payload any) (any, error) {
+
+	var digitsBuffer bytes.Buffer
+	err := scc.Digits.Execute(&digitsBuffer, payload)
+
+	if err != nil {
+		return nil, err
+	}
+
+	digitsString := digitsBuffer.String()
+
+	if !scc.validDTMF.MatchString(digitsString) {
+		return nil, errors.New("sip.response.dtmf.create result of digits template contains invalid characters")
+	}
+
+	return SipDTMFResponse{
+		PreWait:  scc.PreWait,
+		PostWait: scc.PostWait,
+		Digits:   digitsString,
+	}, nil
+}
+
+func (scc *SipResponseDTMFCreate) Type() string {
+	return scc.config.Type
+}
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type: "sip.response.dtmf.create",
+		New: func(config config.ProcessorConfig) (Processor, error) {
+			params := config.Params
+
+			preWait, ok := params["preWait"]
+
+			if !ok {
+				return nil, errors.New("sip.response.dtmf.create requires a preWait parameter")
+			}
+
+			preWaitNum, ok := preWait.(float64)
+
+			if !ok {
+				return nil, errors.New("sip.response.dtmf.create preWait must be a number")
+			}
+
+			postWait, ok := params["postWait"]
+
+			if !ok {
+				return nil, errors.New("sip.response.dtmf.create requires a postWait parameter")
+			}
+
+			postWaitNum, ok := postWait.(float64)
+
+			if !ok {
+				return nil, errors.New("sip.response.dtmf.create postWait must be a number")
+			}
+
+			digits, ok := params["digits"]
+
+			if !ok {
+				return nil, errors.New("sip.response.dtmf.create requires a digits parameter")
+			}
+
+			digitsString, ok := digits.(string)
+
+			if !ok {
+				return nil, errors.New("sip.response.dtmf.create digits must be a string")
+			}
+
+			digitsTemplate, err := template.New("digits").Parse(digitsString)
+
+			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
+		},
+	})
+}

internal/processor/string-create_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 	"text/template"
 
+	"github.com/jwetzell/showbridge-go/internal/config"
 	"github.com/jwetzell/showbridge-go/internal/processor"
 )
 
@@ -15,6 +16,39 @@ func (t TestStruct) GetData() string {
 	return t.Data
 }
 
+func TestStringCreateFromRegistry(t *testing.T) {
+	registration, ok := processor.ProcessorRegistry["string.create"]
+	if !ok {
+		t.Fatalf("string.create processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "string.create",
+		Params: map[string]any{
+			"template": "{{.}}",
+		},
+	})
+	if err != nil {
+		t.Fatalf("failed to create string.create processor: %s", err)
+	}
+
+	if processorInstance.Type() != "string.create" {
+		t.Fatalf("string.create processor has wrong type: %s", processorInstance.Type())
+	}
+
+	payload := "hello"
+	expected := "hello"
+
+	got, err := processorInstance.Process(t.Context(), payload)
+	if err != nil {
+		t.Fatalf("string.create processing failed: %s", err)
+	}
+
+	if got != expected {
+		t.Fatalf("string.create got %+v, expected %+v", got, expected)
+	}
+}
+
 func TestGoodStringCreate(t *testing.T) {
 
 	tests := []struct {
@@ -79,3 +113,75 @@ func TestGoodStringCreate(t *testing.T) {
 		})
 	}
 }
+
+func TestBadStringCreate(t *testing.T) {
+	tests := []struct {
+		name        string
+		params      map[string]any
+		payload     any
+		errorString string
+	}{
+		{
+			name:        "no template param",
+			payload:     "hello",
+			params:      map[string]any{},
+			errorString: "string.create requires a template parameter",
+		},
+		{
+			name:    "non string template",
+			payload: "hello",
+			params: map[string]any{
+				"template": 1,
+			},
+			errorString: "string.create template must be a string",
+		},
+		{
+			name:    "invalid template",
+			payload: "hello",
+			params: map[string]any{
+				"template": "{{.",
+			},
+			errorString: "template: template:1: illegal number syntax: \".\"",
+		},
+		{
+			name:    "bad property in template",
+			payload: "hello",
+			params: map[string]any{
+				"template": "{{.Invalid}}",
+			},
+			errorString: "template: template:1:2: executing \"template\" at <.Invalid>: can't evaluate field Invalid in type string",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			registration, ok := processor.ProcessorRegistry["string.create"]
+			if !ok {
+				t.Fatalf("string.create processor not registered")
+			}
+
+			processorInstance, err := registration.New(config.ProcessorConfig{
+				Type:   "string.create",
+				Params: test.params,
+			})
+
+			if err != nil {
+				if test.errorString != err.Error() {
+					t.Fatalf("string.encode got error '%s', expected '%s'", err.Error(), test.errorString)
+				}
+				return
+			}
+
+			got, err := processorInstance.Process(t.Context(), test.payload)
+
+			if err == nil {
+				t.Fatalf("string.encode expected to fail but got payload: %s", got)
+			}
+
+			if err.Error() != test.errorString {
+				t.Fatalf("string.encode got error '%s', expected '%s'", err.Error(), test.errorString)
+			}
+		})
+	}
+}

internal/processor/string-decode_test.go

@@ -3,9 +3,40 @@ package processor_test
 import (
 	"testing"
 
+	"github.com/jwetzell/showbridge-go/internal/config"
 	"github.com/jwetzell/showbridge-go/internal/processor"
 )
 
+func TestStringDecodeFromRegistry(t *testing.T) {
+	registration, ok := processor.ProcessorRegistry["string.decode"]
+	if !ok {
+		t.Fatalf("string.decode processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "string.decode",
+	})
+	if err != nil {
+		t.Fatalf("failed to create string.decode processor: %s", err)
+	}
+
+	if processorInstance.Type() != "string.decode" {
+		t.Fatalf("string.decode processor has wrong type: %s", processorInstance.Type())
+	}
+
+	payload := []byte{'h', 'e', 'l', 'l', 'o'}
+	expected := "hello"
+
+	got, err := processorInstance.Process(t.Context(), payload)
+	if err != nil {
+		t.Fatalf("string.decode processing failed: %s", err)
+	}
+
+	if got != expected {
+		t.Fatalf("string.decode got %+v, expected %+v", got, expected)
+	}
+}
+
 func TestGoodStringDecode(t *testing.T) {
 	stringDecoder := processor.StringDecode{}
 	tests := []struct {

internal/processor/string-encode_test.go

@@ -4,9 +4,46 @@ import (
 	"slices"
 	"testing"
 
+	"github.com/jwetzell/showbridge-go/internal/config"
 	"github.com/jwetzell/showbridge-go/internal/processor"
 )
 
+func TestStringEncodeFromRegistry(t *testing.T) {
+	registration, ok := processor.ProcessorRegistry["string.encode"]
+	if !ok {
+		t.Fatalf("string.encode processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "string.encode",
+	})
+	if err != nil {
+		t.Fatalf("failed to create string.encode processor: %s", err)
+	}
+
+	if processorInstance.Type() != "string.encode" {
+		t.Fatalf("string.encode processor has wrong type: %s", processorInstance.Type())
+	}
+
+	payload := "hello"
+	expected := []byte{'h', 'e', 'l', 'l', 'o'}
+
+	got, err := processorInstance.Process(t.Context(), payload)
+	if err != nil {
+		t.Fatalf("string.encode processing failed: %s", err)
+	}
+
+	gotBytes, ok := got.([]byte)
+
+	if !ok {
+		t.Fatalf("string.encode should return byte slice")
+	}
+
+	if !slices.Equal(gotBytes, expected) {
+		t.Fatalf("string.encode got %+v, expected %+v", got, expected)
+	}
+}
+
 func TestGoodStringEncode(t *testing.T) {
 	stringEncoder := processor.StringEncode{}
 	tests := []struct {

internal/route/route.go

@@ -20,13 +20,15 @@ type RouteError struct {
 }
 
 type RouteIOError struct {
-	Index int
+	Index        int
-	Error error
+	OutputErrors []error
+	ProcessError error
+	InputError   error
 }
 
 type RouteIO interface {
-	HandleInput(sourceId string, payload any) []RouteIOError
+	HandleInput(ctx context.Context, sourceId string, payload any) (bool, []RouteIOError)
-	HandleOutput(ctx context.Context, destinationId string, payload any) error
+	HandleOutput(ctx context.Context, destinationId string, payload any) []error
 }
 
 type Route interface {

internal/route/route_test.go

@@ -30,8 +30,8 @@ func TestRouteCreate(t *testing.T) {
 
 type MockRouter struct{}
 
-func (mr *MockRouter) HandleInput(sourceId string, payload any) []route.RouteIOError {
+func (mr *MockRouter) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []route.RouteIOError) {
-	return nil
+	return false, []route.RouteIOError{}
 }
 
 func (mr *MockRouter) HandleOutput(ctx context.Context, destinationId string, payload any) error {

router.go

@@ -3,7 +3,6 @@ package showbridge
 import (
 	"context"
 	"errors"
-	"fmt"
 	"log/slog"
 	"sync"
 
@@ -21,25 +20,32 @@ type Router struct {
 	logger          *slog.Logger
 }
 
-func NewRouter(ctx context.Context, config config.Config) (*Router, []module.ModuleError, []route.RouteError) {
+func NewRouter(config config.Config) (*Router, []module.ModuleError, []route.RouteError) {
-
-	routerContext, cancel := context.WithCancel(ctx)
 
 	router := Router{
-		contextCancel:   cancel,
 		ModuleInstances: []module.Module{},
 		RouteInstances:  []route.Route{},
 		logger:          slog.Default().With("component", "router"),
 	}
 
-	router.Context = context.WithValue(routerContext, route.RouterContextKey, &router)
-
 	router.logger.Debug("creating")
 
 	var moduleErrors []module.ModuleError
 
 	for moduleIndex, moduleDecl := range config.Modules {
 
+		if moduleDecl.Id == "" {
+			if moduleErrors == nil {
+				moduleErrors = []module.ModuleError{}
+			}
+			moduleErrors = append(moduleErrors, module.ModuleError{
+				Index:  moduleIndex,
+				Config: moduleDecl,
+				Error:  errors.New("module id cannot be empty"),
+			})
+			continue
+		}
+
 		moduleInfo, ok := module.ModuleRegistry[moduleDecl.Type]
 		if !ok {
 			if moduleErrors == nil {
@@ -70,7 +76,7 @@ func NewRouter(ctx context.Context, config config.Config) (*Router, []module.Mod
 		}
 
 		if !moduleInstanceExists {
-			moduleInstance, err := moduleInfo.New(router.Context, moduleDecl)
+			moduleInstance, err := moduleInfo.New(moduleDecl)
 			if err != nil {
 				if moduleErrors == nil {
 					moduleErrors = []module.ModuleError{}
@@ -108,11 +114,16 @@ func NewRouter(ctx context.Context, config config.Config) (*Router, []module.Mod
 	return &router, moduleErrors, routeErrors
 }
 
-func (r *Router) Run() {
+func (r *Router) Run(ctx context.Context) {
 	r.logger.Info("running")
+	routerContext, cancel := context.WithCancel(ctx)
+	r.Context = routerContext
+	r.contextCancel = cancel
+	contextWithRouter := context.WithValue(routerContext, route.RouterContextKey, r)
+
 	for _, moduleInstance := range r.ModuleInstances {
 		r.moduleWait.Go(func() {
-			err := moduleInstance.Run()
+			err := moduleInstance.Run(contextWithRouter)
 			if err != nil {
 				r.logger.Error("error encountered running module", "error", err)
 			}
@@ -125,38 +136,73 @@ func (r *Router) Run() {
 }
 
 func (r *Router) Stop() {
-	r.logger.Debug("stopping")
+	r.logger.Info("stopping")
 	r.contextCancel()
 }
 
-func (r *Router) HandleInput(sourceId string, payload any) []route.RouteIOError {
+func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []route.RouteIOError) {
-	var routingErrors []route.RouteIOError
+	var routeIOErrors []route.RouteIOError
+	routeFound := false
+
+	var routeWaitGroup sync.WaitGroup
+
 	for routeIndex, routeInstance := range r.RouteInstances {
 		if routeInstance.Input() == sourceId {
-			routeContext := context.WithValue(r.Context, route.SourceContextKey, sourceId)
+			routeWaitGroup.Go(func() {
 
-			payload, err := routeInstance.ProcessPayload(routeContext, payload)
+				routeFound = true
-			if err != nil {
+				routeContext := context.WithValue(ctx, route.SourceContextKey, sourceId)
-				if routingErrors == nil {
+
-					routingErrors = []route.RouteIOError{}
+				payload, err := routeInstance.ProcessPayload(routeContext, payload)
+				if err != nil {
+					if routeIOErrors == nil {
+						routeIOErrors = []route.RouteIOError{}
+					}
+					r.logger.Error("unable to process input", "route", routeIndex, "source", sourceId, "error", err)
+					routeIOErrors = append(routeIOErrors, route.RouteIOError{
+						Index:        routeIndex,
+						ProcessError: err,
+					})
+					return
 				}
-				routingErrors = append(routingErrors, route.RouteIOError{
+
-					Index: routeIndex,
+				if payload == nil {
-					Error: err,
+					r.logger.Error("no input after processing", "route", routeIndex, "source", sourceId)
-				})
+					return
-				r.logger.Error("unable to route input", "route", routeIndex, "source", sourceId, "error", err)
+				}
-			}
+
-			r.HandleOutput(routeContext, routeInstance.Output(), payload)
+				outputErrors := r.HandleOutput(routeContext, routeInstance.Output(), payload)
+				if outputErrors != nil {
+					if routeIOErrors == nil {
+						routeIOErrors = []route.RouteIOError{}
+					}
+					routeIOErrors = append(routeIOErrors, route.RouteIOError{
+						Index:        routeIndex,
+						OutputErrors: outputErrors,
+					})
+				}
+			})
+
 		}
 	}
-	return routingErrors
+	routeWaitGroup.Wait()
+	return routeFound, routeIOErrors
 }
 
-func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload any) error {
+func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload any) []error {
+
+	var outputErrors []error
 	for _, moduleInstance := range r.ModuleInstances {
 		if moduleInstance.Id() == destinationId {
-			return moduleInstance.Output(ctx, payload)
+			err := moduleInstance.Output(ctx, payload)
+			if err != nil {
+				if outputErrors == nil {
+					outputErrors = []error{}
+				}
+				outputErrors = append(outputErrors, err)
+				r.logger.Error("unable to route output", "module", moduleInstance.Id(), "error", err)
+			}
 		}
 	}
-	return fmt.Errorf("router could not find module instance for destination %s", destinationId)
+	return outputErrors
 }

router_test.go

@@ -0,0 +1,341 @@
+package showbridge_test
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jwetzell/showbridge-go"
+	"github.com/jwetzell/showbridge-go/internal/config"
+	"github.com/jwetzell/showbridge-go/internal/module"
+	"github.com/jwetzell/showbridge-go/internal/route"
+)
+
+type MockModule struct {
+	config      config.ModuleConfig
+	ctx         context.Context
+	outputCount int
+	router      route.RouteIO
+	logger      *slog.Logger
+}
+
+func (m *MockModule) Id() string {
+	return m.config.Id
+}
+
+func (m *MockModule) Output(context.Context, any) error {
+	m.outputCount += 1
+	return nil
+}
+
+func (m *MockModule) Run(ctx context.Context) error {
+	m.ctx = ctx
+	<-m.ctx.Done()
+	return nil
+}
+
+func (m *MockModule) Type() string {
+	return m.config.Type
+}
+
+func init() {
+	module.RegisterModule(module.ModuleRegistration{
+		Type: "mock.counter",
+		New: func(config config.ModuleConfig) (module.Module, error) {
+
+			return &MockModule{config: config, logger: slog.Default()}, nil
+		},
+	})
+}
+
+func TestNewRouter(t *testing.T) {
+	routerConfig := config.Config{
+		Modules: []config.ModuleConfig{
+			{
+				Id:   "mock",
+				Type: "mock.counter",
+			},
+		},
+	}
+
+	_, moduleErrors, routeErrors := showbridge.NewRouter(routerConfig)
+
+	if moduleErrors != nil {
+		t.Fatalf("router should not have returned any module errors: %v", moduleErrors)
+	}
+
+	if routeErrors != nil {
+		t.Fatalf("router should not have returned any route errors: %v", routeErrors)
+	}
+}
+
+func TestNewRouterUnknownModuleType(t *testing.T) {
+	routerConfig := config.Config{
+		Modules: []config.ModuleConfig{
+			{
+				Id:   "mock",
+				Type: "asd.fjlkj23oiu4ksldj",
+			},
+		},
+	}
+
+	_, moduleErrors, _ := showbridge.NewRouter(routerConfig)
+
+	if moduleErrors == nil {
+		t.Fatalf("router should have returned 'unknown module' module errors")
+	}
+}
+
+func TestNewRouterDuplicateModuleId(t *testing.T) {
+	routerConfig := config.Config{
+		Modules: []config.ModuleConfig{
+			{
+				Id:   "mock",
+				Type: "mock.counter",
+			},
+			{
+				Id:   "mock",
+				Type: "mock.counter",
+			},
+		},
+	}
+
+	_, moduleErrors, _ := showbridge.NewRouter(routerConfig)
+
+	if moduleErrors == nil {
+		t.Fatalf("router should have returned 'duplicate id' module error")
+	}
+}
+
+func TestRouterInputSingleRoute(t *testing.T) {
+	routerConfig := config.Config{
+		Modules: []config.ModuleConfig{
+			{
+				Id:   "mock",
+				Type: "mock.counter",
+			},
+		},
+		Routes: []config.RouteConfig{
+			{
+				Input:  "mock",
+				Output: "mock",
+			},
+		},
+	}
+
+	router, moduleErrors, routeErrors := showbridge.NewRouter(routerConfig)
+
+	if moduleErrors != nil {
+		t.Fatalf("router should not have returned any module errors: %v", moduleErrors)
+	}
+
+	if routeErrors != nil {
+		t.Fatalf("router should not have returned any route errors: %v", routeErrors)
+	}
+
+	routerRunner := sync.WaitGroup{}
+
+	routerRunner.Go(func() {
+		router.Run(t.Context())
+	})
+
+	time.Sleep(time.Second * 1)
+
+	defer router.Stop()
+
+	mockModuleInputCount := 3
+	for i := range mockModuleInputCount {
+		aRouteFound, routingErrors := router.HandleInput(t.Context(), "mock", fmt.Sprintf("test %d", i))
+
+		if routingErrors != nil {
+			t.Fatalf("router should not have encountered routing errors")
+		}
+
+		if !aRouteFound {
+			t.Fatalf("router should have found a valid route for the input")
+		}
+	}
+
+	for _, moduleInstance := range router.ModuleInstances {
+		if moduleInstance.Id() == "mock" {
+			mockModuleInstance, ok := moduleInstance.(*MockModule)
+			if !ok {
+				t.Fatalf("couldn't get mock module")
+			}
+
+			if mockModuleInstance.outputCount != mockModuleInputCount {
+				t.Fatalf("mock module output count did not matched expected: %d got: %d", mockModuleInputCount, mockModuleInstance.outputCount)
+			}
+		}
+	}
+}
+
+func TestRouterInputMultipleRoutes(t *testing.T) {
+	routerConfig := config.Config{
+		Modules: []config.ModuleConfig{
+			{
+				Id:   "mock",
+				Type: "mock.counter",
+			},
+		},
+		Routes: []config.RouteConfig{
+			{
+				Input:  "mock",
+				Output: "mock",
+			},
+			{
+				Input:  "mock",
+				Output: "mock",
+			},
+			{
+				Input:  "mock",
+				Output: "mock",
+			},
+		},
+	}
+
+	router, moduleErrors, routeErrors := showbridge.NewRouter(routerConfig)
+
+	if moduleErrors != nil {
+		t.Fatalf("router should not have returned any module errors: %v", moduleErrors)
+	}
+
+	if routeErrors != nil {
+		t.Fatalf("router should not have returned any route errors: %v", routeErrors)
+	}
+
+	routerRunner := sync.WaitGroup{}
+
+	routerRunner.Go(func() {
+		router.Run(t.Context())
+	})
+	time.Sleep(time.Second * 1)
+
+	defer router.Stop()
+
+	mockModuleInputCount := 3
+	for i := range mockModuleInputCount {
+		aRouteFound, routingErrors := router.HandleInput(t.Context(), "mock", fmt.Sprintf("test %d", i))
+
+		if routingErrors != nil {
+			t.Fatalf("router should not have encountered routing errors")
+		}
+
+		if !aRouteFound {
+			t.Fatalf("router should have found a valid route for the input")
+		}
+	}
+
+	for _, moduleInstance := range router.ModuleInstances {
+		if moduleInstance.Id() == "mock" {
+			mockModuleInstance, ok := moduleInstance.(*MockModule)
+			if !ok {
+				t.Fatalf("couldn't get mock module")
+			}
+
+			if mockModuleInstance.outputCount != len(router.RouteInstances)*mockModuleInputCount {
+				t.Fatalf("mock module output count did not matched expected: %d got: %d", len(router.RouteInstances)*mockModuleInputCount, mockModuleInstance.outputCount)
+			}
+			break
+		}
+	}
+}
+
+func TestRouterInputMultipleModules(t *testing.T) {
+	routerConfig := config.Config{
+		Modules: []config.ModuleConfig{
+			{
+				Id:   "mock1",
+				Type: "mock.counter",
+			},
+			{
+				Id:   "mock2",
+				Type: "mock.counter",
+			},
+		},
+		Routes: []config.RouteConfig{
+			{
+				Input:  "mock1",
+				Output: "mock1",
+			},
+			{
+				Input:  "mock2",
+				Output: "mock2",
+			},
+		},
+	}
+
+	router, moduleErrors, routeErrors := showbridge.NewRouter(routerConfig)
+
+	if moduleErrors != nil {
+		t.Fatalf("router should not have returned any module errors: %v", moduleErrors)
+	}
+
+	if routeErrors != nil {
+		t.Fatalf("router should not have returned any route errors: %v", routeErrors)
+	}
+
+	routerRunner := sync.WaitGroup{}
+
+	routerRunner.Go(func() {
+		router.Run(t.Context())
+	})
+
+	time.Sleep(time.Second * 1)
+
+	defer router.Stop()
+
+	mock1ModuleInputCount := 3
+	for i := range mock1ModuleInputCount {
+		aRouteFound, routingErrors := router.HandleInput(t.Context(), "mock1", fmt.Sprintf("test %d", i))
+
+		if routingErrors != nil {
+			t.Fatalf("router should not have encountered routing errors")
+		}
+
+		if !aRouteFound {
+			t.Fatalf("router should have found a valid route for the input")
+		}
+	}
+
+	mock2ModuleInputCount := 2
+	for i := range mock2ModuleInputCount {
+		aRouteFound, routingErrors := router.HandleInput(t.Context(), "mock2", fmt.Sprintf("test %d", i))
+
+		if routingErrors != nil {
+			t.Fatalf("router should not have encountered routing errors")
+		}
+
+		if !aRouteFound {
+			t.Fatalf("router should have found a valid route for the input")
+		}
+	}
+
+	for _, moduleInstance := range router.ModuleInstances {
+		if moduleInstance.Id() == "mock1" {
+			mockModuleInstance, ok := moduleInstance.(*MockModule)
+			if !ok {
+				t.Fatalf("couldn't get mock module")
+			}
+
+			if mockModuleInstance.outputCount != mock1ModuleInputCount {
+				t.Fatalf("mock module output count did not matched expected: %d got: %d", mock1ModuleInputCount, mockModuleInstance.outputCount)
+			}
+			break
+		}
+		if moduleInstance.Id() == "mock2" {
+			mockModuleInstance, ok := moduleInstance.(*MockModule)
+			if !ok {
+				t.Fatalf("couldn't get mock module")
+			}
+
+			if mockModuleInstance.outputCount != mock2ModuleInputCount {
+				t.Fatalf("mock module output count did not matched expected: %d got: %d", mock2ModuleInputCount, mockModuleInstance.outputCount)
+			}
+			break
+		}
+	}
+}