update cli usage in README

support responding to sip dtmf/audio calls

.vscode/launch.json

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

README.md

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

cmd/showbridge/main.go

@@ -3,9 +3,11 @@ package main
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"os/signal"
+	"slices"
 	"sync"
 
 	"github.com/jwetzell/showbridge-go"
@@ -29,15 +31,29 @@ func main() {
 				Value: "./config.yaml",
 				Usage: "path to config file",
 			},
-			&cli.BoolFlag{
+			&cli.StringFlag{
-				Name:  "debug",
+				Name:  "log-level",
-				Value: false,
+				Value: "info",
-				Usage: "set log level to DEBUG",
+				Usage: "set log level",
+				Validator: func(level string) error {
+					levels := []string{"debug", "info", "warn", "error"}
+					if !slices.Contains(levels, level) {
+						return fmt.Errorf("unknown log level: %s", level)
+					}
+					return nil
+				},
 			},
-			&cli.BoolFlag{
+			&cli.StringFlag{
-				Name:  "json",
+				Name:  "log-format",
-				Value: false,
+				Value: "text",
-				Usage: "log using JSON",
+				Usage: "log format to use",
+				Validator: func(format string) error {
+					formats := []string{"text", "json"}
+					if !slices.Contains(formats, format) {
+						return fmt.Errorf("unknown log format: %s", format)
+					}
+					return nil
+				},
 			},
 		},
 		Action: run,
@@ -83,8 +99,19 @@ func run(ctx context.Context, c *cli.Command) error {
 
 	logLevel := slog.LevelInfo
 
-	if c.Bool("debug") {
+	logLevelFromFlag := c.String("log-level")
+
+	switch logLevelFromFlag {
+	case "debug":
 		logLevel = slog.LevelDebug
+	case "info":
+		logLevel = slog.LevelInfo
+	case "warn":
+		logLevel = slog.LevelWarn
+	case "error":
+		logLevel = slog.LevelError
+	default:
+		logLevel = slog.LevelInfo
 	}
 
 	logHandlerOptions := &slog.HandlerOptions{
@@ -93,10 +120,17 @@ func run(ctx context.Context, c *cli.Command) error {
 
 	logOutput := os.Stderr
 
-	var logHandler slog.Handler = slog.NewTextHandler(logOutput, logHandlerOptions)
+	var logHandler slog.Handler
 
-	if c.Bool("json") {
+	logFormat := c.String("log-format")
+
+	switch logFormat {
+	case "json":
 		logHandler = slog.NewJSONHandler(logOutput, logHandlerOptions)
+	case "text":
+		logHandler = slog.NewTextHandler(logOutput, logHandlerOptions)
+	default:
+		logHandler = slog.NewTextHandler(logOutput, logHandlerOptions)
 	}
 
 	slog.SetDefault(slog.New(logHandler))

internal/module/http-client.go

@@ -73,7 +73,7 @@ func (hc *HTTPClient) Output(ctx context.Context, payload any) error {
 	}
 
 	if hc.router != nil {
-		hc.router.HandleInput(hc.Id(), response)
+		hc.router.HandleInput(hc.ctx, hc.Id(), response)
 	}
 
 	return nil

internal/module/http-server.go

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

internal/module/midi-input.go

@@ -69,7 +69,7 @@ func (mi *MIDIInput) Run() error {
 
 	stop, err := midi.ListenTo(in, func(msg midi.Message, timestampms int32) {
 		if mi.router != nil {
-			mi.router.HandleInput(mi.Id(), msg)
+			mi.router.HandleInput(mi.ctx, mi.Id(), msg)
 		}
 	}, midi.UseSysEx())
 

internal/module/mqtt-client.go

@@ -90,7 +90,7 @@ func (mc *MQTTClient) Run() error {
 
 	opts.OnConnect = func(c mqtt.Client) {
 		token := mc.client.Subscribe(mc.Topic, 1, func(c mqtt.Client, m mqtt.Message) {
-			mc.router.HandleInput(mc.Id(), m)
+			mc.router.HandleInput(mc.ctx, mc.Id(), m)
 		})
 		token.Wait()
 	}

internal/module/nats-client.go

@@ -83,7 +83,7 @@ func (nc *NATSClient) Run() error {
 
 	sub, err := nc.client.Subscribe(nc.Subject, func(msg *nats.Msg) {
 		if nc.router != nil {
-			nc.router.HandleInput(nc.Id(), msg)
+			nc.router.HandleInput(nc.ctx, nc.Id(), msg)
 		}
 	})
 

internal/module/psn-client.go

@@ -87,7 +87,7 @@ func (pc *PSNClient) Run() error {
 
 				if pc.router != nil {
 					for _, tracker := range pc.decoder.Trackers {
-						pc.router.HandleInput(pc.Id(), tracker)
+						pc.router.HandleInput(pc.ctx, pc.Id(), tracker)
 					}
 				} else {
 					pc.logger.Error("has no router")

internal/module/serial-client.go

@@ -155,7 +155,7 @@ func (sc *SerialClient) Run() error {
 							messages := sc.Framer.Decode(buffer[0:byteCount])
 							for _, message := range messages {
 								if sc.router != nil {
-									sc.router.HandleInput(sc.Id(), message)
+									sc.router.HandleInput(sc.ctx, sc.Id(), message)
 								} else {
 									sc.logger.Error("input received but no router is configured")
 								}

internal/module/sip-call-server.go

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

internal/module/sip-dtmf-server.go

@@ -3,9 +3,12 @@ package module
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
+	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/emiago/diago"
@@ -13,6 +16,7 @@ import (
 	"github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
 	"github.com/jwetzell/showbridge-go/internal/config"
+	"github.com/jwetzell/showbridge-go/internal/processor"
 	"github.com/jwetzell/showbridge-go/internal/route"
 )
 
@@ -32,6 +36,11 @@ type SIPDTMFMessage struct {
 	Digits string
 }
 
+type SIPDTMFCall struct {
+	inDialog *diago.DialogServerSession
+	lock     sync.Mutex
+}
+
 func init() {
 	RegisterModule(ModuleRegistration{
 		Type: "sip.dtmf.server",
@@ -148,10 +157,14 @@ func (sds *SIPDTMFServer) HandleCall(inDialog *diago.DialogServerSession) error
 
 	reader := inDialog.AudioReaderDTMF()
 	userString := ""
+
 	return reader.Listen(func(dtmf rune) error {
 		if dtmf == rune(sds.Separator[0]) {
 			if sds.router != nil {
-				sds.router.HandleInput(sds.Id(), SIPDTMFMessage{
+				dialogContext := context.WithValue(sds.ctx, sipCallContextKey("call"), &SIPDTMFCall{
+					inDialog: inDialog,
+				})
+				sds.router.HandleInput(dialogContext, sds.Id(), SIPDTMFMessage{
 					To:     inDialog.ToUser(),
 					Digits: userString,
 				})
@@ -165,5 +178,65 @@ func (sds *SIPDTMFServer) HandleCall(inDialog *diago.DialogServerSession) error
 }
 
 func (sds *SIPDTMFServer) Output(ctx context.Context, payload any) error {
-	return errors.New("sip.dtmf.server output is not implemented")
+	call, ok := ctx.Value(sipCallContextKey("call")).(*SIPDTMFCall)
+
+	if !ok {
+		return errors.New("sip.dtmf.server output must originate from sip.dtmf.server input")
+	}
+
+	gotLock := call.lock.TryLock()
+
+	if !gotLock {
+		return errors.New("sip.dtmf.server call is already locked")
+	}
+
+	if call.inDialog.LoadState() == sip.DialogStateEnded {
+		return errors.New("sip.dtmf.server inDialog already ended")
+	}
+
+	payloadDTMFResponse, ok := payload.(processor.SipDTMFResponse)
+
+	if ok {
+		dtmfWriter := call.inDialog.AudioWriterDTMF()
+
+		time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PreWait))
+		for i, dtmfRune := range payloadDTMFResponse.Digits {
+			err := dtmfWriter.WriteDTMF(dtmfRune)
+
+			if err != nil {
+				return fmt.Errorf("sip.dtmf.server error output dtmf digit at index %d", i)
+			}
+		}
+		time.Sleep(time.Millisecond * time.Duration(payloadDTMFResponse.PreWait))
+		return nil
+	}
+
+	payloadAudioFileResponse, ok := payload.(processor.SipAudioFileResponse)
+
+	if ok {
+		audioFile, err := os.Open(payloadAudioFileResponse.AudioFile)
+		if err != nil {
+			return err
+		}
+		defer audioFile.Close()
+
+		playback, err := call.inDialog.PlaybackCreate()
+
+		if err != nil {
+			return err
+		}
+
+		time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PreWait))
+
+		_, err = playback.Play(audioFile, "audio/wav")
+
+		time.Sleep(time.Millisecond * time.Duration(payloadAudioFileResponse.PostWait))
+
+		if err != nil {
+			return err
+		}
+		return nil
+	}
+
+	return errors.New("sip.dtmf.server can only output SipDTMFResponse or SipAudioFileResponse")
 }

internal/module/tcp-client.go

@@ -142,7 +142,7 @@ func (tc *TCPClient) Run() error {
 							messages := tc.framer.Decode(buffer[0:byteCount])
 							for _, message := range messages {
 								if tc.router != nil {
-									tc.router.HandleInput(tc.Id(), message)
+									tc.router.HandleInput(tc.ctx, tc.Id(), message)
 								} else {
 									tc.logger.Error("input received but no router is configured")
 								}

internal/module/tcp-server.go

@@ -155,7 +155,7 @@ ClientRead:
 					messages := ts.Framer.Decode(buffer[0:byteCount])
 					for _, message := range messages {
 						if ts.router != nil {
-							ts.router.HandleInput(ts.Id(), message)
+							ts.router.HandleInput(ts.ctx, ts.Id(), message)
 						} else {
 							ts.logger.Error("input received but no router is configured")
 						}

internal/module/time-interval.go

@@ -67,7 +67,7 @@ func (i *TimeInterval) Run() error {
 			return nil
 		case <-ticker.C:
 			if i.router != nil {
-				i.router.HandleInput(i.Id(), time.Now())
+				i.router.HandleInput(i.ctx, i.Id(), time.Now())
 			}
 		}
 	}

internal/module/time-timer.go

@@ -66,7 +66,7 @@ func (t *TimeTimer) Run() error {
 			return nil
 		case time := <-t.timer.C:
 			if t.router != nil {
-				t.router.HandleInput(t.Id(), time)
+				t.router.HandleInput(t.ctx, t.Id(), time)
 			}
 		}
 	}

internal/module/udp-multicast.go

@@ -105,7 +105,7 @@ func (um *UDPMulticast) Run() error {
 				message := buffer[:numBytes]
 
 				if um.router != nil {
-					um.router.HandleInput(um.Id(), message)
+					um.router.HandleInput(um.ctx, um.Id(), message)
 				} else {
 					um.logger.Error("input received but no router is configured")
 				}

internal/module/udp-server.go

@@ -115,7 +115,7 @@ func (us *UDPServer) Run() error {
 			}
 			message := buffer[:numBytes]
 			if us.router != nil {
-				us.router.HandleInput(us.Id(), message)
+				us.router.HandleInput(us.ctx, us.Id(), message)
 			} else {
 				us.logger.Error("input received but no router is configured")
 			}

internal/processor/http-response-create.go

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

internal/processor/json-encode.go

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

internal/processor/json-encode_test.go

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

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

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

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

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

internal/route/route.go

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

internal/route/route_test.go

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

router.go

@@ -3,7 +3,6 @@ package showbridge
 import (
 	"context"
 	"errors"
-	"fmt"
 	"log/slog"
 	"sync"
 
@@ -129,34 +128,69 @@ func (r *Router) Stop() {
 	r.contextCancel()
 }
 
-func (r *Router) HandleInput(sourceId string, payload any) []route.RouteIOError {
+func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []route.RouteIOError) {
-	var routingErrors []route.RouteIOError
+	var routeIOErrors []route.RouteIOError
+	routeFound := false
+
+	var routeWaitGroup sync.WaitGroup
+
 	for routeIndex, routeInstance := range r.RouteInstances {
 		if routeInstance.Input() == sourceId {
-			routeContext := context.WithValue(r.Context, route.SourceContextKey, sourceId)
+			routeWaitGroup.Go(func() {
 
-			payload, err := routeInstance.ProcessPayload(routeContext, payload)
+				routeFound = true
-			if err != nil {
+				routeContext := context.WithValue(ctx, route.SourceContextKey, sourceId)
-				if routingErrors == nil {
+
-					routingErrors = []route.RouteIOError{}
+				payload, err := routeInstance.ProcessPayload(routeContext, payload)
+				if err != nil {
+					if routeIOErrors == nil {
+						routeIOErrors = []route.RouteIOError{}
+					}
+					r.logger.Error("unable to process input", "route", routeIndex, "source", sourceId, "error", err)
+					routeIOErrors = append(routeIOErrors, route.RouteIOError{
+						Index:        routeIndex,
+						ProcessError: err,
+					})
+					return
 				}
-				routingErrors = append(routingErrors, route.RouteIOError{
+
-					Index: routeIndex,
+				if payload == nil {
-					Error: err,
+					r.logger.Error("no input after processing", "route", routeIndex, "source", sourceId)
-				})
+					return
-				r.logger.Error("unable to route input", "route", routeIndex, "source", sourceId, "error", err)
+				}
-			}
+
-			r.HandleOutput(routeContext, routeInstance.Output(), payload)
+				outputErrors := r.HandleOutput(routeContext, routeInstance.Output(), payload)
+				if outputErrors != nil {
+					if routeIOErrors == nil {
+						routeIOErrors = []route.RouteIOError{}
+					}
+					routeIOErrors = append(routeIOErrors, route.RouteIOError{
+						Index:        routeIndex,
+						OutputErrors: outputErrors,
+					})
+				}
+			})
+
 		}
 	}
-	return routingErrors
+	routeWaitGroup.Wait()
+	return routeFound, routeIOErrors
 }
 
-func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload any) error {
+func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload any) []error {
+
+	var outputErrors []error
 	for _, moduleInstance := range r.ModuleInstances {
 		if moduleInstance.Id() == destinationId {
-			return moduleInstance.Output(ctx, payload)
+			err := moduleInstance.Output(ctx, payload)
+			if err != nil {
+				if outputErrors == nil {
+					outputErrors = []error{}
+				}
+				outputErrors = append(outputErrors, err)
+				r.logger.Error("unable to route output", "module", moduleInstance.Id(), "error", err)
+			}
 		}
 	}
-	return fmt.Errorf("router could not find module instance for destination %s", destinationId)
+	return outputErrors
 }

router_test.go

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