Merge pull request #179 from jwetzell/fix/in-out-framer

make sure to use a distinct framer for both directions of communication

.github/workflows/release-showbridge.yaml

@@ -211,7 +211,8 @@ jobs:
           files: showbridge_*.tar.gz
   docker:
     runs-on: ubuntu-latest
-    needs: linux_amd64
+    needs: setup_release
+    if: github.ref_type == 'tag' && github.event_name != 'workflow_dispatch'
     steps:
       - name: Check out repository
         uses: actions/checkout@1af3b93b6815bc44a9784bd300feb67ff0d1eeb3 # v6.0.0
@@ -220,24 +221,23 @@ jobs:
         uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v4.0.0
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0
+        uses: docker/setup-buildx-action@d7f5e7f509e45cec5c76c4d5afdd7de93d0b3df5 # v4.1.0
 
       - name: Login to Docker Hub
-        uses: docker/login-action@4907a6ddec9925e35a0a9e82d7399ccc52663121 # v4.1.0
+        uses: docker/login-action@650006c6eb7dba73a995cc03b0b2d7f5ca915bee # v4.2.0
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_TOKEN }}
 
       - name: Setup Docker metadata
         id: meta
-        uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v6.0.0
+        uses: docker/metadata-action@80c7e94dd9b9319bd5eb7a0e0fe9291e23a2a2e9 # v6.1.0
         with:
           images: |
             jwetzell/showbridge
 
       - name: Build and push
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
-        if: github.ref_type == 'tag'
+        uses: docker/build-push-action@f9f3042f7e2789586610d6e8b85c8f03e5195baf # v7.2.0
         with:
           push: true
           context: ./

cmd/showbridge/main.go

@@ -106,12 +106,12 @@ func readConfig(configPath string) (config.Config, error) {
 
 	err = schema.ApplyDefaults(&yamlMap)
 	if err != nil {
-		return config.Config{}, err
+		return config.Config{}, fmt.Errorf("failed to apply defaults: %w", err)
 	}
 
 	err = schema.ValidateConfig(yamlMap)
 	if err != nil {
-		return config.Config{}, err
+		return config.Config{}, fmt.Errorf("failed to validate config: %w", err)
 	}
 
 	validatedConfigBytes, err := json.Marshal(yamlMap)

config.smoke.yaml

@@ -57,14 +57,25 @@ routes:
     processors:
       - type: string.create
         params:
-          template: "hello"
+          template: "wsecho: {{.Payload.UnixMilli}}"
       - type: module.output
         params:
           module: wsecho
+  - input: ticker
+    processors:
+      - type: string.create
+        params:
+          template: "tcpout: {{.Payload.UnixMilli}}"
       - type: string.encode
       - type: module.output
         params:
           module: tcpout
+  - input: ticker
+    processors:
+      - type: string.create
+        params:
+          template: "udpout: {{.Payload.UnixMilli}}"
+      - type: string.encode
       - type: module.output
         params:
           module: udpout
@@ -74,14 +85,15 @@ routes:
         params:
           url: "http://localhost:3000/echo"
           method: "GET"
+      - type: struct.field.get
+        params:
+          name: "Body"
+      - type: string.decode
       - type: debug.log
   - input: tcpin
     processors:
       - type: string.decode
       - type: debug.log
-  - input: httpin
-    processors:
-      - type: debug.log
   - input: udpin
     processors:
       - type: string.decode
@@ -98,12 +110,25 @@ routes:
       - type: debug.log
   - input: natsin
     processors:
+      - type: struct.field.get
+        params:
+          name: "Data"
+      - type: string.decode
       - type: debug.log
+  - input: httpin
+    processors:
+      - type: http.response.create
+        params:
+          status: 200
+          bodyTemplate: "Hello, World!"
+      - type: module.output
+        params:
+          module: httpin
   - input: ticker
     processors:
       - type: string.create
         params:
-          template: "{{.Payload}}"
+          template: "natsout: {{.Payload}}"
       - type: pubsub.publish
         params:
           module: natsout

go.mod

@@ -10,16 +10,17 @@ require (
 	github.com/extism/go-sdk v1.7.1
 	github.com/google/jsonschema-go v0.4.3
 	github.com/gorilla/websocket v1.5.3
-	github.com/jwetzell/artnet-go v0.2.3
+	github.com/jwetzell/artnet-go v0.3.0
 	github.com/jwetzell/free-d-go v0.2.0
 	github.com/jwetzell/osc-go v0.3.0
 	github.com/jwetzell/psn-go v0.3.0
-	github.com/nats-io/nats-server/v2 v2.14.0
+	github.com/nats-io/nats-server/v2 v2.14.1
 	github.com/nats-io/nats.go v1.52.0
 	github.com/redis/go-redis/v9 v9.19.0
 	github.com/urfave/cli/v3 v3.9.0
 	gitlab.com/gomidi/midi/v2 v2.3.23
-	go.bug.st/serial v1.7.0
+	go.bug.st/serial v1.7.1
+	golang.org/x/time v0.15.0
 	modernc.org/quickjs v0.18.2
 	modernc.org/sqlite v1.50.1
 	sigs.k8s.io/yaml v1.6.0
@@ -66,7 +67,6 @@ require (
 	golang.org/x/net v0.54.0 // indirect
 	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.44.0 // indirect
-	golang.org/x/time v0.15.0 // indirect
 	golang.org/x/tools v0.44.0 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/hraban/opus.v2 v2.0.0-20230925203106-0188a62cb302 // indirect

go.sum

@@ -52,8 +52,8 @@ github.com/ianlancetaylor/demangle v0.0.0-20260505044615-1ff4bf46051f h1:NW3E2QS
 github.com/ianlancetaylor/demangle v0.0.0-20260505044615-1ff4bf46051f/go.mod h1:gx7rwoVhcfuVKG5uya9Hs3Sxj7EIvldVofAWIUtGouw=
 github.com/icholy/digest v1.1.0 h1:HfGg9Irj7i+IX1o1QAmPfIBNu/Q5A5Tu3n/MED9k9H4=
 github.com/icholy/digest v1.1.0/go.mod h1:QNrsSGQ5v7v9cReDI0+eyjsXGUoRSUZQHeQ5C4XLa0Y=
-github.com/jwetzell/artnet-go v0.2.3 h1:yPDckAFFSbJnpjMX5Y+tFh8xiP2zH3iqZxrXZrJg91g=
-github.com/jwetzell/artnet-go v0.2.3/go.mod h1:gli97Z32a0kMkZ6taoTiK7/lqHcF/dhiGjGJdx/PxqA=
+github.com/jwetzell/artnet-go v0.3.0 h1:MDHElOSbpgmIbYGHO9SoxsT7ynrYtV5/UHGT82geHk0=
+github.com/jwetzell/artnet-go v0.3.0/go.mod h1:zqRsSsrd6rPwJacwscPteDqGIqevjZ3Z6dpp092VJ1w=
 github.com/jwetzell/free-d-go v0.2.0 h1:8WQW4du8Sf3d18aUzk9n5jZtHE6WdDQHDFjkFffvycc=
 github.com/jwetzell/free-d-go v0.2.0/go.mod h1:yHuRy51NeDfN26eJ6wzZgys8qyQiS7hkYILw8yF+UJ4=
 github.com/jwetzell/osc-go v0.3.0 h1:z75TxuQSEmdcmZ56OAepkDa3m88SdZh//3m4nBb/XZI=
@@ -74,8 +74,8 @@ github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clg
 github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
 github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
 github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
-github.com/nats-io/nats-server/v2 v2.14.0 h1:+8q0HrDFotwLLcGH/legOEOnowunhK+aZ4GYBIWpQlM=
-github.com/nats-io/nats-server/v2 v2.14.0/go.mod h1:ImVUUDvfClJbb6cuJQRc1VmgDCXKM5ds0OoiG9MVOKo=
+github.com/nats-io/nats-server/v2 v2.14.1 h1:wXs/a5fw9Hzm3CvuzLxGeIwpjPulSa7gMT3eSuhGkcg=
+github.com/nats-io/nats-server/v2 v2.14.1/go.mod h1:4N17zLpuS7WMbG8T9gsE2B7z9hC9PraPyulVBfpK6nU=
 github.com/nats-io/nats.go v1.52.0 h1:n3avV4VBsCgsdwh71TppsTwtv+QdPs7ntSKM8qJLGsc=
 github.com/nats-io/nats.go v1.52.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
 github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
@@ -118,8 +118,8 @@ github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
 github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
 gitlab.com/gomidi/midi/v2 v2.3.23 h1:P8NxV4EzV9c+BjpwTeB+G/qa+Xdq/UTazS2fKxY0O0g=
 gitlab.com/gomidi/midi/v2 v2.3.23/go.mod h1:jDpP4O4skYi+7iVwt6Zyp18bd2M4hkjtMuw2cmgKgfw=
-go.bug.st/serial v1.7.0 h1:Svs5xR0s4JF+KIg+1d9QE4iXqEIJvTr5vESFUjSu7mE=
-go.bug.st/serial v1.7.0/go.mod h1:PZfOSahry47TWtrd0tvoKZvmVa7FGrc/1xG9q8Ho3xU=
+go.bug.st/serial v1.7.1 h1:5aP8wYL0UjEYOVs3oPAGscjaSfRQLHtCvBFXNN/rwtc=
+go.bug.st/serial v1.7.1/go.mod h1:d0MmS16Qt9b1m06yoYRNUXhRRTJV5Qg2S5EKqQtnayQ=
 go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g=
 go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=

internal/common/module.go

@@ -17,8 +17,8 @@ type OutputModule interface {
 }
 
 type KeyValueModule interface {
-	Get(key string) (any, error)
-	Set(key string, value any) error
+	Get(ctx context.Context, key string) (any, error)
+	Set(ctx context.Context, key string, value any) error
 }
 
 type DatabaseModule interface {

internal/framer/framer.go

@@ -10,15 +10,15 @@ type Framer interface {
 func GetFramer(framingType string) Framer {
 	switch framingType {
 	case "CR":
-		return NewByteSeparatorFramer([]byte{'\r'})
+		return newByteSeparatorFramer([]byte{'\r'})
 	case "LF":
-		return NewByteSeparatorFramer([]byte{'\n'})
+		return newByteSeparatorFramer([]byte{'\n'})
 	case "CRLF":
-		return NewByteSeparatorFramer([]byte{'\r', '\n'})
+		return newByteSeparatorFramer([]byte{'\r', '\n'})
 	case "SLIP":
-		return NewSlipFramer()
+		return newSlipFramer()
 	case "RAW":
-		return NewRawFramer()
+		return newRawFramer()
 	default:
 		return nil
 	}

internal/framer/raw.go

@@ -1,23 +1,23 @@
 package framer
 
-type RawFramer struct{}
+type rawFramer struct{}
 
-func NewRawFramer() *RawFramer {
-	return &RawFramer{}
+func newRawFramer() *rawFramer {
+	return &rawFramer{}
 }
 
-func (rf *RawFramer) Decode(data []byte) [][]byte {
+func (rf *rawFramer) Decode(data []byte) [][]byte {
 	return [][]byte{data}
 }
 
-func (rf *RawFramer) Encode(data []byte) []byte {
+func (rf *rawFramer) Encode(data []byte) []byte {
 	return data
 }
 
-func (rf *RawFramer) Clear() {
+func (rf *rawFramer) Clear() {
 	// NOTE(jwetzell): no internal state to clear
 }
 
-func (rf *RawFramer) Buffer() []byte {
+func (rf *rawFramer) Buffer() []byte {
 	return []byte{}
 }

internal/framer/separator.go

@@ -4,16 +4,16 @@ import (
 	"bytes"
 )
 
-type ByteSeparatorFramer struct {
+type byteSeparatorFramer struct {
 	buffer    []byte
 	separator []byte
 }
 
-func NewByteSeparatorFramer(separator []byte) *ByteSeparatorFramer {
-	return &ByteSeparatorFramer{separator: separator, buffer: []byte{}}
+func newByteSeparatorFramer(separator []byte) *byteSeparatorFramer {
+	return &byteSeparatorFramer{separator: separator, buffer: []byte{}}
 }
 
-func (bsf *ByteSeparatorFramer) Decode(data []byte) [][]byte {
+func (bsf *byteSeparatorFramer) Decode(data []byte) [][]byte {
 	messages := [][]byte{}
 
 	bsf.buffer = append(bsf.buffer, data...)
@@ -28,14 +28,14 @@ func (bsf *ByteSeparatorFramer) Decode(data []byte) [][]byte {
 	return messages
 }
 
-func (bsf *ByteSeparatorFramer) Encode(data []byte) []byte {
+func (bsf *byteSeparatorFramer) Encode(data []byte) []byte {
 	return append(data, bsf.separator...)
 }
 
-func (bsf *ByteSeparatorFramer) Clear() {
+func (bsf *byteSeparatorFramer) Clear() {
 	bsf.buffer = []byte{}
 }
 
-func (bsf *ByteSeparatorFramer) Buffer() []byte {
+func (bsf *byteSeparatorFramer) Buffer() []byte {
 	return bsf.buffer
 }

internal/framer/slip.go

@@ -1,14 +1,14 @@
 package framer
 
-type SlipFramer struct {
+type slipFramer struct {
 	buffer []byte
 }
 
-func NewSlipFramer() *SlipFramer {
-	return &SlipFramer{buffer: []byte{}}
+func newSlipFramer() *slipFramer {
+	return &slipFramer{buffer: []byte{}}
 }
 
-func (sf *SlipFramer) Decode(data []byte) [][]byte {
+func (sf *slipFramer) Decode(data []byte) [][]byte {
 	messages := [][]byte{}
 
 	END := byte(0xc0)
@@ -49,7 +49,7 @@ func (sf *SlipFramer) Decode(data []byte) [][]byte {
 	return messages
 }
 
-func (sf *SlipFramer) Encode(data []byte) []byte {
+func (sf *slipFramer) Encode(data []byte) []byte {
 	END := byte(0xc0)
 	ESC := byte(0xdb)
 	ESC_END := byte(0xdc)
@@ -72,10 +72,10 @@ func (sf *SlipFramer) Encode(data []byte) []byte {
 	return encodedBytes
 }
 
-func (sf *SlipFramer) Clear() {
+func (sf *slipFramer) Clear() {
 	sf.buffer = []byte{}
 }
 
-func (sf *SlipFramer) Buffer() []byte {
+func (sf *slipFramer) Buffer() []byte {
 	return sf.buffer
 }

internal/module/db-sqlite.go

@@ -22,8 +22,10 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"dsn": {
-					Type:      "string",
-					MinLength: new(1),
+					Title:       "Data Source Name",
+					Description: "the data source name (DSN) for the SQLite database",
+					Type:        "string",
+					MinLength:   new(1),
 				},
 			},
 			Required:             []string{"dsn"},
@@ -76,20 +78,19 @@ func (dbs *DbSqlite) Start(ctx context.Context, inputHandler common.InputHandler
 	dbs.db = db
 	dbs.dbMu.Unlock()
 	<-dbs.ctx.Done()
+	dbs.logger.Debug("done")
 	return nil
 }
 
 func (dbs *DbSqlite) Stop() {
 	if dbs.cancel != nil {
-		dbs.cancel()
+		defer dbs.cancel()
 	}
 	dbs.dbMu.Lock()
 	defer dbs.dbMu.Unlock()
 	if dbs.db != nil {
 		dbs.db.Close()
-		dbs.db = nil
 	}
-	dbs.logger.Debug("done")
 }
 
 func (dbs *DbSqlite) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {

internal/module/http-server.go

@@ -24,10 +24,11 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"port": {
-					Title:   "Port",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1024),
-					Maximum: jsonschema.Ptr[float64](65535),
+					Title:       "Port",
+					Description: "the port for the HTTP server to listen on",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1024),
+					Maximum:     jsonschema.Ptr[float64](65535),
 				},
 			},
 			Required:             []string{"port"},
@@ -167,12 +168,13 @@ func (hs *HTTPServer) Start(ctx context.Context, inputHandler common.InputHandle
 	err := httpServer.ListenAndServe()
 	// TODO(jwetzell): handle server closed error differently
 	if err != nil {
-		if err.Error() != "http: Server closed" {
+		if !errors.Is(err, http.ErrServerClosed) {
 			return err
 		}
 	}
 
 	<-hs.ctx.Done()
+	hs.logger.Debug("done")
 	return nil
 }
 
@@ -200,7 +202,7 @@ func (hs *HTTPServer) Output(ctx context.Context, payload any) error {
 
 func (hs *HTTPServer) Stop() {
 	if hs.cancel != nil {
-		hs.cancel()
+		defer hs.cancel()
 	}
 	hs.serverMu.Lock()
 	defer hs.serverMu.Unlock()
@@ -209,7 +211,5 @@ func (hs *HTTPServer) Stop() {
 		hs.server.Shutdown(shutdownCtx)
 		shutdownCancel()
 		<-shutdownCtx.Done()
-		hs.server = nil
 	}
-	hs.logger.Debug("done")
 }

internal/module/midi-input.go

@@ -22,8 +22,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"port": {
-					Title: "Port",
-					Type:  "string",
+					Title:       "Port",
+					Description: "the name of the MIDI port to listen to",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"port"},
@@ -83,17 +84,16 @@ func (mi *MIDIInput) Start(ctx context.Context, inputHandler common.InputHandler
 	mi.stop = stop
 
 	<-mi.ctx.Done()
+	mi.logger.Debug("done")
 	return nil
 }
 
 func (mi *MIDIInput) Stop() {
 	if mi.cancel != nil {
-		mi.cancel()
+		defer mi.cancel()
 	}
 	if mi.stop != nil {
 		mi.stop()
-		mi.stop = nil
 	}
 	midi.CloseDriver()
-	mi.logger.Debug("done")
 }

internal/module/midi-output.go

@@ -24,8 +24,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"port": {
-					Title: "Port",
-					Type:  "string",
+					Title:       "Port",
+					Description: "the name of the MIDI port to send messages to",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"port"},
@@ -86,6 +87,7 @@ func (mo *MIDIOutput) Start(ctx context.Context, inputHandler common.InputHandle
 	mo.sendFuncMu.Unlock()
 
 	<-mo.ctx.Done()
+	mo.logger.Debug("done")
 	return nil
 }
 
@@ -107,8 +109,7 @@ func (mo *MIDIOutput) Output(ctx context.Context, payload any) error {
 
 func (mo *MIDIOutput) Stop() {
 	if mo.cancel != nil {
-		mo.cancel()
+		defer mo.cancel()
 	}
 	midi.CloseDriver()
-	mo.logger.Debug("done")
 }

internal/module/module.go

@@ -30,15 +30,36 @@ func RegisterModule(mod ModuleRegistration) {
 	moduleRegistryMu.Lock()
 	defer moduleRegistryMu.Unlock()
 
-	if _, ok := ModuleRegistry[string(mod.Type)]; ok {
+	if _, ok := moduleRegistry[string(mod.Type)]; ok {
 		panic(fmt.Sprintf("module already registered: %s", mod.Type))
 	}
-	ModuleRegistry[string(mod.Type)] = mod
+	moduleRegistry[string(mod.Type)] = mod
+}
+
+type ModuleRegistry map[string]ModuleRegistration
+
+func GetModuleRegistration(moduleType string) (ModuleRegistration, bool) {
+	moduleRegistryMu.RLock()
+	defer moduleRegistryMu.RUnlock()
+
+	mod, ok := moduleRegistry[moduleType]
+	return mod, ok
+}
+
+func GetModuleRegistrations() []ModuleRegistration {
+	moduleRegistryMu.RLock()
+	defer moduleRegistryMu.RUnlock()
+
+	registrations := make([]ModuleRegistration, 0, len(moduleRegistry))
+	for _, mod := range moduleRegistry {
+		registrations = append(registrations, mod)
+	}
+	return registrations
 }
 
 var (
 	moduleRegistryMu sync.RWMutex
-	ModuleRegistry   = make(map[string]ModuleRegistration)
+	moduleRegistry   = make(ModuleRegistry)
 )
 
 func CreateLogger(config config.ModuleConfig) *slog.Logger {

internal/module/mqtt-client.go

@@ -22,28 +22,33 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"broker": {
-					Title: "Broker URL",
-					Type:  "string",
+					Title:       "Broker URL",
+					Description: "the URL of the MQTT broker to connect to",
+					Type:        "string",
 				},
 				"topic": {
-					Title: "Topic",
-					Type:  "string",
+					Title:       "Topic",
+					Description: "an MQTT topic to subscribe to",
+					Type:        "string",
 				},
 				"clientId": {
-					Title: "Client ID",
-					Type:  "string",
+					Title:       "Client ID",
+					Description: "the client ID to use when connecting to the MQTT broker",
+					Type:        "string",
 				},
 				"qos": {
-					Title:   "QoS",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](0),
-					Maximum: jsonschema.Ptr[float64](2),
-					Default: json.RawMessage(`0`),
+					Title:       "QoS",
+					Description: "the QoS level to use when publishing messages",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](0),
+					Maximum:     jsonschema.Ptr[float64](2),
+					Default:     json.RawMessage(`0`),
 				},
 				"retained": {
-					Title:   "Retained",
-					Type:    "boolean",
-					Default: json.RawMessage(`false`),
+					Title:       "Retained",
+					Description: "whether to set the retained flag when publishing messages",
+					Type:        "boolean",
+					Default:     json.RawMessage(`false`),
 				},
 			},
 			Required:             []string{"broker", "topic", "clientId"},
@@ -152,6 +157,7 @@ func (mc *MQTTClient) Start(ctx context.Context, inputHandler common.InputHandle
 	mc.clientMu.Unlock()
 
 	<-mc.ctx.Done()
+	mc.logger.Debug("done")
 	return nil
 }
 
@@ -189,7 +195,5 @@ func (mc *MQTTClient) Stop() {
 	defer mc.clientMu.Unlock()
 	if mc.client != nil {
 		mc.client.Disconnect(250)
-		mc.client = nil
 	}
-	mc.logger.Debug("done")
 }

internal/module/nats-client.go

@@ -20,12 +20,14 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"url": {
-					Title: "NATS Server URL",
-					Type:  "string",
+					Title:       "NATS Server URL",
+					Description: "the URL of the NATS server to connect to",
+					Type:        "string",
 				},
 				"subject": {
-					Title: "Subject",
-					Type:  "string",
+					Title:       "Subject",
+					Description: "the subject to subscribe to",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"url", "subject"},
@@ -102,6 +104,7 @@ func (nc *NATSClient) Start(ctx context.Context, inputHandler common.InputHandle
 	nc.subMu.Unlock()
 
 	<-nc.ctx.Done()
+	nc.logger.Debug("done")
 	return nil
 }
 
@@ -135,13 +138,12 @@ func (nc *NATSClient) Publish(ctx context.Context, topic string, payload any) er
 
 func (nc *NATSClient) Stop() {
 	if nc.cancel != nil {
-		nc.cancel()
+		defer nc.cancel()
 	}
 	nc.subMu.Lock()
 	defer nc.subMu.Unlock()
 	if nc.sub != nil {
 		nc.sub.Unsubscribe()
-		nc.sub = nil
 	}
 
 	nc.clientMu.Lock()
@@ -150,7 +152,5 @@ func (nc *NATSClient) Stop() {
 		nc.client.Drain()
 		// TODO(jwetzell): setup closed callback to get when client is fully closed
 		nc.client.Close()
-		nc.client = nil
 	}
-	nc.logger.Debug("done")
 }

internal/module/nats-server.go

@@ -24,16 +24,18 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"ip": {
-					Title:   "IP",
-					Type:    "string",
-					Default: json.RawMessage(`"0.0.0.0"`),
+					Title:       "IP",
+					Description: "the IP address to bind the NATS server to",
+					Type:        "string",
+					Default:     json.RawMessage(`"0.0.0.0"`),
 				},
 				"port": {
-					Title:   "Port",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1024),
-					Maximum: jsonschema.Ptr[float64](65535),
-					Default: json.RawMessage(`4222`),
+					Title:       "Port",
+					Description: "the port for the NATS server to listen on",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1024),
+					Maximum:     jsonschema.Ptr[float64](65535),
+					Default:     json.RawMessage(`4222`),
 				},
 			},
 			Required:             []string{},
@@ -96,9 +98,10 @@ func (ns *NATSServer) Start(ctx context.Context, inputHandler common.InputHandle
 	ns.cancel = cancel
 
 	natsServer, err := server.NewServer(&server.Options{
-		Host:  ns.Ip,
-		Port:  ns.Port,
-		NoLog: true,
+		Host:   ns.Ip,
+		Port:   ns.Port,
+		NoLog:  true,
+		NoSigs: true,
 	})
 
 	if err != nil {
@@ -107,26 +110,26 @@ func (ns *NATSServer) Start(ctx context.Context, inputHandler common.InputHandle
 
 	ns.serverMu.Lock()
 	ns.server = natsServer
-	defer ns.serverMu.Unlock()
 	natsServer.Start()
 
 	if !natsServer.ReadyForConnections(5 * time.Second) {
 		return errors.New("nats.server failed to start")
 	}
+	ns.serverMu.Unlock()
 
 	<-ns.ctx.Done()
-
+	ns.logger.Debug("done")
 	return nil
 }
 
 func (ns *NATSServer) Stop() {
 	if ns.cancel != nil {
-		ns.cancel()
+		defer ns.cancel()
 	}
 	ns.serverMu.Lock()
 	defer ns.serverMu.Unlock()
 	if ns.server != nil {
 		ns.server.Shutdown()
+		ns.server.WaitForShutdown()
 	}
-	ns.logger.Debug("done")
 }

internal/module/psn-client.go

@@ -63,7 +63,7 @@ func (pc *PSNClient) Start(ctx context.Context, inputHandler common.InputHandler
 	pc.connMu.Unlock()
 
 	buffer := make([]byte, 2048)
-	for {
+	for pc.ctx.Err() == nil {
 		select {
 		case <-pc.ctx.Done():
 			return nil
@@ -99,17 +99,18 @@ func (pc *PSNClient) Start(ctx context.Context, inputHandler common.InputHandler
 			}
 		}
 	}
+	<-pc.ctx.Done()
+	pc.logger.Debug("done")
+	return nil
 }
 
 func (pc *PSNClient) Stop() {
 	if pc.cancel != nil {
-		pc.cancel()
+		defer pc.cancel()
 	}
 	pc.connMu.Lock()
 	defer pc.connMu.Unlock()
 	if pc.conn != nil {
 		pc.conn.Close()
-		pc.conn = nil
 	}
-	pc.logger.Debug("done")
 }

internal/module/redis-client.go

@@ -21,12 +21,16 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"host": {
-					Type: "string",
+					Title:       "Host",
+					Description: "the hostname or IP address of the Redis server to connect to",
+					Type:        "string",
 				},
 				"port": {
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1),
-					Maximum: jsonschema.Ptr[float64](65535),
+					Title:       "Port",
+					Description: "the port to use when connecting to the Redis server",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1),
+					Maximum:     jsonschema.Ptr[float64](65535),
 				},
 			},
 			Required:             []string{"host", "port"},
@@ -94,25 +98,24 @@ func (rc *RedisClient) Start(ctx context.Context, inputHandler common.InputHandl
 	rc.clientMu.Unlock()
 
 	<-rc.ctx.Done()
+	rc.logger.Debug("done")
 	return nil
 }
 
 func (rc *RedisClient) Stop() {
 	if rc.cancel != nil {
-		rc.cancel()
+		defer rc.cancel()
 	}
 	rc.clientMu.Lock()
 	defer rc.clientMu.Unlock()
 	if rc.client != nil {
 		rc.client.Close()
-		rc.client = nil
 	}
-	rc.logger.Debug("done")
 }
 
-func (rc *RedisClient) Get(key string) (any, error) {
+func (rc *RedisClient) Get(ctx context.Context, key string) (any, error) {
 	if rc.client != nil {
-		val, err := rc.client.Get(rc.ctx, key).Result()
+		val, err := rc.client.Get(ctx, key).Result()
 		if err != nil {
 			return nil, err
 		}
@@ -121,9 +124,9 @@ func (rc *RedisClient) Get(key string) (any, error) {
 	return nil, errors.New("redis.client not setup")
 }
 
-func (rc *RedisClient) Set(key string, value any) error {
+func (rc *RedisClient) Set(ctx context.Context, key string, value any) error {
 	if rc.client != nil {
-		status := rc.client.Set(rc.ctx, key, value, 0)
+		status := rc.client.Set(ctx, key, value, 0)
 		return status.Err()
 	}
 	return errors.New("redis.client not setup")

internal/module/serial-client.go

@@ -25,17 +25,20 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"port": {
-					Title: "Port",
-					Type:  "string",
+					Title:       "Port",
+					Description: "the name of the serial port to connect to",
+					Type:        "string",
 				},
 				"baudRate": {
-					Title: "Baud Rate",
-					Type:  "integer",
+					Title:       "Baud Rate",
+					Description: "the baud rate to use when connecting to the serial port",
+					Type:        "integer",
 				},
 				"framing": {
-					Title: "Framing Method",
-					Type:  "string",
-					Enum:  []any{"LF", "CR", "CRLF", "SLIP", "RAW"},
+					Title:       "Framing Method",
+					Description: "the method to use for framing messages on the serial port",
+					Type:        "string",
+					Enum:        []any{"LF", "CR", "CRLF", "SLIP", "RAW"},
 				},
 			},
 			Required:             []string{"port", "baudRate", "framing"},
@@ -53,12 +56,14 @@ func init() {
 				return nil, fmt.Errorf("serial.client framing error: %w", err)
 			}
 
-			framer := framer.GetFramer(framingMethodString)
+			inFramer := framer.GetFramer(framingMethodString)
 
-			if framer == nil {
+			if inFramer == nil {
 				return nil, fmt.Errorf("serial.client unknown framing method: %s", framingMethodString)
 			}
 
+			outFramer := framer.GetFramer(framingMethodString)
+
 			baudRateInt, err := params.GetInt("baudRate")
 			if err != nil {
 				return nil, fmt.Errorf("serial.client baudRate error: %w", err)
@@ -68,7 +73,7 @@ func init() {
 				BaudRate: baudRateInt,
 			}
 
-			return &SerialClient{config: config, Port: portString, Framer: framer, Mode: &mode, logger: CreateLogger(config)}, nil
+			return &SerialClient{config: config, Port: portString, inFramer: inFramer, outFramer: outFramer, Mode: &mode, logger: CreateLogger(config)}, nil
 		},
 	})
 }
@@ -78,7 +83,8 @@ type SerialClient struct {
 	ctx          context.Context
 	inputHandler common.InputHandler
 	Port         string
-	Framer       framer.Framer
+	inFramer     framer.Framer
+	outFramer    framer.Framer
 	Mode         *serial.Mode
 	port         serial.Port
 	logger       *slog.Logger
@@ -139,13 +145,13 @@ func (sc *SerialClient) Start(ctx context.Context, inputHandler common.InputHand
 					byteCount, err := sc.port.Read(buffer)
 
 					if err != nil {
-						sc.Framer.Clear()
+						sc.inFramer.Clear()
 						break READ
 					}
 
-					if sc.Framer != nil {
+					if sc.inFramer != nil {
 						if byteCount > 0 {
-							messages := sc.Framer.Decode(buffer[0:byteCount])
+							messages := sc.inFramer.Decode(buffer[0:byteCount])
 							for _, message := range messages {
 								if sc.inputHandler != nil {
 									sc.inputHandler(sc.ctx, sc.Id(), message)
@@ -159,6 +165,8 @@ func (sc *SerialClient) Start(ctx context.Context, inputHandler common.InputHand
 			}
 		}
 	}
+	<-sc.ctx.Done()
+	sc.logger.Debug("done")
 	return nil
 }
 
@@ -170,19 +178,25 @@ func (sc *SerialClient) Output(ctx context.Context, payload any) error {
 		return errors.New("serial.client can only output bytes")
 	}
 
-	_, err := sc.port.Write(sc.Framer.Encode(payloadBytes))
+	_, err := sc.port.Write(sc.outFramer.Encode(payloadBytes))
 	return err
 }
 
 func (sc *SerialClient) Stop() {
 	if sc.cancel != nil {
-		sc.cancel()
+		defer sc.cancel()
 	}
 	sc.portMu.Lock()
 	defer sc.portMu.Unlock()
 	if sc.port != nil {
 		sc.port.Close()
-		sc.port = nil
 	}
-	sc.logger.Debug("done")
+
+	if sc.inFramer != nil {
+		sc.inFramer.Clear()
+	}
+
+	if sc.outFramer != nil {
+		sc.outFramer.Clear()
+	}
 }

internal/module/sip-call-server.go

@@ -29,27 +29,31 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"ip": {
-					Title:   "IP",
-					Type:    "string",
-					Default: json.RawMessage(`"0.0.0.0"`),
+					Title:       "IP",
+					Description: "the IP address to bind the SIP server to",
+					Type:        "string",
+					Default:     json.RawMessage(`"0.0.0.0"`),
 				},
 				"port": {
-					Title:   "Port",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1024),
-					Maximum: jsonschema.Ptr[float64](65535),
-					Default: json.RawMessage(`5060`),
+					Title:       "Port",
+					Description: "the port for the SIP server to listen on",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1024),
+					Maximum:     jsonschema.Ptr[float64](65535),
+					Default:     json.RawMessage(`5060`),
 				},
 				"transport": {
-					Title:   "Transport",
-					Type:    "string",
-					Enum:    []any{"udp", "tcp", "ws", "udp4", "tcp4"},
-					Default: json.RawMessage(`"udp"`),
+					Title:       "Transport",
+					Description: "the transport protocol to use for the SIP server",
+					Type:        "string",
+					Enum:        []any{"udp", "tcp", "ws", "udp4", "tcp4"},
+					Default:     json.RawMessage(`"udp"`),
 				},
 				"userAgent": {
-					Title:   "User Agent",
-					Type:    "string",
-					Default: json.RawMessage(`"showbridge"`),
+					Title:       "User Agent",
+					Description: "the user agent string to use",
+					Type:        "string",
+					Default:     json.RawMessage(`"showbridge"`),
 				},
 			},
 			Required:             []string{},
@@ -168,6 +172,7 @@ func (scs *SIPCallServer) Start(ctx context.Context, inputHandler common.InputHa
 	}
 
 	<-scs.ctx.Done()
+	scs.logger.Debug("done")
 	return nil
 }
 
@@ -251,12 +256,11 @@ func (scs *SIPCallServer) Output(ctx context.Context, payload any) error {
 
 func (scs *SIPCallServer) Stop() {
 	if scs.cancel != nil {
-		scs.cancel()
+		defer scs.cancel()
 	}
 	scs.uaMu.Lock()
 	defer scs.uaMu.Unlock()
 	if scs.ua != nil {
 		scs.ua.Close()
 	}
-	scs.logger.Debug("done")
 }

internal/module/sip-dtmf-server.go

@@ -30,33 +30,38 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"ip": {
-					Title:   "IP",
-					Type:    "string",
-					Default: json.RawMessage(`"0.0.0.0"`),
+					Title:       "IP",
+					Description: "the IP address to bind the SIP server to",
+					Type:        "string",
+					Default:     json.RawMessage(`"0.0.0.0"`),
 				},
 				"port": {
-					Title:   "Port",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1024),
-					Maximum: jsonschema.Ptr[float64](65535),
-					Default: json.RawMessage(`5060`),
+					Title:       "Port",
+					Description: "the port for the SIP server to listen on",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1024),
+					Maximum:     jsonschema.Ptr[float64](65535),
+					Default:     json.RawMessage(`5060`),
 				},
 				"transport": {
-					Title:   "Transport",
-					Type:    "string",
-					Enum:    []any{"udp", "tcp", "ws", "udp4", "tcp4"},
-					Default: json.RawMessage(`"udp"`),
+					Title:       "Transport",
+					Description: "the transport protocol to use for the SIP server",
+					Type:        "string",
+					Enum:        []any{"udp", "tcp", "ws", "udp4", "tcp4"},
+					Default:     json.RawMessage(`"udp"`),
 				},
 				"userAgent": {
-					Title:   "User Agent",
-					Type:    "string",
-					Default: json.RawMessage(`"showbridge"`),
+					Title:       "User Agent",
+					Description: "the user agent string to use",
+					Type:        "string",
+					Default:     json.RawMessage(`"showbridge"`),
 				},
 				"separator": {
-					Title:     "DTMF Separator",
-					Type:      "string",
-					MinLength: new(1),
-					MaxLength: new(1),
+					Title:       "DTMF Separator",
+					Description: "the DTMF character to use as a separator between DTMF digit groups",
+					Type:        "string",
+					MinLength:   new(1),
+					MaxLength:   new(1),
 				},
 			},
 			Required:             []string{"separator"},
@@ -190,6 +195,7 @@ func (sds *SIPDTMFServer) Start(ctx context.Context, inputHandler common.InputHa
 	}
 
 	<-sds.ctx.Done()
+	sds.logger.Debug("done")
 	return nil
 }
 
@@ -286,12 +292,11 @@ func (sds *SIPDTMFServer) Output(ctx context.Context, payload any) error {
 
 func (sds *SIPDTMFServer) Stop() {
 	if sds.cancel != nil {
-		sds.cancel()
+		defer sds.cancel()
 	}
 	sds.uaMu.Lock()
 	defer sds.uaMu.Unlock()
 	if sds.ua != nil {
 		sds.ua.Close()
 	}
-	sds.logger.Debug("done")
 }

internal/module/tcp-client.go

@@ -23,19 +23,22 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"host": {
-					Title: "Host",
-					Type:  "string",
+					Title:       "Host",
+					Description: "the hostname or IP address of the TCP server to connect to",
+					Type:        "string",
 				},
 				"port": {
-					Title:   "Port",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1),
-					Maximum: jsonschema.Ptr[float64](65535),
+					Title:       "Port",
+					Description: "the port of the TCP server to connect to",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1),
+					Maximum:     jsonschema.Ptr[float64](65535),
 				},
 				"framing": {
-					Title: "Framing Method",
-					Type:  "string",
-					Enum:  []any{"LF", "CR", "CRLF", "SLIP", "RAW"},
+					Title:       "Framing Method",
+					Description: "the method used to frame messages over the TCP connection",
+					Type:        "string",
+					Enum:        []any{"LF", "CR", "CRLF", "SLIP", "RAW"},
 				},
 			},
 			Required:             []string{"host", "port", "framing"},
@@ -63,19 +66,22 @@ func init() {
 				return nil, fmt.Errorf("net.tcp.client framing error: %w", err)
 			}
 
-			framer := framer.GetFramer(framingMethodString)
+			inFramer := framer.GetFramer(framingMethodString)
 
-			if framer == nil {
+			if inFramer == nil {
 				return nil, fmt.Errorf("net.tcp.client unknown framing method: %s", framingMethodString)
 			}
-			return &TCPClient{framer: framer, Addr: addr, config: config, logger: CreateLogger(config)}, nil
+
+			outFramer := framer.GetFramer(framingMethodString)
+			return &TCPClient{inFramer: inFramer, outFramer: outFramer, Addr: addr, config: config, logger: CreateLogger(config)}, nil
 		},
 	})
 }
 
 type TCPClient struct {
 	config       config.ModuleConfig
-	framer       framer.Framer
+	inFramer     framer.Framer
+	outFramer    framer.Framer
 	conn         *net.TCPConn
 	ctx          context.Context
 	inputHandler common.InputHandler
@@ -100,11 +106,12 @@ func (tc *TCPClient) Start(ctx context.Context, inputHandler common.InputHandler
 	tc.ctx = moduleContext
 	tc.cancel = cancel
 
+CONNECT_RETRY:
 	for tc.ctx.Err() == nil {
 		err := tc.SetupConn()
 		if err != nil {
 			if tc.ctx.Err() != nil {
-				return nil
+				break CONNECT_RETRY
 			}
 			tc.logger.Error("connection error", "error", err.Error())
 			time.Sleep(time.Second * 2)
@@ -112,39 +119,41 @@ func (tc *TCPClient) Start(ctx context.Context, inputHandler common.InputHandler
 		}
 
 		buffer := make([]byte, 1024)
-		select {
-		case <-tc.ctx.Done():
-			return nil
-		default:
-		READ:
-			for {
-				select {
-				case <-tc.ctx.Done():
-					return nil
-				default:
-					byteCount, err := tc.conn.Read(buffer)
 
-					if err != nil {
-						tc.framer.Clear()
-						break READ
+	READ:
+		for tc.ctx.Err() == nil {
+			tc.conn.SetReadDeadline(time.Now().Add(time.Millisecond * 200))
+			byteCount, err := tc.conn.Read(buffer)
+
+			if err != nil {
+				if opErr, ok := err.(*net.OpError); ok {
+					//NOTE(jwetzell) we hit deadline
+					if opErr.Timeout() {
+						continue
 					}
+				}
+				if errors.Is(err, net.ErrClosed) {
+					break CONNECT_RETRY
+				}
+				break READ
+			}
 
-					if tc.framer != nil {
-						if byteCount > 0 {
-							messages := tc.framer.Decode(buffer[0:byteCount])
-							for _, message := range messages {
-								if tc.inputHandler != nil {
-									tc.inputHandler(tc.ctx, tc.Id(), message)
-								} else {
-									tc.logger.Error("input received but no input handler is configured")
-								}
-							}
+			if tc.inFramer != nil {
+				if byteCount > 0 {
+					messages := tc.inFramer.Decode(buffer[0:byteCount])
+					for _, message := range messages {
+						if tc.inputHandler != nil {
+							tc.inputHandler(tc.ctx, tc.Id(), message)
+						} else {
+							tc.logger.Error("input received but no input handler is configured")
 						}
 					}
 				}
 			}
 		}
 	}
+	<-tc.ctx.Done()
+	tc.logger.Debug("done")
 	return nil
 }
 
@@ -166,20 +175,25 @@ func (tc *TCPClient) Output(ctx context.Context, payload any) error {
 	if !ok {
 		return errors.New("net.tcp.client is only able to output bytes")
 	}
-	_, err := tc.conn.Write(tc.framer.Encode(payloadBytes))
+	_, err := tc.conn.Write(tc.inFramer.Encode(payloadBytes))
 	return err
 }
 
 func (tc *TCPClient) Stop() {
 	if tc.cancel != nil {
-		tc.cancel()
+		defer tc.cancel()
 	}
 	tc.connMu.Lock()
 	defer tc.connMu.Unlock()
 	if tc.conn != nil {
 		tc.conn.Close()
-		tc.conn = nil
 	}
-	tc.logger.Debug("done")
 
+	if tc.inFramer != nil {
+		tc.inFramer.Clear()
+	}
+
+	if tc.outFramer != nil {
+		tc.outFramer.Clear()
+	}
 }

internal/module/tcp-server.go

@@ -10,7 +10,6 @@ import (
 	"slices"
 	"strings"
 	"sync"
-	"syscall"
 	"time"
 
 	"github.com/google/jsonschema-go/jsonschema"
@@ -27,20 +26,23 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"ip": {
-					Title:   "IP",
-					Type:    "string",
-					Default: json.RawMessage(`"0.0.0.0"`),
+					Title:       "IP",
+					Description: "the IP address to bind the TCP server to",
+					Type:        "string",
+					Default:     json.RawMessage(`"0.0.0.0"`),
 				},
 				"port": {
-					Title:   "Port",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1024),
-					Maximum: jsonschema.Ptr[float64](65535),
+					Title:       "Port",
+					Description: "the port for the TCP server to listen on",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1024),
+					Maximum:     jsonschema.Ptr[float64](65535),
 				},
 				"framing": {
-					Title: "Framing Method",
-					Type:  "string",
-					Enum:  []any{"LF", "CR", "CRLF", "SLIP", "RAW"},
+					Title:       "Framing Method",
+					Description: "the method used to frame messages over the TCP connection",
+					Type:        "string",
+					Enum:        []any{"LF", "CR", "CRLF", "SLIP", "RAW"},
 				},
 			},
 			Required:             []string{"port", "framing"},
@@ -58,12 +60,14 @@ func init() {
 				return nil, fmt.Errorf("net.tcp.server framing error: %w", err)
 			}
 
-			framer := framer.GetFramer(framingMethodString)
+			inFramer := framer.GetFramer(framingMethodString)
 
-			if framer == nil {
+			if inFramer == nil {
 				return nil, fmt.Errorf("net.tcp.server unknown framing method: %s", framingMethodString)
 			}
 
+			outFramer := framer.GetFramer(framingMethodString)
+
 			ipString, err := params.GetString("ip")
 			if err != nil {
 				if errors.Is(err, config.ErrParamNotFound) {
@@ -77,24 +81,33 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
-			return &TCPServer{Framer: framer, Addr: addr, config: moduleConfig, logger: CreateLogger(moduleConfig)}, nil
+			return &TCPServer{inFramer: inFramer, outFramer: outFramer, framerType: framingMethodString, Addr: addr, config: moduleConfig, logger: CreateLogger(moduleConfig)}, nil
 		},
 	})
 }
 
+type tcpConnection struct {
+	conn   *net.TCPConn
+	framer framer.Framer
+}
+
 type TCPServer struct {
-	config        config.ModuleConfig
-	Addr          *net.TCPAddr
-	Framer        framer.Framer
-	ctx           context.Context
-	inputHandler  common.InputHandler
-	wg            sync.WaitGroup
-	connections   []*net.TCPConn
-	connectionsMu sync.RWMutex
-	logger        *slog.Logger
-	cancel        context.CancelFunc
-	listener      *net.TCPListener
-	listenerMu    sync.Mutex
+	config                config.ModuleConfig
+	Addr                  *net.TCPAddr
+	inFramer              framer.Framer
+	outFramer             framer.Framer
+	framerType            string
+	ctx                   context.Context
+	inputHandler          common.InputHandler
+	wg                    sync.WaitGroup
+	connections           []tcpConnection
+	connectionsMu         sync.RWMutex
+	logger                *slog.Logger
+	cancel                context.CancelFunc
+	listener              *net.TCPListener
+	listenerMu            sync.Mutex
+	connectionShutdownCtx context.Context
+	connectionShutdown    context.CancelFunc
 }
 
 func (ts *TCPServer) Id() string {
@@ -107,70 +120,43 @@ func (ts *TCPServer) Type() string {
 
 func (ts *TCPServer) handleClient(client *net.TCPConn) {
 	ts.connectionsMu.Lock()
-	ts.connections = append(ts.connections, client)
+	ts.connections = append(ts.connections, tcpConnection{conn: client, framer: framer.GetFramer(ts.framerType)})
 	ts.connectionsMu.Unlock()
 	ts.logger.Debug("connection accepted", "remoteAddr", client.RemoteAddr().String())
-	defer client.Close()
+	defer func() {
+		client.Close()
+		ts.connectionsMu.Lock()
+		for i := 0; i < len(ts.connections); i++ {
+			if ts.connections[i].conn == client {
+				ts.connections = slices.Delete(ts.connections, i, i+1)
+				break
+			}
+		}
+		ts.connectionsMu.Unlock()
+		ts.logger.Debug("connection closed", "remoteAddr", client.RemoteAddr().String())
+	}()
 
 	buffer := make([]byte, 1024)
-ClientRead:
-	for ts.ctx.Err() == nil {
-		select {
-		case <-ts.ctx.Done():
-			client.Close()
-			ts.connectionsMu.Lock()
-			for i := 0; i < len(ts.connections); i++ {
-				if ts.connections[i] == client {
-					ts.connections = slices.Delete(ts.connections, i, i+1)
-					break
+	for ts.ctx.Err() == nil && ts.connectionShutdownCtx.Err() == nil {
+		client.SetDeadline(time.Now().Add(time.Millisecond * 200))
+		byteCount, err := client.Read(buffer)
+		if err != nil {
+			if opErr, ok := err.(*net.OpError); ok {
+				//NOTE(jwetzell) we hit deadline
+				if opErr.Timeout() {
+					continue
 				}
 			}
-			ts.connectionsMu.Unlock()
-			return
-		default:
-			client.SetDeadline(time.Now().Add(time.Millisecond * 200))
-			byteCount, err := client.Read(buffer)
-
-			if err != nil {
-				if opErr, ok := err.(*net.OpError); ok {
-					//NOTE(jwetzell) we hit deadline
-					if opErr.Timeout() {
-						continue ClientRead
-					}
-					if errors.Is(opErr, syscall.ECONNRESET) {
-						ts.connectionsMu.Lock()
-						for i := 0; i < len(ts.connections); i++ {
-							if ts.connections[i] == client {
-								ts.connections = slices.Delete(ts.connections, i, i+1)
-								break
-							}
-						}
-						ts.logger.Debug("connection reset", "remoteAddr", client.RemoteAddr().String())
-						ts.connectionsMu.Unlock()
-					}
-				}
-
-				if err.Error() == "EOF" {
-					ts.connectionsMu.Lock()
-					for i := 0; i < len(ts.connections); i++ {
-						if ts.connections[i] == client {
-							ts.connections = slices.Delete(ts.connections, i, i+1)
-							break
-						}
-					}
-					ts.connectionsMu.Unlock()
-				}
-				return
-			}
-			if ts.Framer != nil {
-				if byteCount > 0 {
-					messages := ts.Framer.Decode(buffer[0:byteCount])
-					for _, message := range messages {
-						if ts.inputHandler != nil {
-							ts.inputHandler(ts.ctx, ts.Id(), message)
-						} else {
-							ts.logger.Error("input received but no input handler is configured")
-						}
+			break
+		}
+		if ts.inFramer != nil {
+			if byteCount > 0 {
+				messages := ts.inFramer.Decode(buffer[0:byteCount])
+				for _, message := range messages {
+					if ts.inputHandler != nil {
+						ts.inputHandler(ts.ctx, ts.Id(), message)
+					} else {
+						ts.logger.Error("input received but no input handler is configured")
 					}
 				}
 			}
@@ -185,6 +171,10 @@ func (ts *TCPServer) Start(ctx context.Context, inputHandler common.InputHandler
 	ts.ctx = moduleContext
 	ts.cancel = cancel
 
+	shutdownCtx, shutdownCancel := context.WithCancel(context.Background())
+	ts.connectionShutdownCtx = shutdownCtx
+	ts.connectionShutdown = shutdownCancel
+
 	listener, err := net.ListenTCP("tcp", ts.Addr)
 	if err != nil {
 		return err
@@ -198,6 +188,9 @@ AcceptLoop:
 	for ts.ctx.Err() == nil {
 		conn, err := listener.AcceptTCP()
 		if err != nil {
+			if errors.Is(err, net.ErrClosed) {
+				break AcceptLoop
+			}
 			select {
 			case <-ts.ctx.Done():
 				break AcceptLoop
@@ -211,6 +204,8 @@ AcceptLoop:
 		}
 	}
 	ts.wg.Done()
+	<-ts.ctx.Done()
+	ts.logger.Debug("done")
 	return nil
 }
 
@@ -221,15 +216,21 @@ func (ts *TCPServer) Output(ctx context.Context, payload any) error {
 		return errors.New("net.tcp.server is only able to output bytes")
 	}
 	ts.connectionsMu.Lock()
+	defer ts.connectionsMu.Unlock()
 	var errorString strings.Builder
 
+	if ts.outFramer == nil {
+		return errors.New("no output framer configured")
+	}
+
+	outputBytes := ts.outFramer.Encode(payloadBytes)
+
 	for _, connection := range ts.connections {
-		_, err := connection.Write(payloadBytes)
+		_, err := connection.conn.Write(outputBytes)
 		if err != nil {
 			fmt.Fprintf(&errorString, "%s\n", err.Error())
 		}
 	}
-	ts.connectionsMu.Unlock()
 
 	if errorString.String() == "" {
 		return nil
@@ -239,14 +240,18 @@ func (ts *TCPServer) Output(ctx context.Context, payload any) error {
 
 func (ts *TCPServer) Stop() {
 	if ts.cancel != nil {
-		ts.cancel()
+		defer ts.cancel()
 	}
+	if ts.connectionShutdown != nil {
+		ts.connectionShutdown()
+	}
+
 	ts.listenerMu.Lock()
 	defer ts.listenerMu.Unlock()
 	if ts.listener != nil {
 		ts.listener.Close()
-		ts.listener = nil
 	}
+	ts.logger.Debug("waiting for connections to close")
 	ts.wg.Wait()
-	ts.logger.Debug("done")
+	ts.logger.Debug("all connections closed")
 }

internal/module/test/db-sqlite_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestDbSqliteFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["db.sqlite"]
+	registration, ok := module.GetModuleRegistration("db.sqlite")
 	if !ok {
 		t.Fatalf("db.sqlite module not registered")
 	}
@@ -58,7 +58,7 @@ func TestGoodDbSqlite(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["db.sqlite"]
+			registration, ok := module.GetModuleRegistration("db.sqlite")
 			if !ok {
 				t.Fatalf("db.sqlite module not registered")
 			}
@@ -107,7 +107,7 @@ func TestBadDbSqlite(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["db.sqlite"]
+			registration, ok := module.GetModuleRegistration("db.sqlite")
 			if !ok {
 				t.Fatalf("db.sqlite module not registered")
 			}

internal/module/test/http-server_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestHTTPServerFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["http.server"]
+	registration, ok := module.GetModuleRegistration("http.server")
 	if !ok {
 		t.Fatalf("http.server module not registered")
 	}
@@ -52,7 +52,7 @@ func TestGoodHTTPServer(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["http.server"]
+			registration, ok := module.GetModuleRegistration("http.server")
 			if !ok {
 				t.Fatalf("http.server module not registered")
 			}
@@ -102,7 +102,7 @@ func TestBadHTTPServer(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["http.server"]
+			registration, ok := module.GetModuleRegistration("http.server")
 			if !ok {
 				t.Fatalf("http.server module not registered")
 			}

internal/module/test/midi-input_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestMIDIInputFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["midi.input"]
+	registration, ok := module.GetModuleRegistration("midi.input")
 	if !ok {
 		t.Fatalf("midi.input module not registered")
 	}
@@ -55,7 +55,7 @@ func TestBadMIDIInput(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["midi.input"]
+			registration, ok := module.GetModuleRegistration("midi.input")
 			if !ok {
 				t.Fatalf("midi.input module not registered")
 			}

internal/module/test/midi-output_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestMIDIOutputFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["midi.output"]
+	registration, ok := module.GetModuleRegistration("midi.output")
 	if !ok {
 		t.Fatalf("midi.output module not registered")
 	}
@@ -55,7 +55,7 @@ func TestBadMIDIOutput(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["midi.output"]
+			registration, ok := module.GetModuleRegistration("midi.output")
 			if !ok {
 				t.Fatalf("midi.output module not registered")
 			}

internal/module/test/mqtt-client_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestMQTTClientFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["mqtt.client"]
+	registration, ok := module.GetModuleRegistration("mqtt.client")
 	if !ok {
 		t.Fatalf("mqtt.client module not registered")
 	}
@@ -98,7 +98,7 @@ func TestBadMQTTClient(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["mqtt.client"]
+			registration, ok := module.GetModuleRegistration("mqtt.client")
 			if !ok {
 				t.Fatalf("mqtt.client module not registered")
 			}

internal/module/test/nats-client_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestNATSClientFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["nats.client"]
+	registration, ok := module.GetModuleRegistration("nats.client")
 	if !ok {
 		t.Fatalf("nats.client module not registered")
 	}
@@ -76,7 +76,7 @@ func TestBadNATSClient(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["nats.client"]
+			registration, ok := module.GetModuleRegistration("nats.client")
 			if !ok {
 				t.Fatalf("nats.client module not registered")
 			}

internal/module/test/nats-server_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestNATSServerFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["nats.server"]
+	registration, ok := module.GetModuleRegistration("nats.server")
 	if !ok {
 		t.Fatalf("nats.server module not registered")
 	}
@@ -53,7 +53,7 @@ func TestGoodNATSServer(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["nats.server"]
+			registration, ok := module.GetModuleRegistration("nats.server")
 			if !ok {
 				t.Fatalf("nats.server module not registered")
 			}
@@ -100,7 +100,7 @@ func TestBadNATSServer(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["nats.server"]
+			registration, ok := module.GetModuleRegistration("nats.server")
 			if !ok {
 				t.Fatalf("nats.server module not registered")
 			}

internal/module/test/psn-client_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestPSNClientFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["psn.client"]
+	registration, ok := module.GetModuleRegistration("psn.client")
 	if !ok {
 		t.Fatalf("psn.client module not registered")
 	}
@@ -41,7 +41,7 @@ func TestBadPSNClient(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["psn.client"]
+			registration, ok := module.GetModuleRegistration("psn.client")
 			if !ok {
 				t.Fatalf("psn.client module not registered")
 			}

internal/module/test/redis-client_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestRedisClientFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["redis.client"]
+	registration, ok := module.GetModuleRegistration("redis.client")
 	if !ok {
 		t.Fatalf("redis.client module not registered")
 	}
@@ -76,7 +76,7 @@ func TestBadRedisClient(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["redis.client"]
+			registration, ok := module.GetModuleRegistration("redis.client")
 			if !ok {
 				t.Fatalf("redis.client module not registered")
 			}

internal/module/test/serial-client_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestSerialClientFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["serial.client"]
+	registration, ok := module.GetModuleRegistration("serial.client")
 	if !ok {
 		t.Fatalf("serial.client module not registered")
 	}
@@ -85,7 +85,7 @@ func TestBadSerialClient(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["serial.client"]
+			registration, ok := module.GetModuleRegistration("serial.client")
 			if !ok {
 				t.Fatalf("serial.client module not registered")
 			}

internal/module/test/sip-call-server_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestSIPCallServerFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["sip.call.server"]
+	registration, ok := module.GetModuleRegistration("sip.call.server")
 	if !ok {
 		t.Fatalf("sip.call.server module not registered")
 	}
@@ -70,7 +70,7 @@ func TestBadSIPCallServer(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["sip.call.server"]
+			registration, ok := module.GetModuleRegistration("sip.call.server")
 			if !ok {
 				t.Fatalf("sip.call.server module not registered")
 			}

internal/module/test/sip-dtmf-server_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestSIPDTMFServerFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["sip.dtmf.server"]
+	registration, ok := module.GetModuleRegistration("sip.dtmf.server")
 	if !ok {
 		t.Fatalf("sip.dtmf.server module not registered")
 	}
@@ -89,7 +89,7 @@ func TestBadSIPDTMFServer(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["sip.dtmf.server"]
+			registration, ok := module.GetModuleRegistration("sip.dtmf.server")
 			if !ok {
 				t.Fatalf("sip.dtmf.server module not registered")
 			}

internal/module/test/tcp-client_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestTCPClientFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["net.tcp.client"]
+	registration, ok := module.GetModuleRegistration("net.tcp.client")
 	if !ok {
 		t.Fatalf("net.tcp.client module not registered")
 	}
@@ -77,7 +77,7 @@ func TestBadTCPClient(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["net.tcp.client"]
+			registration, ok := module.GetModuleRegistration("net.tcp.client")
 			if !ok {
 				t.Fatalf("net.tcp.client module not registered")
 			}

internal/module/test/tcp-server_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestTCPServerFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["net.tcp.server"]
+	registration, ok := module.GetModuleRegistration("net.tcp.server")
 	if !ok {
 		t.Fatalf("net.tcp.server module not registered")
 	}
@@ -54,7 +54,7 @@ func TestGoodTCPServer(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["net.tcp.server"]
+			registration, ok := module.GetModuleRegistration("net.tcp.server")
 			if !ok {
 				t.Fatalf("net.tcp.server module not registered")
 			}
@@ -149,7 +149,7 @@ func TestBadTCPServer(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["net.tcp.server"]
+			registration, ok := module.GetModuleRegistration("net.tcp.server")
 			if !ok {
 				t.Fatalf("net.tcp.server module not registered")
 			}

internal/module/test/time-interval_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestTimeIntervalFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["time.interval"]
+	registration, ok := module.GetModuleRegistration("time.interval")
 	if !ok {
 		t.Fatalf("time.interval module not registered")
 	}
@@ -52,7 +52,7 @@ func TestGoodTimeInterval(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["time.interval"]
+			registration, ok := module.GetModuleRegistration("time.interval")
 			if !ok {
 				t.Fatalf("time.interval module not registered")
 			}
@@ -103,7 +103,7 @@ func TestBadTimeInterval(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["time.interval"]
+			registration, ok := module.GetModuleRegistration("time.interval")
 			if !ok {
 				t.Fatalf("time.interval module not registered")
 			}

internal/module/test/time-timer_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestTimeTimerFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["time.timer"]
+	registration, ok := module.GetModuleRegistration("time.timer")
 	if !ok {
 		t.Fatalf("time.timer module not registered")
 	}
@@ -52,7 +52,7 @@ func TestGoodTimeTimer(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["time.timer"]
+			registration, ok := module.GetModuleRegistration("time.timer")
 			if !ok {
 				t.Fatalf("time.timer module not registered")
 			}
@@ -103,7 +103,7 @@ func TestBadTimeTimer(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["time.timer"]
+			registration, ok := module.GetModuleRegistration("time.timer")
 			if !ok {
 				t.Fatalf("time.timer module not registered")
 			}

internal/module/test/udp-client_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestUDPClientFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["net.udp.client"]
+	registration, ok := module.GetModuleRegistration("net.udp.client")
 	if !ok {
 		t.Fatalf("udp.client module not registered")
 	}
@@ -55,7 +55,7 @@ func TestGoodUDPClient(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["net.udp.client"]
+			registration, ok := module.GetModuleRegistration("net.udp.client")
 			if !ok {
 				t.Fatalf("net.udp.client module not registered")
 			}
@@ -125,7 +125,7 @@ func TestBadUDPClient(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["net.udp.client"]
+			registration, ok := module.GetModuleRegistration("net.udp.client")
 			if !ok {
 				t.Fatalf("net.udp.client module not registered")
 			}

internal/module/test/udp-multicast_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestUDPMulticastFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["net.udp.multicast"]
+	registration, ok := module.GetModuleRegistration("net.udp.multicast")
 	if !ok {
 		t.Fatalf("udp.multicast module not registered")
 	}
@@ -84,7 +84,7 @@ func TestBadUDPMulticast(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["net.udp.multicast"]
+			registration, ok := module.GetModuleRegistration("net.udp.multicast")
 			if !ok {
 				t.Fatalf("net.udp.multicast module not registered")
 			}

internal/module/test/udp-server_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestUDPServerFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["net.udp.server"]
+	registration, ok := module.GetModuleRegistration("net.udp.server")
 	if !ok {
 		t.Fatalf("net.udp.server module not registered")
 	}
@@ -52,7 +52,7 @@ func TestGoodUDPServer(t *testing.T) {
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["net.udp.server"]
+			registration, ok := module.GetModuleRegistration("net.udp.server")
 			if !ok {
 				t.Fatalf("net.udp.server module not registered")
 			}
@@ -128,7 +128,7 @@ func TestBadUDPServer(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["net.udp.server"]
+			registration, ok := module.GetModuleRegistration("net.udp.server")
 			if !ok {
 				t.Fatalf("net.udp.server module not registered")
 			}

internal/module/test/websocket-client_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestWebSocketClientFromRegistry(t *testing.T) {
-	registration, ok := module.ModuleRegistry["websocket.client"]
+	registration, ok := module.GetModuleRegistration("websocket.client")
 	if !ok {
 		t.Fatalf("websocket.client module not registered")
 	}
@@ -64,7 +64,7 @@ func TestBadWebSocketClient(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := module.ModuleRegistry["websocket.client"]
+			registration, ok := module.GetModuleRegistration("websocket.client")
 			if !ok {
 				t.Fatalf("websocket.client module not registered")
 			}

internal/module/time-interval.go

@@ -20,8 +20,8 @@ func init() {
 			Properties: map[string]*jsonschema.Schema{
 				"duration": {
 					Title:       "Duration",
+					Description: "time in milliseconds between emitted events",
 					Type:        "integer",
-					Description: "Interval duration in milliseconds",
 				},
 			},
 			Required:             []string{"duration"},
@@ -67,25 +67,26 @@ func (i *TimeInterval) Start(ctx context.Context, inputHandler common.InputHandl
 	ticker := time.NewTicker(time.Millisecond * time.Duration(i.Duration))
 	i.ticker = ticker
 
-	for {
+	for i.ctx.Err() == nil {
 		select {
-		case <-i.ctx.Done():
-			return nil
 		case <-ticker.C:
 			if i.inputHandler != nil {
 				i.inputHandler(i.ctx, i.Id(), time.Now())
 			}
+		default:
+			continue
 		}
 	}
+	<-i.ctx.Done()
+	i.logger.Debug("done")
+	return nil
 }
 
 func (i *TimeInterval) Stop() {
 	if i.cancel != nil {
-		i.cancel()
+		defer i.cancel()
 	}
 	if i.ticker != nil {
 		i.ticker.Stop()
-		i.ticker = nil
 	}
-	i.logger.Debug("done")
 }

internal/module/time-timer.go

@@ -20,8 +20,8 @@ func init() {
 			Properties: map[string]*jsonschema.Schema{
 				"duration": {
 					Title:       "Duration",
+					Description: "time in milliseconds before the timer fires",
 					Type:        "integer",
-					Description: "Interval duration in milliseconds",
 				},
 			},
 			Required:             []string{"duration"},
@@ -66,7 +66,7 @@ func (t *TimeTimer) Start(ctx context.Context, inputHandler common.InputHandler)
 	t.cancel = cancel
 
 	t.timer = time.NewTimer(time.Millisecond * time.Duration(t.Duration))
-	for {
+	for t.ctx.Err() == nil {
 		select {
 		case <-t.ctx.Done():
 			return nil
@@ -76,15 +76,16 @@ func (t *TimeTimer) Start(ctx context.Context, inputHandler common.InputHandler)
 			}
 		}
 	}
+	<-t.ctx.Done()
+	t.logger.Debug("done")
+	return nil
 }
 
 func (t *TimeTimer) Stop() {
 	if t.cancel != nil {
-		t.cancel()
+		defer t.cancel()
 	}
 	if t.timer != nil {
 		t.timer.Stop()
-		t.timer = nil
 	}
-	t.logger.Debug("done")
 }

internal/module/udp-client.go

@@ -21,14 +21,16 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"host": {
-					Title: "Host",
-					Type:  "string",
+					Title:       "Host",
+					Description: "the hostname or IP address of the UDP server to connect to",
+					Type:        "string",
 				},
 				"port": {
-					Title:   "Port",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1),
-					Maximum: jsonschema.Ptr[float64](65535),
+					Title:       "Port",
+					Description: "the port of the UDP server to connect to",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1),
+					Maximum:     jsonschema.Ptr[float64](65535),
 				},
 			},
 			Required:             []string{"host", "port"},
@@ -96,6 +98,7 @@ func (uc *UDPClient) Start(ctx context.Context, inputHandler common.InputHandler
 	}
 
 	<-uc.ctx.Done()
+	uc.logger.Debug("done")
 	return nil
 }
 
@@ -120,14 +123,11 @@ func (uc *UDPClient) Output(ctx context.Context, payload any) error {
 
 func (uc *UDPClient) Stop() {
 	if uc.cancel != nil {
-		uc.cancel()
+		defer uc.cancel()
 	}
 	uc.connMu.Lock()
 	defer uc.connMu.Unlock()
 	if uc.conn != nil {
 		uc.conn.Close()
-		uc.conn = nil
 	}
-
-	uc.logger.Debug("done")
 }

internal/module/udp-multicast.go

@@ -22,14 +22,16 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"ip": {
-					Title: "IP",
-					Type:  "string",
+					Title:       "IP",
+					Description: "the multicast address to listen on",
+					Type:        "string",
 				},
 				"port": {
-					Title:   "Port",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1024),
-					Maximum: jsonschema.Ptr[float64](65535),
+					Title:       "Port",
+					Description: "the port to listen on",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1024),
+					Maximum:     jsonschema.Ptr[float64](65535),
 				},
 			},
 			Required:             []string{"ip", "port"},
@@ -93,35 +95,34 @@ func (um *UDPMulticast) Start(ctx context.Context, inputHandler common.InputHand
 	um.connMu.Unlock()
 
 	buffer := make([]byte, 2048)
-	for {
-		select {
-		case <-um.ctx.Done():
-			return nil
-		default:
-			um.connMu.Lock()
-			um.conn.SetDeadline(time.Now().Add(time.Millisecond * 200))
+	for um.ctx.Err() == nil {
+		um.conn.SetDeadline(time.Now().Add(time.Millisecond * 200))
 
-			numBytes, _, err := um.conn.ReadFromUDP(buffer)
-			um.connMu.Unlock()
-			if err != nil {
-				//NOTE(jwetzell) we hit deadline
-				if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() {
-					continue
-				}
-				return err
+		numBytes, _, err := um.conn.ReadFromUDP(buffer)
+		if err != nil {
+			if errors.Is(err, net.ErrClosed) {
+				break
 			}
+			//NOTE(jwetzell) we hit deadline
+			if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() {
+				continue
+			}
+			return err
+		}
 
-			if numBytes > 0 {
-				message := buffer[:numBytes]
+		if numBytes > 0 {
+			message := buffer[:numBytes]
 
-				if um.inputHandler != nil {
-					um.inputHandler(um.ctx, um.Id(), message)
-				} else {
-					um.logger.Error("input received but no input handler is configured")
-				}
+			if um.inputHandler != nil {
+				um.inputHandler(um.ctx, um.Id(), message)
+			} else {
+				um.logger.Error("input received but no input handler is configured")
 			}
 		}
 	}
+	<-um.ctx.Done()
+	um.logger.Debug("done")
+	return nil
 }
 
 func (um *UDPMulticast) Output(ctx context.Context, payload any) error {
@@ -141,13 +142,11 @@ func (um *UDPMulticast) Output(ctx context.Context, payload any) error {
 
 func (um *UDPMulticast) Stop() {
 	if um.cancel != nil {
-		um.cancel()
+		defer um.cancel()
 	}
 	um.connMu.Lock()
 	defer um.connMu.Unlock()
 	if um.conn != nil {
 		um.conn.Close()
-		um.conn = nil
 	}
-	um.logger.Debug("done")
 }

internal/module/udp-server.go

@@ -23,22 +23,25 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"ip": {
-					Title:   "IP",
-					Type:    "string",
-					Default: json.RawMessage(`"0.0.0.0"`),
+					Title:       "IP",
+					Description: "the IP address to bind the UDP server to",
+					Type:        "string",
+					Default:     json.RawMessage(`"0.0.0.0"`),
 				},
 				"port": {
-					Title:   "Port",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1024),
-					Maximum: jsonschema.Ptr[float64](65535),
+					Title:       "Port",
+					Description: "the port for the UDP server to listen on",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1024),
+					Maximum:     jsonschema.Ptr[float64](65535),
 				},
 				"bufferSize": {
-					Title:   "Buffer Size",
-					Type:    "integer",
-					Minimum: jsonschema.Ptr[float64](1),
-					Maximum: jsonschema.Ptr[float64](65535),
-					Default: json.RawMessage("2048"),
+					Title:       "Buffer Size",
+					Description: "the size of the buffer for incoming UDP messages",
+					Type:        "integer",
+					Minimum:     jsonschema.Ptr[float64](1),
+					Maximum:     jsonschema.Ptr[float64](65535),
+					Default:     json.RawMessage("2048"),
 				},
 			},
 			Required:             []string{"port"},
@@ -111,32 +114,29 @@ func (us *UDPServer) Start(ctx context.Context, inputHandler common.InputHandler
 	}
 	us.listenerMu.Lock()
 	us.listener = listener
+	us.listenerMu.Unlock()
 
 	buffer := make([]byte, us.BufferSize)
 	for us.ctx.Err() == nil {
-		select {
-		case <-us.ctx.Done():
-			return nil
-		default:
-			listener.SetDeadline(time.Now().Add(time.Millisecond * 200))
+		listener.SetDeadline(time.Now().Add(time.Millisecond * 200))
 
-			numBytes, _, err := listener.ReadFromUDP(buffer)
-			if err != nil {
-				//NOTE(jwetzell) we hit deadline
-				if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() {
-					continue
-				}
-				return err
-			}
-			message := buffer[:numBytes]
-			if us.inputHandler != nil {
-				us.inputHandler(us.ctx, us.Id(), message)
-			} else {
-				us.logger.Error("input received but no input handler is configured")
+		numBytes, _, err := listener.ReadFromUDP(buffer)
+		if err != nil {
+			//NOTE(jwetzell) we hit deadline
+			if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() {
+				continue
 			}
+			break
+		}
+		message := buffer[:numBytes]
+		if us.inputHandler != nil {
+			us.inputHandler(us.ctx, us.Id(), message)
+		} else {
+			us.logger.Error("input received but no input handler is configured")
 		}
 	}
-	us.listenerMu.Unlock()
+	<-us.ctx.Done()
+	us.logger.Debug("done")
 	return nil
 }
 
@@ -146,13 +146,11 @@ func (us *UDPServer) Output(ctx context.Context, payload any) error {
 
 func (us *UDPServer) Stop() {
 	if us.cancel != nil {
-		us.cancel()
+		defer us.cancel()
 	}
 	us.listenerMu.Lock()
 	defer us.listenerMu.Unlock()
 	if us.listener != nil {
 		us.listener.Close()
-		us.listener = nil
 	}
-	us.logger.Debug("done")
 }

internal/module/websocket-client.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
-	"net"
 	"net/url"
 	"sync"
 	"time"
@@ -24,8 +23,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"url": {
-					Title: "URL",
-					Type:  "string",
+					Title:       "URL",
+					Description: "the URL of the WebSocket server to connect",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"url"},
@@ -74,7 +74,30 @@ func (wc *WebSocketClient) Type() string {
 func (wc *WebSocketClient) SetupConn() error {
 	wc.connMu.Lock()
 	defer wc.connMu.Unlock()
+	if wc.conn != nil {
+		wc.conn.Close()
+	}
 	conn, _, err := websocket.DefaultDialer.Dial(wc.URL.String(), nil)
+
+	if err != nil {
+		return fmt.Errorf("websocket.client dial error: %w", err)
+	}
+
+	conn.SetCloseHandler(func(code int, text string) error {
+		// NOTE(jwetzell): attempt to send close message back to server before closing connection
+		err := wc.conn.WriteControl(
+			websocket.CloseMessage,
+			websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""),
+			time.Now().Add(time.Minute),
+		)
+		wc.connMu.Lock()
+		defer wc.connMu.Unlock()
+		if wc.conn != nil {
+			wc.conn.Close()
+		}
+		return err
+	})
+
 	wc.conn = conn
 	return err
 }
@@ -92,13 +115,13 @@ func (wc *WebSocketClient) Start(ctx context.Context, inputHandler common.InputH
 			wc.logger.Error("connection error", "error", err)
 		} else {
 			// NOTE(jwetzell): enter read loop until an error occurs
-			wc.logger.Debug("websocket connection established entering read loop")
 			wc.readLoop()
 		}
 		// NOTE(jwetzell): if connection is lost or read error wait before trying again
 		time.Sleep(2 * time.Second)
 	}
 	<-wc.ctx.Done()
+	wc.logger.Debug("done")
 	return nil
 }
 
@@ -108,24 +131,22 @@ func (wc *WebSocketClient) readLoop() {
 			wc.logger.Error("websocket connection is not established")
 			return
 		}
+		// TODO(jwetzell): other ways to timeout?
 		wc.conn.SetReadDeadline(time.Now().Add(5 * time.Second))
 		messageType, message, err := wc.conn.ReadMessage()
 		if err != nil {
-			if opErr, ok := err.(*net.OpError); ok {
-				// NOTE(jwetzell) we hit deadline
-				if opErr.Timeout() {
-					continue
-				}
-				// NOTE(jwetzell) connection was closed
-				if errors.Is(opErr, net.ErrClosed) {
-					continue
-				}
-			}
-			wc.logger.Error("websocket read error", "error", err)
 			return
 		}
 		if wc.inputHandler != nil {
 			switch messageType {
+			case websocket.CloseMessage:
+				return
+			case websocket.PingMessage:
+				err := wc.conn.WriteMessage(websocket.PongMessage, nil)
+				if err != nil {
+					wc.logger.Error("websocket pong error", "error", err)
+					return
+				}
 			case websocket.TextMessage:
 				wc.inputHandler(wc.ctx, wc.Id(), string(message))
 			case websocket.BinaryMessage:
@@ -182,13 +203,15 @@ func (wc *WebSocketClient) Output(ctx context.Context, payload any) error {
 
 func (wc *WebSocketClient) Stop() {
 	if wc.cancel != nil {
-		wc.cancel()
+		defer wc.cancel()
 	}
 	wc.connMu.Lock()
 	defer wc.connMu.Unlock()
 	if wc.conn != nil {
+		err := wc.conn.WriteControl(websocket.CloseMessage, nil, time.Now().Add(time.Minute))
+		if err != nil {
+			wc.logger.Error("websocket close error", "error", err)
+		}
 		wc.conn.Close()
-		wc.conn = nil
 	}
-	wc.logger.Debug("done")
 }

internal/processor/db-query.go

@@ -22,13 +22,13 @@ func init() {
 			Properties: map[string]*jsonschema.Schema{
 				"module": {
 					Title:       "Module ID",
-					Type:        "string",
 					Description: "ID of the database module to query",
+					Type:        "string",
 				},
 				"query": {
 					Title:       "Query",
-					Type:        "string",
 					Description: "SQL query to execute",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"module", "query"},

internal/processor/filter-expr.go

@@ -19,8 +19,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"expression": {
-					Title: "Expression",
-					Type:  "string",
+					Title:       "Expression",
+					Description: "Expr expression to evaluate, must return a boolean",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"expression"},

internal/processor/filter-rate.go

@@ -0,0 +1,58 @@
+package processor
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/jsonschema-go/jsonschema"
+	"github.com/jwetzell/showbridge-go/internal/common"
+	"github.com/jwetzell/showbridge-go/internal/config"
+	"golang.org/x/time/rate"
+)
+
+func init() {
+	RegisterProcessor(ProcessorRegistration{
+		Type:  "filter.rate",
+		Title: "Filter by Rate",
+		ParamsSchema: &jsonschema.Schema{
+			Type: "object",
+			Properties: map[string]*jsonschema.Schema{
+				"rate": {
+					Title:       "Rate",
+					Description: "number of events to allow per second",
+					Type:        "integer",
+				},
+			},
+			Required: []string{"rate"},
+		},
+		New: func(config config.ProcessorConfig) (Processor, error) {
+			params := config.Params
+			rateInt, err := params.GetInt("rate")
+			if err != nil {
+				return nil, fmt.Errorf("filter.rate rate error: %w", err)
+			}
+
+			limiter := rate.NewLimiter(rate.Limit(rateInt), rateInt*2)
+
+			return &FilterRate{config: config, limiter: limiter}, nil
+		},
+	})
+}
+
+type FilterRate struct {
+	config  config.ProcessorConfig
+	limiter *rate.Limiter
+}
+
+func (fc *FilterRate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
+	err := fc.limiter.Wait(ctx)
+	if err != nil {
+		wrappedPayload.End = true
+		return wrappedPayload, err
+	}
+	return wrappedPayload, nil
+}
+
+func (fc *FilterRate) Type() string {
+	return fc.config.Type
+}

internal/processor/filter-regex.go

@@ -19,8 +19,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"pattern": {
-					Title: "Pattern",
-					Type:  "string",
+					Title:       "Pattern",
+					Description: "regex pattern to match against",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"pattern"},

internal/processor/float-parse.go

@@ -20,10 +20,11 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"bitSize": {
-					Title:   "Bit Size",
-					Type:    "integer",
-					Enum:    []any{32, 64},
-					Default: json.RawMessage("64"),
+					Title:       "Bit Size",
+					Description: "bit size of the resulting float",
+					Type:        "integer",
+					Enum:        []any{32, 64},
+					Default:     json.RawMessage("64"),
 				},
 			},
 			AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},

internal/processor/float-random.go

@@ -20,18 +20,21 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"bitSize": {
-					Title:   "Bit Size",
-					Type:    "integer",
-					Enum:    []any{32, 64},
-					Default: json.RawMessage("32"),
+					Title:       "Bit Size",
+					Description: "bit size of the resulting float",
+					Type:        "integer",
+					Enum:        []any{32, 64},
+					Default:     json.RawMessage("32"),
 				},
 				"min": {
-					Title: "Minimum",
-					Type:  "number",
+					Title:       "Minimum",
+					Description: "inclusive minimum value of the random float",
+					Type:        "number",
 				},
 				"max": {
-					Title: "Maximum",
-					Type:  "number",
+					Title:       "Maximum",
+					Description: "exclusive maximum value of the random float",
+					Type:        "number",
 				},
 			},
 			Required:             []string{"min", "max"},

internal/processor/free-d-create.go

@@ -21,40 +21,49 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"id": {
-					Title: "Camera ID",
-					Type:  "string",
+					Title:       "Camera ID",
+					Description: "ID of the camera this FreeD position corresponds to",
+					Type:        "string",
 				},
 				"pan": {
-					Title: "Pan",
-					Type:  "string",
+					Title:       "Pan",
+					Description: "pan angle of the camera",
+					Type:        "string",
 				},
 				"tilt": {
-					Title: "Tilt",
-					Type:  "string",
+					Title:       "Tilt",
+					Description: "tilt angle of the camera",
+					Type:        "string",
 				},
 				"roll": {
-					Title: "Roll",
-					Type:  "string",
+					Title:       "Roll",
+					Description: "roll angle of the camera",
+					Type:        "string",
 				},
 				"posX": {
-					Title: "Position X",
-					Type:  "string",
+					Title:       "Position X",
+					Description: "X coordinate of the camera's position",
+					Type:        "string",
 				},
 				"posY": {
-					Title: "Position Y",
-					Type:  "string",
+					Title:       "Position Y",
+					Description: "Y coordinate of the camera's position",
+					Type:        "string",
 				},
 				"posZ": {
-					Title: "Position Z",
-					Type:  "string",
+					Title:       "Position Z",
+					Description: "Z coordinate of the camera's position",
+					Type:        "string",
 				},
 				"zoom": {
-					Title: "Zoom",
-					Type:  "string",
+					Title:       "Zoom",
+					Description: "zoom level of the camera",
+					Type:        "string",
 				},
 				"focus": {
-					Title: "Focus",
-					Type:  "string",
+					Title:       "Focus",
+					Description: "focus level of the camera",
+					Type:        "string",
 				},
 			},
 			Required: []string{

internal/processor/http-request-do.go

@@ -22,13 +22,15 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"method": {
-					Title: "HTTP Method",
-					Type:  "string",
-					Enum:  []any{"GET", "POST", "PUT", "PATCH", "DELETE"},
+					Title:       "HTTP Method",
+					Description: "method to use for the HTTP request",
+					Type:        "string",
+					Enum:        []any{"GET", "POST", "PUT", "PATCH", "DELETE"},
 				},
 				"url": {
-					Title: "URL",
-					Type:  "string",
+					Title:       "URL",
+					Description: "URL to send the HTTP request to",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"method", "url"},

internal/processor/http-response-create.go

@@ -19,15 +19,17 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"status": {
-					Title: "Status Code",
-					Type:  "integer",
+					Title:       "Status Code",
+					Description: "status code to set on the response",
+					Type:        "integer",
 				},
-				"body": {
-					Title: "Body",
-					Type:  "string",
+				"bodyTemplate": {
+					Title:       "Body Template",
+					Description: "template for the response body",
+					Type:        "string",
 				},
 			},
-			Required:             []string{"status", "body"},
+			Required:             []string{"status", "bodyTemplate"},
 			AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
 		},
 		New: func(config config.ProcessorConfig) (Processor, error) {

internal/processor/int-parse.go

@@ -20,16 +20,18 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"base": {
-					Title:   "Base",
-					Type:    "integer",
-					Enum:    []any{0, 2, 8, 10, 16},
-					Default: json.RawMessage("10"),
+					Title:       "Base",
+					Description: "numerical base to use for parsing the integer",
+					Type:        "integer",
+					Enum:        []any{0, 2, 8, 10, 16},
+					Default:     json.RawMessage("10"),
 				},
 				"bitSize": {
-					Title:   "Bit Size",
-					Type:    "integer",
-					Enum:    []any{0, 8, 16, 32, 64},
-					Default: json.RawMessage("64"),
+					Title:       "Bit Size",
+					Description: "bit size of the resulting integer",
+					Type:        "integer",
+					Enum:        []any{0, 8, 16, 32, 64},
+					Default:     json.RawMessage("64"),
 				},
 			},
 			AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},

internal/processor/int-random.go

@@ -19,12 +19,14 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"min": {
-					Title: "Minimum",
-					Type:  "integer",
+					Title:       "Minimum",
+					Description: "inclusive minimum value of the random integer",
+					Type:        "integer",
 				},
 				"max": {
-					Title: "Maximum",
-					Type:  "integer",
+					Title:       "Maximum",
+					Description: "inclusive maximum value of the random integer",
+					Type:        "integer",
 				},
 			},
 			Required:             []string{"min", "max"},

internal/processor/int-scale.go

@@ -18,20 +18,24 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"inMin": {
-					Title: "Input Minimum",
-					Type:  "integer",
+					Title:       "Input Minimum",
+					Description: "minimum value of the input integer",
+					Type:        "integer",
 				},
 				"inMax": {
-					Title: "Input Maximum",
-					Type:  "integer",
+					Title:       "Input Maximum",
+					Description: "maximum value of the input integer",
+					Type:        "integer",
 				},
 				"outMin": {
-					Title: "Output Minimum",
-					Type:  "integer",
+					Title:       "Output Minimum",
+					Description: "minimum value of the output integer",
+					Type:        "integer",
 				},
 				"outMax": {
-					Title: "Output Maximum",
-					Type:  "integer",
+					Title:       "Output Maximum",
+					Description: "maximum value of the output integer",
+					Type:        "integer",
 				},
 			},
 			Required:             []string{"inMin", "inMax", "outMin", "outMax"},

internal/processor/kv-get.go

@@ -19,12 +19,14 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"module": {
-					Title: "Module ID",
-					Type:  "string",
+					Title:       "Module ID",
+					Description: "ID of the key-value module to get the value from",
+					Type:        "string",
 				},
 				"key": {
-					Title: "Key",
-					Type:  "string",
+					Title:       "Key",
+					Description: "key to retrieve from the key-value module",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"module", "key"},
@@ -77,7 +79,7 @@ func (kvg *KVGet) Process(ctx context.Context, wrappedPayload common.WrappedPayl
 		kvg.module = kvModule
 	}
 
-	value, err := kvg.module.Get(kvg.Key)
+	value, err := kvg.module.Get(ctx, kvg.Key)
 	if err != nil {
 		wrappedPayload.End = true
 		return wrappedPayload, fmt.Errorf("kv.get error getting key: %w", err)

internal/processor/kv-set.go

@@ -19,12 +19,14 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"module": {
-					Title: "Module ID",
-					Type:  "string",
+					Title:       "Module ID",
+					Description: "ID of the key-value module to set the value in",
+					Type:        "string",
 				},
 				"key": {
-					Title: "Key",
-					Type:  "string",
+					Title:       "Key",
+					Description: "key to set in the key-value module",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"module", "key"},
@@ -78,7 +80,7 @@ func (kvs *KVSet) Process(ctx context.Context, wrappedPayload common.WrappedPayl
 		kvs.module = kvModule
 	}
 
-	err := kvs.module.Set(kvs.Key, wrappedPayload.Payload)
+	err := kvs.module.Set(ctx, kvs.Key, wrappedPayload.Payload)
 	if err != nil {
 		wrappedPayload.End = true
 		return wrappedPayload, fmt.Errorf("kv.set error setting key: %w", err)

internal/processor/midi-control_change-create.go

@@ -23,16 +23,19 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"channel": {
-					Title: "Channel",
-					Type:  "string",
+					Title:       "Channel",
+					Description: "channel number for the control change message",
+					Type:        "string",
 				},
 				"control": {
-					Title: "Control",
-					Type:  "string",
+					Title:       "Control",
+					Description: "control number for the control change message",
+					Type:        "string",
 				},
 				"value": {
-					Title: "Value",
-					Type:  "string",
+					Title:       "Value",
+					Description: "value for the control change message",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"channel", "control", "value"},

internal/processor/midi-note_off-create.go

@@ -23,16 +23,19 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"channel": {
-					Title: "Channel",
-					Type:  "string",
+					Title:       "Channel",
+					Description: "channel number for the note off message",
+					Type:        "string",
 				},
 				"note": {
-					Title: "Note",
-					Type:  "string",
+					Title:       "Note",
+					Description: "note number for the note off message",
+					Type:        "string",
 				},
 				"velocity": {
-					Title: "Velocity",
-					Type:  "string",
+					Title:       "Velocity",
+					Description: "velocity for the note off message",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"channel", "note", "velocity"},

internal/processor/midi-note_on-create.go

@@ -23,16 +23,19 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"channel": {
-					Title: "Channel",
-					Type:  "string",
+					Title:       "Channel",
+					Description: "channel number for the note on message",
+					Type:        "string",
 				},
 				"note": {
-					Title: "Note",
-					Type:  "string",
+					Title:       "Note",
+					Description: "note number for the note on message",
+					Type:        "string",
 				},
 				"velocity": {
-					Title: "Velocity",
-					Type:  "string",
+					Title:       "Velocity",
+					Description: "velocity for the note on message",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"channel", "note", "velocity"},

internal/processor/midi-program_change-create.go

@@ -23,12 +23,14 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"channel": {
-					Title: "Channel",
-					Type:  "string",
+					Title:       "Channel",
+					Description: "channel number for the program change message",
+					Type:        "string",
 				},
 				"program": {
-					Title: "Program",
-					Type:  "string",
+					Title:       "Program",
+					Description: "program number for the program change message",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"type", "channel", "program"},

internal/processor/module-output.go

@@ -20,8 +20,8 @@ func init() {
 			Properties: map[string]*jsonschema.Schema{
 				"module": {
 					Title:       "Module ID",
-					Type:        "string",
 					Description: "ID of module to send output to",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"module"},

internal/processor/osc-message-create.go

@@ -23,19 +23,22 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"address": {
-					Title: "Address",
-					Type:  "string",
+					Title:       "Address",
+					Description: "OSC address for the message",
+					Type:        "string",
 				},
 				"args": {
-					Title: "Arguments",
-					Type:  "array",
+					Title:       "Arguments",
+					Description: "arguments for the OSC message",
+					Type:        "array",
 					Items: &jsonschema.Schema{
 						Type: "string",
 					},
 				},
 				"types": {
-					Title: "Argument Types",
-					Type:  "string",
+					Title:       "Argument Types",
+					Description: "string of OSC types corresponding to the arguments in args",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"address"},

internal/processor/processor.go

@@ -35,13 +35,33 @@ func RegisterProcessor(processor ProcessorRegistration) {
 	processorRegistryMu.Lock()
 	defer processorRegistryMu.Unlock()
 
-	if _, ok := ProcessorRegistry[string(processor.Type)]; ok {
+	if _, ok := processorRegistry[string(processor.Type)]; ok {
 		panic(fmt.Sprintf("processor already registered: %s", processor.Type))
 	}
-	ProcessorRegistry[string(processor.Type)] = processor
+	processorRegistry[string(processor.Type)] = processor
+}
+
+type ProcessorRegistry map[string]ProcessorRegistration
+
+func GetProcessorRegistration(processorType string) (ProcessorRegistration, bool) {
+	processorRegistryMu.RLock()
+	defer processorRegistryMu.RUnlock()
+	processor, ok := processorRegistry[processorType]
+	return processor, ok
+}
+
+func GetProcessorRegistrations() []ProcessorRegistration {
+	processorRegistryMu.RLock()
+	defer processorRegistryMu.RUnlock()
+
+	registrations := make([]ProcessorRegistration, 0, len(processorRegistry))
+	for _, processor := range processorRegistry {
+		registrations = append(registrations, processor)
+	}
+	return registrations
 }
 
 var (
 	processorRegistryMu sync.RWMutex
-	ProcessorRegistry   = make(map[string]ProcessorRegistration)
+	processorRegistry   = make(map[string]ProcessorRegistration)
 )

internal/processor/pubsub-publish.go

@@ -22,13 +22,13 @@ func init() {
 			Properties: map[string]*jsonschema.Schema{
 				"module": {
 					Title:       "Module ID",
-					Type:        "string",
 					Description: "ID of the module to publish to",
+					Type:        "string",
 				},
 				"topic": {
 					Title:       "Topic",
+					Description: "topic to publish to",
 					Type:        "string",
-					Description: "Topic to publish to",
 				},
 			},
 			Required:             []string{"module", "topic"},

internal/processor/router-input.go

@@ -20,8 +20,8 @@ func init() {
 			Properties: map[string]*jsonschema.Schema{
 				"source": {
 					Title:       "Source",
-					Type:        "string",
 					Description: "source to report as to the router",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"source"},

internal/processor/script-expr.go

@@ -25,8 +25,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"expression": {
-					Title: "Expression",
-					Type:  "string",
+					Title:       "Expression",
+					Description: "Expr expression to evaluate",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"expression"},

internal/processor/script-js.go

@@ -18,8 +18,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"program": {
-					Title: "Program",
-					Type:  "string",
+					Title:       "Program",
+					Description: "JavaScript program to run",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"program"},

internal/processor/script-wasm.go

@@ -20,18 +20,21 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"path": {
-					Title: "Path",
-					Type:  "string",
+					Title:       "Path",
+					Description: "path to the WASM plugin to load",
+					Type:        "string",
 				},
 				"function": {
-					Title:   "Function",
-					Type:    "string",
-					Default: json.RawMessage(`"process"`),
+					Title:       "Function",
+					Description: "exported function to call on the WASM plugin",
+					Type:        "string",
+					Default:     json.RawMessage(`"process"`),
 				},
 				"enableWasi": {
-					Title:   "Enable WASI",
-					Type:    "boolean",
-					Default: json.RawMessage("false"),
+					Title:       "Enable WASI",
+					Description: "whether to enable WASI when running the WASM plugin",
+					Type:        "boolean",
+					Default:     json.RawMessage("false"),
 				},
 			},
 			Required:             []string{"path"},

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

@@ -19,15 +19,19 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"preWait": {
-					Title: "Pre Wait (ms)",
-					Type:  "integer",
-				},
-				"postWait": {
-					Title: "Post Wait (ms)",
-					Type:  "integer",
+					Title:       "Pre Wait (ms)",
+					Description: "number of milliseconds to wait before playing the audio",
+					Type:        "integer",
 				},
 				"audioFile": {
-					Type: "string",
+					Title:       "Audio File",
+					Description: "path to the audio file to play",
+					Type:        "string",
+				},
+				"postWait": {
+					Title:       "Post Wait (ms)",
+					Description: "number of milliseconds to wait after playing the audio",
+					Type:        "integer",
 				},
 			},
 			Required:             []string{"preWait", "postWait", "audioFile"},

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

@@ -21,15 +21,19 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"preWait": {
-					Title: "Pre Wait (ms)",
-					Type:  "integer",
-				},
-				"postWait": {
-					Title: "Post Wait (ms)",
-					Type:  "integer",
+					Title:       "Pre Wait (ms)",
+					Description: "number of milliseconds to wait before sending the DTMF tones",
+					Type:        "integer",
 				},
 				"digits": {
-					Type: "string",
+					Title:       "Digits",
+					Description: "DTMF digits to send",
+					Type:        "string",
+				},
+				"postWait": {
+					Title:       "Post Wait (ms)",
+					Description: "number of milliseconds to wait after sending the DTMF tones",
+					Type:        "integer",
 				},
 			},
 			Required:             []string{"preWait", "postWait", "digits"},

internal/processor/string-create.go

@@ -19,8 +19,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"template": {
-					Title: "Template",
-					Type:  "string",
+					Title:       "Template",
+					Description: "template to evaluate",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"template"},

internal/processor/string-split.go

@@ -19,8 +19,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"separator": {
-					Title: "Separator",
-					Type:  "string",
+					Title:       "Separator",
+					Description: "separator to split the string on",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"separator"},

internal/processor/struct-field-get.go

@@ -19,8 +19,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"name": {
-					Title: "Field Name",
-					Type:  "string",
+					Title:       "Field Name",
+					Description: "name of the struct field to extract",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"name"},

internal/processor/struct-method-get.go

@@ -19,8 +19,9 @@ func init() {
 			Type: "object",
 			Properties: map[string]*jsonschema.Schema{
 				"name": {
-					Title: "Method Name",
-					Type:  "string",
+					Title:       "Method Name",
+					Description: "name of the struct method to extract and call (with no arguments)",
+					Type:        "string",
 				},
 			},
 			Required:             []string{"name"},

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

@@ -11,7 +11,7 @@ import (
 )
 
 func TestArtnetPacketDecodeFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["artnet.packet.decode"]
+	registration, ok := processor.GetProcessorRegistration("artnet.packet.decode")
 	if !ok {
 		t.Fatalf("artnet.packet.decode processor not registered")
 	}
@@ -41,15 +41,11 @@ func TestGoodArtnetPacketDecode(t *testing.T) {
 			name:    "number",
 			payload: []byte{65, 114, 116, 45, 78, 101, 116, 0, 0, 80, 0, 14, 237, 0, 1, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
 			expected: &artnet.ArtDmx{
-				ID:        [8]byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
-				OpCode:    artnet.OpDmx,
-				ProtVerHi: 0,
-				ProtVerLo: 14,
-				Sequence:  237,
-				Physical:  0,
-				SubUni:    1,
-				Net:       0,
-				Data:      make([]uint8, 512),
+				Sequence: 237,
+				Physical: 0,
+				SubUni:   1,
+				Net:      0,
+				Data:     make([]uint8, 512),
 			},
 		},
 	}

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

@@ -11,7 +11,7 @@ import (
 )
 
 func TestArtnetPacketEncodeFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["artnet.packet.encode"]
+	registration, ok := processor.GetProcessorRegistration("artnet.packet.encode")
 	if !ok {
 		t.Fatalf("artnet.packet.encode processor not registered")
 	}
@@ -41,15 +41,11 @@ func TestGoodArtnetPacketEncode(t *testing.T) {
 			name:     "number",
 			expected: []byte{65, 114, 116, 45, 78, 101, 116, 0, 0, 80, 0, 14, 237, 0, 1, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
 			payload: &artnet.ArtDmx{
-				ID:        [8]byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
-				OpCode:    artnet.OpDmx,
-				ProtVerHi: 0,
-				ProtVerLo: 14,
-				Sequence:  237,
-				Physical:  0,
-				SubUni:    1,
-				Net:       0,
-				Data:      make([]uint8, 512),
+				Sequence: 237,
+				Physical: 0,
+				SubUni:   1,
+				Net:      0,
+				Data:     make([]uint8, 512),
 			},
 		},
 	}
@@ -104,15 +100,11 @@ func TestBadArtnetPacketEncode(t *testing.T) {
 func BenchmarkArtnetPacketEncode(b *testing.B) {
 	processorInstance := processor.ArtNetPacketEncode{}
 	payload := &artnet.ArtDmx{
-		ID:        [8]byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
-		OpCode:    artnet.OpDmx,
-		ProtVerHi: 0,
-		ProtVerLo: 14,
-		Sequence:  237,
-		Physical:  0,
-		SubUni:    1,
-		Net:       0,
-		Data:      make([]uint8, 512),
+		Sequence: 237,
+		Physical: 0,
+		SubUni:   1,
+		Net:      0,
+		Data:     make([]uint8, 512),
 	}
 
 	for b.Loop() {

internal/processor/test/db-query_test.go

@@ -12,7 +12,7 @@ import (
 )
 
 func TestDbQueryFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["db.query"]
+	registration, ok := processor.GetProcessorRegistration("db.query")
 	if !ok {
 		t.Fatalf("db.query processor not registered")
 	}
@@ -100,7 +100,7 @@ func TestGoodDbQuery(t *testing.T) {
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["db.query"]
+			registration, ok := processor.GetProcessorRegistration("db.query")
 			if !ok {
 				t.Fatalf("db.query processor not registered")
 			}
@@ -243,7 +243,7 @@ func TestBadDbQuery(t *testing.T) {
 			errorString:           "db.query unable to find module with id: test",
 		},
 		{
-			name:    "module not a DatabseModule",
+			name:    "module not a DatabaseModule",
 			payload: test.TestStruct{Data: "hello"},
 			params: map[string]any{
 				"module": "test",
@@ -259,7 +259,7 @@ func TestBadDbQuery(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["db.query"]
+			registration, ok := processor.GetProcessorRegistration("db.query")
 			if !ok {
 				t.Fatalf("db.query processor not registered")
 			}
@@ -293,7 +293,7 @@ func TestBadDbQuery(t *testing.T) {
 }
 
 func BenchmarkDbQuery(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["db.query"]
+	registration, ok := processor.GetProcessorRegistration("db.query")
 	if !ok {
 		b.Fatalf("db.query processor not registered")
 	}

internal/processor/test/debug-log_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestDebugLogFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["debug.log"]
+	registration, ok := processor.GetProcessorRegistration("debug.log")
 	if !ok {
 		t.Fatalf("debug.log processor not registered")
 	}
@@ -52,7 +52,7 @@ func TestGoodDebugLog(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["debug.log"]
+			registration, ok := processor.GetProcessorRegistration("debug.log")
 			if !ok {
 				t.Fatalf("debug.log processor not registered")
 			}
@@ -89,7 +89,7 @@ func TestBadDebugLog(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["debug.log"]
+			registration, ok := processor.GetProcessorRegistration("debug.log")
 			if !ok {
 				t.Fatalf("debug.log processor not registered")
 			}
@@ -120,7 +120,7 @@ func TestBadDebugLog(t *testing.T) {
 }
 
 func BenchmarkDebugLog(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["debug.log"]
+	registration, ok := processor.GetProcessorRegistration("debug.log")
 	if !ok {
 		b.Fatalf("debug.log processor not registered")
 	}

internal/processor/test/filter-change_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestFilterChangeFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["filter.change"]
+	registration, ok := processor.GetProcessorRegistration("filter.change")
 	if !ok {
 		t.Fatalf("filter.change processor not registered")
 	}
@@ -58,7 +58,7 @@ func TestGoodFilterChange(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["filter.change"]
+			registration, ok := processor.GetProcessorRegistration("filter.change")
 			if !ok {
 				t.Fatalf("filter.change processor not registered")
 			}
@@ -95,7 +95,7 @@ func TestBadFilterChange(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["filter.change"]
+			registration, ok := processor.GetProcessorRegistration("filter.change")
 			if !ok {
 				t.Fatalf("filter.change processor not registered")
 			}
@@ -126,7 +126,7 @@ func TestBadFilterChange(t *testing.T) {
 }
 
 func BenchmarkFilterChange(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["filter.change"]
+	registration, ok := processor.GetProcessorRegistration("filter.change")
 	if !ok {
 		b.Fatalf("filter.change processor not registered")
 	}

internal/processor/test/filter-expr_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestFilterExprFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["filter.expr"]
+	registration, ok := processor.GetProcessorRegistration("filter.expr")
 	if !ok {
 		t.Fatalf("filter.expr processor not registered")
 	}
@@ -71,7 +71,7 @@ func TestGoodFilterExpr(t *testing.T) {
 
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["filter.expr"]
+			registration, ok := processor.GetProcessorRegistration("filter.expr")
 			if !ok {
 				t.Fatalf("filter.expr processor not registered")
 			}
@@ -144,7 +144,7 @@ func TestBadFilterExpr(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["filter.expr"]
+			registration, ok := processor.GetProcessorRegistration("filter.expr")
 			if !ok {
 				t.Fatalf("filter.expr processor not registered")
 			}
@@ -173,7 +173,7 @@ func TestBadFilterExpr(t *testing.T) {
 }
 
 func BenchmarkFilterExpr(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["filter.expr"]
+	registration, ok := processor.GetProcessorRegistration("filter.expr")
 	if !ok {
 		b.Fatalf("filter.expr processor not registered")
 	}

internal/processor/test/filter-rate_test.go

@@ -0,0 +1,116 @@
+package processor_test
+
+import (
+	"testing"
+
+	"github.com/jwetzell/showbridge-go/internal/common"
+	"github.com/jwetzell/showbridge-go/internal/config"
+	"github.com/jwetzell/showbridge-go/internal/processor"
+	"github.com/jwetzell/showbridge-go/internal/test"
+)
+
+func TestFilterRateFromRegistry(t *testing.T) {
+	registration, ok := processor.GetProcessorRegistration("filter.rate")
+	if !ok {
+		t.Fatalf("filter.rate processor not registered")
+	}
+
+	processorInstance, err := registration.New(config.ProcessorConfig{
+		Type: "filter.rate",
+		Params: map[string]any{
+			"rate": 1,
+		},
+	})
+	if err != nil {
+		t.Fatalf("failed to create filter.rate processor: %s", err)
+	}
+
+	if processorInstance.Type() != "filter.rate" {
+		t.Fatalf("filter.rate processor has wrong type: %s", processorInstance.Type())
+	}
+}
+
+func TestGoodFilterRate(t *testing.T) {
+	testCases := []struct {
+		name    string
+		params  map[string]any
+		payload any
+		match   bool
+	}{}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			registration, ok := processor.GetProcessorRegistration("filter.rate")
+			if !ok {
+				t.Fatalf("filter.rate processor not registered")
+			}
+
+			processorInstance, err := registration.New(config.ProcessorConfig{
+				Type:   "filter.rate",
+				Params: testCase.params,
+			})
+
+			if err != nil {
+				t.Fatalf("filter.rate failed to create processor: %s", err)
+			}
+
+			_, err = processorInstance.Process(t.Context(), common.WrappedPayload{Payload: testCase.payload})
+			// TODO(jwetzell): figure out how to test the rate limiting behavior
+		})
+	}
+}
+
+func TestBadFilterRate(t *testing.T) {
+	tests := []struct {
+		name        string
+		params      map[string]any
+		payload     any
+		errorString string
+	}{
+		{
+			name:   "no rate parameter",
+			params: map[string]any{
+				// no rate parameter
+			},
+			payload:     test.TestStruct{},
+			errorString: "filter.rate rate error: not found",
+		},
+		{
+			name: "non-int rate parameter",
+			params: map[string]any{
+				"rate": "12345",
+			},
+			payload:     test.TestStruct{},
+			errorString: "filter.rate rate error: not a number",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			registration, ok := processor.GetProcessorRegistration("filter.rate")
+			if !ok {
+				t.Fatalf("filter.rate processor not registered")
+			}
+
+			processorInstance, err := registration.New(config.ProcessorConfig{
+				Type:   "filter.rate",
+				Params: test.params,
+			})
+			if err != nil {
+				if err.Error() != test.errorString {
+					t.Fatalf("filter.rate got error '%s', expected '%s'", err.Error(), test.errorString)
+				}
+				return
+			}
+			got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Payload: test.payload})
+
+			if err == nil {
+				t.Fatalf("filter.rate expected to fail but succeeded, got: %v", got)
+
+			}
+			if err.Error() != test.errorString {
+				t.Fatalf("filter.rate got error '%s', expected '%s'", err.Error(), test.errorString)
+			}
+		})
+	}
+}

internal/processor/test/filter-regex_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestFilterRegexFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["filter.regex"]
+	registration, ok := processor.GetProcessorRegistration("filter.regex")
 	if !ok {
 		t.Fatalf("filter.regex processor not registered")
 	}
@@ -77,7 +77,7 @@ func TestGoodFilterRegex(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["filter.regex"]
+			registration, ok := processor.GetProcessorRegistration("filter.regex")
 			if !ok {
 				t.Fatalf("filter.regex processor not registered")
 			}
@@ -145,7 +145,7 @@ func TestBadFilterRegex(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["filter.regex"]
+			registration, ok := processor.GetProcessorRegistration("filter.regex")
 			if !ok {
 				t.Fatalf("filter.regex processor not registered")
 			}
@@ -176,7 +176,7 @@ func TestBadFilterRegex(t *testing.T) {
 }
 
 func BenchmarkFilterRegex(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["filter.regex"]
+	registration, ok := processor.GetProcessorRegistration("filter.regex")
 	if !ok {
 		b.Fatalf("filter.regex processor not registered")
 	}

internal/processor/test/float-parse_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestFloatParseFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["float.parse"]
+	registration, ok := processor.GetProcessorRegistration("float.parse")
 	if !ok {
 		t.Fatalf("float.parse processor not registered")
 	}
@@ -63,7 +63,7 @@ func TestGoodFloatParse(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["float.parse"]
+			registration, ok := processor.GetProcessorRegistration("float.parse")
 			if !ok {
 				t.Fatalf("float.parse processor not registered")
 			}
@@ -136,7 +136,7 @@ func TestBadFloatParse(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["float.parse"]
+			registration, ok := processor.GetProcessorRegistration("float.parse")
 			if !ok {
 				t.Fatalf("float.parse processor not registered")
 			}
@@ -167,7 +167,7 @@ func TestBadFloatParse(t *testing.T) {
 }
 
 func BenchmarkFloatParse(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["float.parse"]
+	registration, ok := processor.GetProcessorRegistration("float.parse")
 	if !ok {
 		b.Fatalf("float.parse processor not registered")
 	}

internal/processor/test/float-random_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestFloatRandomFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["float.random"]
+	registration, ok := processor.GetProcessorRegistration("float.random")
 	if !ok {
 		t.Fatalf("float.random processor not registered")
 	}
@@ -57,7 +57,7 @@ func TestGoodFloatRandom(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["float.random"]
+			registration, ok := processor.GetProcessorRegistration("float.random")
 			if !ok {
 				t.Fatalf("float.random processor not registered")
 			}
@@ -155,7 +155,7 @@ func TestBadFloatRandom(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["float.random"]
+			registration, ok := processor.GetProcessorRegistration("float.random")
 			if !ok {
 				t.Fatalf("float.random processor not registered")
 			}
@@ -186,7 +186,7 @@ func TestBadFloatRandom(t *testing.T) {
 }
 
 func BenchmarkFloatRandom(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["float.random"]
+	registration, ok := processor.GetProcessorRegistration("float.random")
 	if !ok {
 		b.Fatalf("float.random processor not registered")
 	}

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

@@ -11,7 +11,7 @@ import (
 )
 
 func TestFreeDCreateFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["freed.create"]
+	registration, ok := processor.GetProcessorRegistration("freed.create")
 	if !ok {
 		t.Fatalf("freed.create processor not registered")
 	}
@@ -89,7 +89,7 @@ func TestGoodFreeDCreate(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["freed.create"]
+			registration, ok := processor.GetProcessorRegistration("freed.create")
 			if !ok {
 				t.Fatalf("freed.create processor not registered")
 			}
@@ -848,7 +848,7 @@ func TestBadFreeDCreate(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["freed.create"]
+			registration, ok := processor.GetProcessorRegistration("freed.create")
 			if !ok {
 				t.Fatalf("freed.create processor not registered")
 			}
@@ -879,7 +879,7 @@ func TestBadFreeDCreate(t *testing.T) {
 }
 
 func BenchmarkFreeDCreate(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["freed.create"]
+	registration, ok := processor.GetProcessorRegistration("freed.create")
 	if !ok {
 		b.Fatalf("freed.create processor not registered")
 	}

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

@@ -11,7 +11,7 @@ import (
 )
 
 func TestFreeDDecodeFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["freed.decode"]
+	registration, ok := processor.GetProcessorRegistration("freed.decode")
 	if !ok {
 		t.Fatalf("freed.decode processor not registered")
 	}

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

@@ -11,7 +11,7 @@ import (
 )
 
 func TestFreeDEncodeFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["freed.encode"]
+	registration, ok := processor.GetProcessorRegistration("freed.encode")
 	if !ok {
 		t.Fatalf("freed.encode processor not registered")
 	}

internal/processor/test/http-request-do_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestHTTPRequestCreateFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["http.request.do"]
+	registration, ok := processor.GetProcessorRegistration("http.request.do")
 	if !ok {
 		t.Fatalf("http.request.do processor not registered")
 	}
@@ -44,7 +44,7 @@ func TestGoodHTTPRequestDo(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["http.request.do"]
+			registration, ok := processor.GetProcessorRegistration("http.request.do")
 			if !ok {
 				t.Fatalf("http.request.do processor not registered")
 			}
@@ -132,7 +132,7 @@ func TestBadHTTPRequestDo(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["http.request.do"]
+			registration, ok := processor.GetProcessorRegistration("http.request.do")
 			if !ok {
 				t.Fatalf("http.request.do processor not registered")
 			}

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

@@ -10,7 +10,7 @@ import (
 )
 
 func TestHTTPResponseCreateFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["http.response.create"]
+	registration, ok := processor.GetProcessorRegistration("http.response.create")
 	if !ok {
 		t.Fatalf("http.response.create processor not registered")
 	}
@@ -54,7 +54,7 @@ func TestGoodHTTPResponseCreate(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["http.response.create"]
+			registration, ok := processor.GetProcessorRegistration("http.response.create")
 			if !ok {
 				t.Fatalf("http.response.create processor not registered")
 			}
@@ -128,7 +128,7 @@ func TestBadHTTPResponseCreate(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["http.response.create"]
+			registration, ok := processor.GetProcessorRegistration("http.response.create")
 			if !ok {
 				t.Fatalf("http.response.create processor not registered")
 			}
@@ -159,7 +159,7 @@ func TestBadHTTPResponseCreate(t *testing.T) {
 }
 
 func BenchmarkHTTPResponseCreate(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["http.response.create"]
+	registration, ok := processor.GetProcessorRegistration("http.response.create")
 	if !ok {
 		b.Fatalf("http.response.create processor not registered")
 	}

internal/processor/test/int-parse_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestIntParseFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["int.parse"]
+	registration, ok := processor.GetProcessorRegistration("int.parse")
 	if !ok {
 		t.Fatalf("int.parse processor not registered")
 	}
@@ -84,7 +84,7 @@ func TestGoodIntParse(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["int.parse"]
+			registration, ok := processor.GetProcessorRegistration("int.parse")
 			if !ok {
 				t.Fatalf("int.parse processor not registered")
 			}
@@ -170,7 +170,7 @@ func TestBadIntParse(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["int.parse"]
+			registration, ok := processor.GetProcessorRegistration("int.parse")
 			if !ok {
 				t.Fatalf("int.parse processor not registered")
 			}
@@ -201,7 +201,7 @@ func TestBadIntParse(t *testing.T) {
 }
 
 func BenchmarkIntParse(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["int.parse"]
+	registration, ok := processor.GetProcessorRegistration("int.parse")
 	if !ok {
 		b.Fatalf("int.parse processor not registered")
 	}

internal/processor/test/int-random_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestIntRandomFromRegistry(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["int.random"]
+	registration, ok := processor.GetProcessorRegistration("int.random")
 	if !ok {
 		t.Fatalf("int.random processor not registered")
 	}
@@ -32,7 +32,7 @@ func TestIntRandomFromRegistry(t *testing.T) {
 }
 
 func TestIntRandomGoodConfig(t *testing.T) {
-	registration, ok := processor.ProcessorRegistry["int.random"]
+	registration, ok := processor.GetProcessorRegistration("int.random")
 	if !ok {
 		t.Fatalf("int.random processor not registered")
 	}
@@ -84,7 +84,7 @@ func TestGoodIntRandom(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			registration, ok := processor.ProcessorRegistry["int.random"]
+			registration, ok := processor.GetProcessorRegistration("int.random")
 			if !ok {
 				t.Fatalf("int.random processor not registered")
 			}
@@ -166,7 +166,7 @@ func TestBadIntRandom(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			registration, ok := processor.ProcessorRegistry["int.random"]
+			registration, ok := processor.GetProcessorRegistration("int.random")
 			if !ok {
 				t.Fatalf("int.random processor not registered")
 			}
@@ -197,7 +197,7 @@ func TestBadIntRandom(t *testing.T) {
 }
 
 func BenchmarkIntRandom(b *testing.B) {
-	registration, ok := processor.ProcessorRegistry["int.random"]
+	registration, ok := processor.GetProcessorRegistration("int.random")
 	if !ok {
 		b.Fatalf("int.random processor not registered")
 	}