Compare commits

..

25 Commits

Author SHA1 Message Date
Joel Wetzell
1f706becdc fix go version in dockerfile 2026-03-08 14:20:28 -05:00
Joel Wetzell
13a494239e Merge pull request #88 from jwetzell/remove-specific-filter-processors
remove specific filter processors
2026-03-08 14:13:11 -05:00
Joel Wetzell
153de944a2 remove filter processors 2026-03-08 14:06:28 -05:00
Joel Wetzell
42c75074fe update Go version 2026-03-08 13:38:52 -05:00
Joel Wetzell
361b07ec00 fix error message 2026-03-08 13:36:06 -05:00
Joel Wetzell
8c4f0591e1 Merge pull request #87 from jwetzell/feat/script-processor-env-data
consistent script processor environments
2026-03-08 13:34:01 -05:00
Joel Wetzell
09ddc40f1f consistent script processor environments 2026-03-08 13:32:34 -05:00
Joel Wetzell
6382cf6944 Merge pull request #86 from jwetzell/feat/struct-get-pointer
allow struct based processors to also operate on pointers to structs
2026-03-08 13:16:08 -05:00
Joel Wetzell
0732113a02 allow struct based processors to also operate on pointers to structs 2026-03-08 13:14:50 -05:00
Joel Wetzell
e5db9a48a9 Merge pull request #85 from jwetzell/feat/json-decode-bytes
allow json.decode to take byte slice and string
2026-03-08 13:11:17 -05:00
Joel Wetzell
9502201261 allow json.decode to take byte slice and string 2026-03-08 13:09:22 -05:00
Joel Wetzell
9859aa8fd4 Merge pull request #84 from jwetzell/feat/http-client-processor
make http client into a processor instead of module
2026-03-08 13:08:26 -05:00
Joel Wetzell
b7b05cbb77 make http client into a processor instead of module 2026-03-08 13:03:54 -05:00
Joel Wetzell
7b1fe47039 align variable names 2026-03-08 09:36:59 -05:00
Joel Wetzell
0848d8ab6e add descriptions to time module params 2026-03-08 09:36:31 -05:00
Joel Wetzell
c1bb2ff4cb update schema descriptions 2026-03-08 09:22:57 -05:00
Joel Wetzell
50898b9130 add missing title for params property 2026-03-07 11:49:29 -06:00
Joel Wetzell
05e68e90af add title and descriptions for params schemas 2026-03-06 18:59:27 -06:00
Joel Wetzell
cd4c3f59f2 fix missing properties on parse processors 2026-03-06 18:37:04 -06:00
Joel Wetzell
50399fa811 update titles on module schemas 2026-03-06 18:36:49 -06:00
Joel Wetzell
2dd44a11fd Merge pull request #83 from jwetzell/feat/router-output-as-processor
switch router output to be a processor instead of specific output per route
2026-03-04 21:34:10 -06:00
Joel Wetzell
b7a8b04a72 switch router output to be a processor instead of specific output per route 2026-03-04 21:21:11 -06:00
Joel Wetzell
078e6ec68c more detailed error message 2026-03-04 19:42:48 -06:00
Joel Wetzell
b58187d6a9 Merge pull request #82 from jwetzell/feat/sender-template-data
add sender to template data for relevant modules
2026-03-04 19:39:54 -06:00
Joel Wetzell
38b8e44f04 add sender to template data for relevant modules 2026-03-04 19:38:30 -06:00
62 changed files with 660 additions and 1473 deletions

View File

@@ -1,4 +1,4 @@
ARG GO_VERSION=1.25.5
ARG GO_VERSION=1.26.0
FROM golang:${GO_VERSION}-alpine AS build
RUN apk --no-cache add ca-certificates tzdata
WORKDIR /build

View File

@@ -13,6 +13,8 @@ routes:
processors:
- type: osc.message.create
params:
address: "{{.URL.Path}}"
address: "{{.Payload.URL.Path}}"
- type: osc.message.encode
output: udp
- type: router.output
params:
module: udp

2
go.mod
View File

@@ -1,6 +1,6 @@
module github.com/jwetzell/showbridge-go
go 1.25.5
go 1.26.0
require (
github.com/eclipse/paho.mqtt.golang v1.5.1

View File

@@ -1,7 +1,22 @@
package common
import "context"
type contextKey string
const RouterContextKey contextKey = contextKey("router")
const SourceContextKey contextKey = contextKey("source")
const ModulesContextKey contextKey = contextKey("modules")
const SenderContextKey contextKey = contextKey("sender")
type RouteIO interface {
HandleInput(ctx context.Context, sourceId string, payload any) (bool, []RouteIOError)
HandleOutput(ctx context.Context, destinationId string, payload any) error
}
type RouteIOError struct {
Index int
OutputError error
ProcessError error
InputError error
}

View File

@@ -273,7 +273,6 @@ type ModuleConfig struct {
type RouteConfig struct {
Input string `json:"input"`
Processors []ProcessorConfig `json:"processors"`
Output string `json:"output"`
}
type ProcessorConfig struct {

View File

@@ -1,95 +0,0 @@
package module
import (
"context"
"errors"
"log/slog"
"net/http"
"time"
"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/route"
)
type HTTPClient struct {
config config.ModuleConfig
ctx context.Context
client *http.Client
router route.RouteIO
logger *slog.Logger
cancel context.CancelFunc
}
func init() {
RegisterModule(ModuleRegistration{
Type: "http.client",
New: func(config config.ModuleConfig) (Module, error) {
return &HTTPClient{config: config, logger: CreateLogger(config)}, nil
},
})
}
func (hc *HTTPClient) Id() string {
return hc.config.Id
}
func (hc *HTTPClient) Type() string {
return hc.config.Type
}
func (hc *HTTPClient) Start(ctx context.Context) error {
hc.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
if !ok {
return errors.New("http.client unable to get router from context")
}
hc.router = router
moduleContext, cancel := context.WithCancel(ctx)
hc.ctx = moduleContext
hc.cancel = cancel
hc.client = &http.Client{
Timeout: 10 * time.Second,
}
<-hc.ctx.Done()
hc.logger.Debug("done")
return nil
}
func (hc *HTTPClient) Output(ctx context.Context, payload any) error {
payloadRequest, ok := processor.GetAnyAs[*http.Request](payload)
if !ok {
return errors.New("http.client is only able to output an http.Request")
}
if hc.client == nil {
return errors.New("http.client client is nil")
}
response, err := hc.client.Do(payloadRequest)
if err != nil {
return err
}
if hc.router != nil {
hc.router.HandleInput(hc.ctx, hc.Id(), response)
}
return nil
}
func (hc *HTTPClient) Stop() {
hc.cancel()
}
func (hc *HTTPClient) Get(key string) (any, error) {
return nil, errors.New("http.client does not support Get")
}

View File

@@ -6,19 +6,19 @@ import (
"errors"
"fmt"
"log/slog"
"net"
"net/http"
"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/route"
)
type HTTPServer struct {
config config.ModuleConfig
Port uint16
ctx context.Context
router route.RouteIO
router common.RouteIO
logger *slog.Logger
cancel context.CancelFunc
}
@@ -84,6 +84,10 @@ func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
if hs.router != nil {
inputContext := context.WithValue(hs.ctx, httpServerContextKey("responseWriter"), &responseWriter)
senderAddr, err := net.ResolveTCPAddr("tcp", r.RemoteAddr)
if err == nil {
inputContext = context.WithValue(inputContext, common.SenderContextKey, senderAddr)
}
aRouteFound, routingErrors := hs.router.HandleInput(inputContext, hs.Id(), r)
if !responseWriter.done {
if aRouteFound {
@@ -143,7 +147,7 @@ func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
func (hs *HTTPServer) Start(ctx context.Context) error {
hs.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("http.server unable to get router from context")

View File

@@ -10,7 +10,6 @@ import (
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/route"
"gitlab.com/gomidi/midi/v2"
_ "gitlab.com/gomidi/midi/v2/drivers/rtmididrv"
)
@@ -18,7 +17,7 @@ import (
type MIDIInput struct {
config config.ModuleConfig
ctx context.Context
router route.RouteIO
router common.RouteIO
Port string
SendFunc func(midi.Message) error
logger *slog.Logger
@@ -51,7 +50,7 @@ func (mi *MIDIInput) Type() string {
func (mi *MIDIInput) Start(ctx context.Context) error {
mi.logger.Debug("running")
defer midi.CloseDriver()
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("midi.input unable to get router from context")

View File

@@ -11,7 +11,6 @@ import (
"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/route"
"gitlab.com/gomidi/midi/v2"
_ "gitlab.com/gomidi/midi/v2/drivers/rtmididrv"
)
@@ -19,7 +18,7 @@ import (
type MIDIOutput struct {
config config.ModuleConfig
ctx context.Context
router route.RouteIO
router common.RouteIO
Port string
SendFunc func(midi.Message) error
logger *slog.Logger
@@ -53,7 +52,7 @@ func (mo *MIDIOutput) Type() string {
func (mo *MIDIOutput) Start(ctx context.Context) error {
mo.logger.Debug("running")
defer midi.CloseDriver()
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("midi.output unable to get router from context")

View File

@@ -10,13 +10,12 @@ import (
"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/route"
)
type MQTTClient struct {
config config.ModuleConfig
ctx context.Context
router route.RouteIO
router common.RouteIO
Broker string
ClientID string
Topic string
@@ -63,7 +62,7 @@ func (mc *MQTTClient) Type() string {
func (mc *MQTTClient) Start(ctx context.Context) error {
mc.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("mqtt.client unable to get router from context")

View File

@@ -8,14 +8,13 @@ import (
"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/route"
"github.com/nats-io/nats.go"
)
type NATSClient struct {
config config.ModuleConfig
ctx context.Context
router route.RouteIO
router common.RouteIO
URL string
Subject string
client *nats.Conn
@@ -54,7 +53,7 @@ func (nc *NATSClient) Type() string {
func (nc *NATSClient) Start(ctx context.Context) error {
nc.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("nats.client unable to get router from context")

View File

@@ -10,7 +10,6 @@ import (
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/route"
"github.com/nats-io/nats-server/v2/server"
)
@@ -19,7 +18,7 @@ type NATSServer struct {
ctx context.Context
Ip string
Port int
router route.RouteIO
router common.RouteIO
server *server.Server
logger *slog.Logger
cancel context.CancelFunc
@@ -67,7 +66,7 @@ func (ns *NATSServer) Type() string {
func (ns *NATSServer) Start(ctx context.Context) error {
ns.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("nats.server unable to get router from context")
@@ -104,7 +103,7 @@ func (ns *NATSServer) Start(ctx context.Context) error {
}
func (ns *NATSServer) Output(ctx context.Context, payload any) error {
return errors.ErrUnsupported
return errors.New("nats.server does not support output")
}
func (ns *NATSServer) Stop() {

View File

@@ -11,14 +11,13 @@ import (
"github.com/jwetzell/psn-go"
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/route"
)
type PSNClient struct {
config config.ModuleConfig
conn *net.UDPConn
ctx context.Context
router route.RouteIO
router common.RouteIO
decoder *psn.Decoder
logger *slog.Logger
cancel context.CancelFunc
@@ -44,7 +43,7 @@ func (pc *PSNClient) Type() string {
func (pc *PSNClient) Start(ctx context.Context) error {
pc.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("psn.client unable to get router from context")

View File

@@ -13,14 +13,13 @@ import (
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/framer"
"github.com/jwetzell/showbridge-go/internal/processor"
"github.com/jwetzell/showbridge-go/internal/route"
"go.bug.st/serial"
)
type SerialClient struct {
config config.ModuleConfig
ctx context.Context
router route.RouteIO
router common.RouteIO
Port string
Framer framer.Framer
Mode *serial.Mode
@@ -86,7 +85,7 @@ func (sc *SerialClient) SetupPort() error {
func (sc *SerialClient) Start(ctx context.Context) error {
sc.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("serial.client unable to get router from context")

View File

@@ -7,7 +7,6 @@ import (
"io"
"log/slog"
"os"
"path"
"sync"
"time"
@@ -18,18 +17,16 @@ import (
"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/route"
)
type SIPCallServer struct {
config config.ModuleConfig
ctx context.Context
router route.RouteIO
router common.RouteIO
IP string
Port int
Transport string
UserAgent string
RecordingPath string
dg *diago.Diago
logger *slog.Logger
cancel context.CancelFunc
@@ -88,19 +85,7 @@ func init() {
}
}
recordingPathString := ""
recordingPath, ok := params["recordingPath"]
if ok {
specificRecordingPath, ok := recordingPath.(string)
if !ok {
return nil, errors.New("sip.call.server recordingPath must be a string")
}
recordingPathString = specificRecordingPath
}
return &SIPCallServer{config: moduleConfig, IP: ipString, Port: int(portNum), Transport: transportString, UserAgent: userAgentString, RecordingPath: recordingPathString, logger: CreateLogger(moduleConfig)}, nil
return &SIPCallServer{config: moduleConfig, IP: ipString, Port: int(portNum), Transport: transportString, UserAgent: userAgentString, logger: CreateLogger(moduleConfig)}, nil
},
})
}
@@ -115,7 +100,7 @@ func (scs *SIPCallServer) Type() string {
func (scs *SIPCallServer) Start(ctx context.Context) error {
scs.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("sip.call.server unable to get router from context")
@@ -165,41 +150,6 @@ func (scs *SIPCallServer) HandleCall(inDialog *diago.DialogServerSession) {
dialogContext := context.WithValue(scs.ctx, sipCallContextKey("call"), &SIPCall{
inDialog: inDialog,
})
if scs.RecordingPath != "" {
filename := path.Join(scs.RecordingPath, fmt.Sprintf("%s-%d.wav", inDialog.ToUser(), time.Now().Unix()))
wavFile, err := os.OpenFile(filename, os.O_RDWR|os.O_CREATE, 0755)
if err != nil {
scs.logger.Error("error creating recording file", "error", err, "filename", filename)
} else {
recorder, err := inDialog.AudioStereoRecordingCreate(wavFile)
if err != nil {
scs.logger.Error("error creating recording", "error", err)
wavFile.Close()
} else {
go func() {
<-inDialog.Context().Done()
err := recorder.Close()
if err != nil {
scs.logger.Error("error closing recording", "error", err)
}
wavFile.Close()
scs.logger.Debug("finished recording", "filename", filename)
}()
go func() {
bytes, err := media.Copy(recorder.AudioReader(), recorder.AudioWriter())
fmt.Println("recorded bytes", bytes)
if err != nil {
if !errors.Is(err, io.EOF) {
scs.logger.Error("error while recording", "error", err)
}
}
}()
}
}
}
scs.router.HandleInput(dialogContext, scs.Id(), SIPCallMessage{
To: inDialog.ToUser(),
})

View File

@@ -18,13 +18,12 @@ import (
"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/route"
)
type SIPDTMFServer struct {
config config.ModuleConfig
ctx context.Context
router route.RouteIO
router common.RouteIO
IP string
Port int
Transport string
@@ -114,7 +113,7 @@ func (sds *SIPDTMFServer) Type() string {
func (sds *SIPDTMFServer) Start(ctx context.Context) error {
sds.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("sip.dtmf.server unable to get router from context")

View File

@@ -12,7 +12,6 @@ import (
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/framer"
"github.com/jwetzell/showbridge-go/internal/processor"
"github.com/jwetzell/showbridge-go/internal/route"
)
type TCPClient struct {
@@ -20,7 +19,7 @@ type TCPClient struct {
framer framer.Framer
conn *net.TCPConn
ctx context.Context
router route.RouteIO
router common.RouteIO
Addr *net.TCPAddr
logger *slog.Logger
cancel context.CancelFunc
@@ -71,7 +70,7 @@ func (tc *TCPClient) Type() string {
func (tc *TCPClient) Start(ctx context.Context) error {
tc.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("net.tcp.client unable to get router from context")

View File

@@ -15,7 +15,6 @@ import (
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/framer"
"github.com/jwetzell/showbridge-go/internal/processor"
"github.com/jwetzell/showbridge-go/internal/route"
)
type TCPServer struct {
@@ -23,7 +22,7 @@ type TCPServer struct {
Addr *net.TCPAddr
Framer framer.Framer
ctx context.Context
router route.RouteIO
router common.RouteIO
quit chan interface{}
wg sync.WaitGroup
connections []*net.TCPConn
@@ -142,7 +141,13 @@ ClientRead:
messages := ts.Framer.Decode(buffer[0:byteCount])
for _, message := range messages {
if ts.router != nil {
senderAddr, ok := client.RemoteAddr().(*net.TCPAddr)
if ok {
senderCtx := context.WithValue(ts.ctx, common.SenderContextKey, senderAddr)
ts.router.HandleInput(senderCtx, ts.Id(), message)
} else {
ts.router.HandleInput(ts.ctx, ts.Id(), message)
}
} else {
ts.logger.Error("input received but no router is configured")
}
@@ -155,7 +160,7 @@ ClientRead:
func (ts *TCPServer) Start(ctx context.Context) error {
ts.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("net.tcp.server unable to get router from context")

View File

@@ -1,73 +0,0 @@
package module_test
import (
"testing"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/module"
)
func TestHTTPClientFromRegistry(t *testing.T) {
registration, ok := module.ModuleRegistry["http.client"]
if !ok {
t.Fatalf("http.client module not registered")
}
moduleInstance, err := registration.New(config.ModuleConfig{
Id: "test",
Type: "http.client",
})
if err != nil {
t.Fatalf("failed to create http.client module: %s", err)
}
if moduleInstance.Id() != "test" {
t.Fatalf("http.client module has wrong id: %s", moduleInstance.Id())
}
if moduleInstance.Type() != "http.client" {
t.Fatalf("http.client module has wrong type: %s", moduleInstance.Type())
}
}
func TestBadHTTPClient(t *testing.T) {
tests := []struct {
name string
params map[string]any
errorString string
}{}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
registration, ok := module.ModuleRegistry["http.client"]
if !ok {
t.Fatalf("http.client module not registered")
}
moduleInstance, err := registration.New(config.ModuleConfig{
Id: "test",
Type: "http.client",
Params: test.params,
})
if err != nil {
if test.errorString != err.Error() {
t.Fatalf("http.client got error '%s', expected '%s'", err.Error(), test.errorString)
}
return
}
err = moduleInstance.Start(t.Context())
if err == nil {
t.Fatalf("http.client expected to fail")
}
if err.Error() != test.errorString {
t.Fatalf("http.client got error '%s', expected '%s'", err.Error(), test.errorString)
}
})
}
}

View File

@@ -9,14 +9,13 @@ import (
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/route"
)
type TimeInterval struct {
config config.ModuleConfig
Duration uint32
ctx context.Context
router route.RouteIO
router common.RouteIO
ticker *time.Ticker
logger *slog.Logger
cancel context.CancelFunc
@@ -47,7 +46,7 @@ func (i *TimeInterval) Type() string {
func (i *TimeInterval) Start(ctx context.Context) error {
i.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("time.interval unable to get router from context")

View File

@@ -9,14 +9,13 @@ import (
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/route"
)
type TimeTimer struct {
config config.ModuleConfig
Duration uint32
ctx context.Context
router route.RouteIO
router common.RouteIO
timer *time.Timer
logger *slog.Logger
cancel context.CancelFunc
@@ -48,7 +47,7 @@ func (t *TimeTimer) Type() string {
func (t *TimeTimer) Start(ctx context.Context) error {
t.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("net.tcp.client unable to get router from context")

View File

@@ -10,7 +10,6 @@ import (
"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/route"
)
type UDPClient struct {
@@ -19,7 +18,7 @@ type UDPClient struct {
Port uint16
conn *net.UDPConn
ctx context.Context
router route.RouteIO
router common.RouteIO
logger *slog.Logger
cancel context.CancelFunc
}
@@ -64,7 +63,7 @@ func (uc *UDPClient) SetupConn() error {
func (uc *UDPClient) Start(ctx context.Context) error {
uc.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("net.udp.client unable to get router from context")

View File

@@ -11,14 +11,13 @@ import (
"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/route"
)
type UDPMulticast struct {
config config.ModuleConfig
conn *net.UDPConn
ctx context.Context
router route.RouteIO
router common.RouteIO
Addr *net.UDPAddr
logger *slog.Logger
cancel context.CancelFunc
@@ -58,7 +57,7 @@ func (um *UDPMulticast) Type() string {
func (um *UDPMulticast) Start(ctx context.Context) error {
um.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("net.udp.multicast unable to get router from context")

View File

@@ -10,7 +10,6 @@ import (
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/route"
)
type UDPServer struct {
@@ -18,7 +17,7 @@ type UDPServer struct {
BufferSize int
config config.ModuleConfig
ctx context.Context
router route.RouteIO
router common.RouteIO
logger *slog.Logger
cancel context.CancelFunc
}
@@ -70,7 +69,7 @@ func (us *UDPServer) Type() string {
func (us *UDPServer) Start(ctx context.Context) error {
us.logger.Debug("running")
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return errors.New("net.udp.server unable to get router from context")
@@ -97,7 +96,7 @@ func (us *UDPServer) Start(ctx context.Context) error {
default:
listener.SetDeadline(time.Now().Add(time.Millisecond * 200))
numBytes, _, err := listener.ReadFromUDP(buffer)
numBytes, senderAddr, err := listener.ReadFromUDP(buffer)
if err != nil {
//NOTE(jwetzell) we hit deadline
if opErr, ok := err.(*net.OpError); ok && opErr.Timeout() {
@@ -107,7 +106,8 @@ func (us *UDPServer) Start(ctx context.Context) error {
}
message := buffer[:numBytes]
if us.router != nil {
us.router.HandleInput(us.ctx, us.Id(), message)
senderCtx := context.WithValue(us.ctx, common.SenderContextKey, senderAddr)
us.router.HandleInput(senderCtx, us.Id(), message)
} else {
us.logger.Error("input received but no router is configured")
}

View File

@@ -1,48 +0,0 @@
package processor
import (
"context"
"fmt"
"github.com/jwetzell/artnet-go"
"github.com/jwetzell/showbridge-go/internal/config"
)
type ArtNetPacketFilter struct {
config config.ProcessorConfig
OpCode uint16
}
func (apf *ArtNetPacketFilter) Process(ctx context.Context, payload any) (any, error) {
payloadPacket, ok := GetAnyAs[artnet.ArtNetPacket](payload)
if !ok {
return nil, fmt.Errorf("artnet.packet.filter processor only accepts an ArtNetPacket")
}
if payloadPacket.GetOpCode() != apf.OpCode {
return nil, nil
}
return payloadPacket, nil
}
func (apf *ArtNetPacketFilter) Type() string {
return apf.config.Type
}
func init() {
RegisterProcessor(ProcessorRegistration{
Type: "artnet.packet.filter",
New: func(config config.ProcessorConfig) (Processor, error) {
params := config.Params
opCodeNum, err := params.GetInt("opCode")
if err != nil {
return nil, fmt.Errorf("artnet.packet.filter opCode error: %w", err)
}
return &ArtNetPacketFilter{config: config, OpCode: uint16(opCodeNum)}, nil
},
})
}

View File

@@ -15,23 +15,11 @@ type FilterExpr struct {
Program *vm.Program
}
func SafeExprEnv(payload any) any {
exprEnv := ExprEnv{
Payload: payload,
}
func (fe *FilterExpr) Process(ctx context.Context, payload any) (any, error) {
return exprEnv
}
exprEnv := GetEnvData(ctx, payload)
type ExprEnv struct {
Payload any
}
func (se *FilterExpr) Process(ctx context.Context, payload any) (any, error) {
exprEnv := SafeExprEnv(payload)
output, err := expr.Run(se.Program, exprEnv)
output, err := expr.Run(fe.Program, exprEnv)
if err != nil {
return nil, err
}
@@ -47,8 +35,8 @@ func (se *FilterExpr) Process(ctx context.Context, payload any) (any, error) {
return payload, nil
}
func (se *FilterExpr) Type() string {
return se.config.Type
func (fe *FilterExpr) Type() string {
return fe.config.Type
}
func init() {

View File

@@ -12,7 +12,7 @@ type FreeDDecode struct {
config config.ProcessorConfig
}
func (fdd *FreeDDecode) Process(ctx context.Context, payload any) (any, error) {
func (fd *FreeDDecode) Process(ctx context.Context, payload any) (any, error) {
payloadBytes, ok := GetAnyAs[[]byte](payload)
if !ok {
@@ -26,8 +26,8 @@ func (fdd *FreeDDecode) Process(ctx context.Context, payload any) (any, error) {
return payloadMessage, nil
}
func (fdd *FreeDDecode) Type() string {
return fdd.config.Type
func (fd *FreeDDecode) Type() string {
return fd.config.Type
}
func init() {

View File

@@ -12,7 +12,7 @@ type FreeDEncode struct {
config config.ProcessorConfig
}
func (fde *FreeDEncode) Process(ctx context.Context, payload any) (any, error) {
func (fe *FreeDEncode) Process(ctx context.Context, payload any) (any, error) {
payloadPosition, ok := GetAnyAs[freeD.FreeDPosition](payload)
if !ok {
@@ -23,8 +23,8 @@ func (fde *FreeDEncode) Process(ctx context.Context, payload any) (any, error) {
return payloadBytes, nil
}
func (fde *FreeDEncode) Type() string {
return fde.config.Type
func (fe *FreeDEncode) Type() string {
return fe.config.Type
}
func init() {

View File

@@ -1,70 +0,0 @@
package processor
import (
"bytes"
"context"
"fmt"
"net/http"
"text/template"
"github.com/jwetzell/showbridge-go/internal/config"
)
type HTTPRequestCreate struct {
config config.ProcessorConfig
Method string
URL *template.Template
}
func (hrc *HTTPRequestCreate) Process(ctx context.Context, payload any) (any, error) {
templateData := GetTemplateData(ctx, payload)
var urlBuffer bytes.Buffer
err := hrc.URL.Execute(&urlBuffer, templateData)
if err != nil {
return nil, err
}
urlString := urlBuffer.String()
//TODO(jwetzell): support body
request, err := http.NewRequest(hrc.Method, urlString, bytes.NewBuffer([]byte{}))
if err != nil {
return nil, err
}
return request, nil
}
func (hrc *HTTPRequestCreate) Type() string {
return hrc.config.Type
}
func init() {
RegisterProcessor(ProcessorRegistration{
Type: "http.request.create",
New: func(config config.ProcessorConfig) (Processor, error) {
params := config.Params
methodString, err := params.GetString("method")
if err != nil {
return nil, fmt.Errorf("http.request.create method error: %w", err)
}
urlString, err := params.GetString("url")
if err != nil {
return nil, fmt.Errorf("http.request.create url error: %w", err)
}
urlTemplate, err := template.New("url").Parse(urlString)
if err != nil {
return nil, err
}
return &HTTPRequestCreate{config: config, URL: urlTemplate, Method: methodString}, nil
},
})
}

View File

@@ -0,0 +1,93 @@
package processor
import (
"bytes"
"context"
"fmt"
"io"
"net/http"
"text/template"
"time"
"github.com/jwetzell/showbridge-go/internal/config"
)
type HTTPRequestDo struct {
config config.ProcessorConfig
client *http.Client
Method string
URL *template.Template
}
func (hrd *HTTPRequestDo) Process(ctx context.Context, payload any) (any, error) {
templateData := GetTemplateData(ctx, payload)
var urlBuffer bytes.Buffer
err := hrd.URL.Execute(&urlBuffer, templateData)
if err != nil {
return nil, err
}
urlString := urlBuffer.String()
//TODO(jwetzell): support body
request, err := http.NewRequest(hrd.Method, urlString, bytes.NewBuffer([]byte{}))
if err != nil {
return nil, err
}
response, err := hrd.client.Do(request)
if err != nil {
return nil, err
}
body, err := io.ReadAll(response.Body)
if err != nil {
return nil, err
}
//TODO(jwetzell): support headers, etc
return HTTPResponse{
Status: response.StatusCode,
Body: body,
}, nil
}
func (hrd *HTTPRequestDo) Type() string {
return hrd.config.Type
}
func init() {
RegisterProcessor(ProcessorRegistration{
Type: "http.request.do",
New: func(config config.ProcessorConfig) (Processor, error) {
params := config.Params
methodString, err := params.GetString("method")
if err != nil {
return nil, fmt.Errorf("http.request.do method error: %w", err)
}
urlString, err := params.GetString("url")
if err != nil {
return nil, fmt.Errorf("http.request.do url error: %w", err)
}
urlTemplate, err := template.New("url").Parse(urlString)
if err != nil {
return nil, err
}
client := &http.Client{
Timeout: 10 * time.Second,
}
return &HTTPRequestDo{config: config, URL: urlTemplate, Method: methodString, client: client}, nil
},
})
}

View File

@@ -1,73 +0,0 @@
package processor
import (
"context"
"errors"
"fmt"
"net/http"
"regexp"
"github.com/jwetzell/showbridge-go/internal/config"
)
type HTTPRequestFilter struct {
config config.ProcessorConfig
Path *regexp.Regexp
Method string
}
func (hrf *HTTPRequestFilter) Process(ctx context.Context, payload any) (any, error) {
payloadRequest, ok := GetAnyAs[*http.Request](payload)
if !ok {
return nil, errors.New("http.request.filter can only operate on http.Request payloads")
}
if hrf.Method != "" {
if payloadRequest.Method != hrf.Method {
return nil, nil
}
}
if !hrf.Path.MatchString(payloadRequest.URL.Path) {
return nil, nil
}
return payloadRequest, nil
}
func (hrf *HTTPRequestFilter) Type() string {
return hrf.config.Type
}
func init() {
RegisterProcessor(ProcessorRegistration{
Type: "http.request.filter",
New: func(moduleConfig config.ProcessorConfig) (Processor, error) {
params := moduleConfig.Params
pathString, err := params.GetString("path")
if err != nil {
return nil, fmt.Errorf("http.request.filter path error: %w", err)
}
pathRegexp, err := regexp.Compile(fmt.Sprintf("^%s$", pathString))
if err != nil {
return nil, err
}
methodString, err := params.GetString("method")
if err != nil {
if errors.Is(err, config.ErrParamNotFound) {
return &HTTPRequestFilter{config: moduleConfig, Path: pathRegexp}, nil
} else {
return nil, fmt.Errorf("http.request.filter method error: %w", err)
}
}
return &HTTPRequestFilter{config: moduleConfig, Path: pathRegexp, Method: methodString}, nil
},
})
}

View File

@@ -20,24 +20,24 @@ type HTTPResponse struct {
Body []byte
}
func (hre *HTTPResponseCreate) Process(ctx context.Context, payload any) (any, error) {
func (hrc *HTTPResponseCreate) Process(ctx context.Context, payload any) (any, error) {
templateData := GetTemplateData(ctx, payload)
var bodyBuffer bytes.Buffer
err := hre.BodyTmpl.Execute(&bodyBuffer, templateData)
err := hrc.BodyTmpl.Execute(&bodyBuffer, templateData)
if err != nil {
return nil, err
}
return HTTPResponse{
Status: hre.Status,
Status: hrc.Status,
Body: bodyBuffer.Bytes(),
}, nil
}
func (hre *HTTPResponseCreate) Type() string {
return hre.config.Type
func (hrc *HTTPResponseCreate) Type() string {
return hrc.config.Type
}
func init() {

View File

@@ -13,15 +13,20 @@ type JsonDecode struct {
}
func (jd *JsonDecode) Process(ctx context.Context, payload any) (any, error) {
payloadString, ok := GetAnyAs[string](payload)
payloadBytes, ok := GetAnyAsByteSlice(payload)
if !ok {
return nil, errors.New("json.decode processor only accepts a string")
payloadString, ok := GetAnyAs[string](payload)
if !ok {
return nil, errors.New("json.decode can only process a string or []byte")
}
payloadBytes = []byte(payloadString)
}
payloadJson := make(map[string]any)
err := json.Unmarshal([]byte(payloadString), &payloadJson)
err := json.Unmarshal(payloadBytes, &payloadJson)
if err != nil {
return nil, err
}

View File

@@ -1,50 +0,0 @@
//go:build cgo
package processor
import (
"context"
"errors"
"fmt"
"github.com/jwetzell/showbridge-go/internal/config"
"gitlab.com/gomidi/midi/v2"
)
type MIDIMessageFilter struct {
config config.ProcessorConfig
MIDIType string
}
func (mmf *MIDIMessageFilter) Process(ctx context.Context, payload any) (any, error) {
payloadMessage, ok := GetAnyAs[midi.Message](payload)
if !ok {
return nil, errors.New("midi.message.filter processor only accepts a midi.Message")
}
if payloadMessage.Type().String() != mmf.MIDIType {
return nil, nil
}
return payloadMessage, nil
}
func (mmf *MIDIMessageFilter) Type() string {
return mmf.config.Type
}
func init() {
RegisterProcessor(ProcessorRegistration{
Type: "midi.message.filter",
New: func(config config.ProcessorConfig) (Processor, error) {
params := config.Params
msgTypeString, err := params.GetString("type")
if err != nil {
return nil, fmt.Errorf("midi.message.filter type error: %w", err)
}
return &MIDIMessageFilter{config: config, MIDIType: msgTypeString}, nil
},
})
}

View File

@@ -1,64 +0,0 @@
package processor
import (
"context"
"errors"
"fmt"
"regexp"
"strings"
"github.com/jwetzell/osc-go"
"github.com/jwetzell/showbridge-go/internal/config"
)
type OSCMessageFilter struct {
config config.ProcessorConfig
Address *regexp.Regexp
}
func (omf *OSCMessageFilter) Process(ctx context.Context, payload any) (any, error) {
payloadMessage, ok := GetAnyAs[osc.OSCMessage](payload)
if !ok {
return nil, errors.New("osc.message.filter can only operate on OSCMessage payloads")
}
if !omf.Address.MatchString(payloadMessage.Address) {
return nil, nil
}
return payloadMessage, nil
}
func (omf *OSCMessageFilter) Type() string {
return omf.config.Type
}
func init() {
RegisterProcessor(ProcessorRegistration{
Type: "osc.message.filter",
New: func(config config.ProcessorConfig) (Processor, error) {
params := config.Params
addressString, err := params.GetString("address")
if err != nil {
return nil, fmt.Errorf("osc.message.filter address error: %w", err)
}
addressPattern := strings.ReplaceAll(addressString, "?", ".")
addressPattern = strings.ReplaceAll(addressPattern, "*", "[^/]*")
addressPattern = strings.ReplaceAll(addressPattern, "[!", "[^")
addressPattern = strings.ReplaceAll(addressPattern, "{", "(")
addressPattern = strings.ReplaceAll(addressPattern, "}", ")")
addressPattern = strings.ReplaceAll(addressPattern, ",", "|")
addressPatternRegexp, err := regexp.Compile(fmt.Sprintf("^%s$", addressPattern))
if err != nil {
return nil, err
}
return &OSCMessageFilter{config: config, Address: addressPatternRegexp}, nil
},
})
}

View File

@@ -3,6 +3,8 @@ package processor
import (
"context"
"fmt"
"math"
"reflect"
"sync"
"github.com/jwetzell/showbridge-go/internal/common"
@@ -47,9 +49,61 @@ func GetAnyAs[T any](p any) (T, bool) {
return typed, ok
}
func GetAnyAsByteSlice(p any) ([]byte, bool) {
v := reflect.ValueOf(p)
if v.Kind() != reflect.Slice {
return nil, false
}
result := make([]byte, v.Len())
for i := 0; i < v.Len(); i++ {
elem := v.Index(i).Interface()
byteValue, ok := elem.(byte)
if ok {
result[i] = byteValue
continue
}
uintValue, ok := elem.(uint)
if ok {
if uintValue > 255 {
return nil, false
}
result[i] = byte(uintValue)
continue
}
intValue, ok := elem.(int)
if ok {
if intValue < 0 || intValue > 255 {
return nil, false
}
result[i] = byte(intValue)
continue
}
floatValue, ok := elem.(float64)
if ok {
if floatValue != math.Floor(floatValue) {
return nil, false
}
if floatValue < 0 || floatValue > 255 {
return nil, false
}
result[i] = byte(floatValue)
continue
}
return nil, false
}
return result, true
}
type TemplateData struct {
Payload any
Modules any
Sender any
}
type EnvData struct {
Payload any
Sender any
}
func GetTemplateData(ctx context.Context, payload any) TemplateData {
@@ -58,5 +112,20 @@ func GetTemplateData(ctx context.Context, payload any) TemplateData {
if modules != nil {
templateData.Modules = modules
}
sender := ctx.Value(common.SenderContextKey)
if sender != nil {
templateData.Sender = sender
}
return templateData
}
func GetEnvData(ctx context.Context, payload any) EnvData {
envData := EnvData{Payload: payload}
sender := ctx.Value(common.SenderContextKey)
if sender != nil {
envData.Sender = sender
}
return envData
}

View File

@@ -0,0 +1,55 @@
package processor
import (
"context"
"errors"
"fmt"
"log/slog"
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
)
type RouterOutput struct {
config config.ProcessorConfig
ModuleId string
logger *slog.Logger
}
func (ro *RouterOutput) Process(ctx context.Context, payload any) (any, error) {
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return nil, errors.New("router.output no router found")
}
err := router.HandleOutput(ctx, ro.ModuleId, payload)
if err != nil {
return nil, fmt.Errorf("router.output failed to send output: %w", err)
}
return payload, nil
}
func (ro *RouterOutput) Type() string {
return ro.config.Type
}
func init() {
RegisterProcessor(ProcessorRegistration{
Type: "router.output",
New: func(config config.ProcessorConfig) (Processor, error) {
params := config.Params
moduleId, err := params.GetString("module")
if err != nil {
return nil, fmt.Errorf("router.output module error: %w", err)
}
return &RouterOutput{config: config, ModuleId: moduleId, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
},
})
}

View File

@@ -17,7 +17,7 @@ type ScriptExpr struct {
func (se *ScriptExpr) Process(ctx context.Context, payload any) (any, error) {
exprEnv := SafeExprEnv(payload)
exprEnv := GetEnvData(ctx, payload)
output, err := expr.Run(se.Program, exprEnv)
if err != nil {

View File

@@ -5,6 +5,7 @@ import (
"encoding/json"
"fmt"
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"modernc.org/quickjs"
)
@@ -30,6 +31,16 @@ func (sj *ScriptJS) Process(ctx context.Context, payload any) (any, error) {
vm.SetProperty(vm.GlobalObject(), payloadAtom, payload)
sender := ctx.Value(common.SenderContextKey)
if sender != nil {
senderAtom, err := vm.NewAtom("sender")
if err != nil {
return nil, err
}
vm.SetProperty(vm.GlobalObject(), senderAtom, sender)
}
_, err = vm.Eval(sj.Program, quickjs.EvalGlobal)
if err != nil {

View File

@@ -15,21 +15,21 @@ type ScriptWASM struct {
Function string
}
func (se *ScriptWASM) Process(ctx context.Context, payload any) (any, error) {
func (sw *ScriptWASM) Process(ctx context.Context, payload any) (any, error) {
payloadBytes, ok := GetAnyAs[[]byte](payload)
if !ok {
return nil, fmt.Errorf("script.wasm can only operator on byte array")
return nil, fmt.Errorf("script.wasm can only process a byte array")
}
program, err := se.Program.Instance(ctx, extism.PluginInstanceConfig{})
program, err := sw.Program.Instance(ctx, extism.PluginInstanceConfig{})
if err != nil {
return nil, err
}
_, output, err := program.Call(se.Function, payloadBytes)
_, output, err := program.Call(sw.Function, payloadBytes)
if err != nil {
return nil, err
@@ -38,8 +38,8 @@ func (se *ScriptWASM) Process(ctx context.Context, payload any) (any, error) {
return output, nil
}
func (se *ScriptWASM) Type() string {
return se.config.Type
func (sw *ScriptWASM) Type() string {
return sw.config.Type
}
func init() {

View File

@@ -22,12 +22,12 @@ type SipAudioFileResponse struct {
AudioFile string
}
func (scc *SipResponseAudioCreate) Process(ctx context.Context, payload any) (any, error) {
func (srac *SipResponseAudioCreate) Process(ctx context.Context, payload any) (any, error) {
templateData := GetTemplateData(ctx, payload)
var audioFileBuffer bytes.Buffer
err := scc.AudioFile.Execute(&audioFileBuffer, templateData)
err := srac.AudioFile.Execute(&audioFileBuffer, templateData)
if err != nil {
return nil, err
@@ -36,14 +36,14 @@ func (scc *SipResponseAudioCreate) Process(ctx context.Context, payload any) (an
audioFileString := audioFileBuffer.String()
return SipAudioFileResponse{
PreWait: scc.PreWait,
PostWait: scc.PostWait,
PreWait: srac.PreWait,
PostWait: srac.PostWait,
AudioFile: audioFileString,
}, nil
}
func (scc *SipResponseAudioCreate) Type() string {
return scc.config.Type
func (srac *SipResponseAudioCreate) Type() string {
return srac.config.Type
}
func init() {

View File

@@ -25,12 +25,12 @@ type SipDTMFResponse struct {
Digits string
}
func (scc *SipResponseDTMFCreate) Process(ctx context.Context, payload any) (any, error) {
func (srdc *SipResponseDTMFCreate) Process(ctx context.Context, payload any) (any, error) {
templateData := GetTemplateData(ctx, payload)
var digitsBuffer bytes.Buffer
err := scc.Digits.Execute(&digitsBuffer, templateData)
err := srdc.Digits.Execute(&digitsBuffer, templateData)
if err != nil {
return nil, err
@@ -38,19 +38,19 @@ func (scc *SipResponseDTMFCreate) Process(ctx context.Context, payload any) (any
digitsString := digitsBuffer.String()
if !scc.validDTMF.MatchString(digitsString) {
if !srdc.validDTMF.MatchString(digitsString) {
return nil, errors.New("sip.response.dtmf.create result of digits template contains invalid characters")
}
return SipDTMFResponse{
PreWait: scc.PreWait,
PostWait: scc.PostWait,
PreWait: srdc.PreWait,
PostWait: srdc.PostWait,
Digits: digitsString,
}, nil
}
func (scc *SipResponseDTMFCreate) Type() string {
return scc.config.Type
func (srdc *SipResponseDTMFCreate) Type() string {
return srdc.config.Type
}
func init() {

View File

@@ -18,8 +18,12 @@ func (sf *StructFieldGet) Process(ctx context.Context, payload any) (any, error)
s := reflect.ValueOf(payload)
if s.Kind() != reflect.Struct {
if s.Kind() == reflect.Pointer && s.Elem().Kind() == reflect.Struct {
s = s.Elem()
} else {
return nil, errors.New("struct.field.get processor only accepts a struct payload")
}
}
field := s.FieldByName(sf.Name)
if !field.IsValid() {

View File

@@ -18,8 +18,12 @@ func (sm *StructMethodGet) Process(ctx context.Context, payload any) (any, error
s := reflect.ValueOf(payload)
if s.Kind() != reflect.Struct {
if s.Kind() == reflect.Pointer && s.Elem().Kind() == reflect.Struct {
s = s.Elem()
} else {
return nil, errors.New("struct.method.get processor only accepts a struct payload")
}
}
method := s.MethodByName(sm.Name)
if !method.IsValid() {

View File

@@ -1,254 +0,0 @@
package processor_test
import (
"reflect"
"testing"
"github.com/jwetzell/artnet-go"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/processor"
)
func TestArtnetPacketFilterFromRegistry(t *testing.T) {
registration, ok := processor.ProcessorRegistry["artnet.packet.filter"]
if !ok {
t.Fatalf("artnet.packet.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "artnet.packet.filter",
Params: map[string]any{
"opCode": float64(artnet.OpTimeCode),
},
})
if err != nil {
t.Fatalf("failed to create artnet.packet.filter processor: %s", err)
}
if processorInstance.Type() != "artnet.packet.filter" {
t.Fatalf("artnet.packet.filter processor has wrong type: %s", processorInstance.Type())
}
payload := &artnet.ArtTimeCode{
ID: []byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
OpCode: artnet.OpTimeCode,
ProtVerHi: 0,
ProtVerLo: 14,
Filler1: 0,
StreamId: 0,
Frames: 11,
Seconds: 17,
Minutes: 3,
Hours: 0,
Type: 0,
}
expected := &artnet.ArtTimeCode{
ID: []byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
OpCode: artnet.OpTimeCode,
ProtVerHi: 0,
ProtVerLo: 14,
Filler1: 0,
StreamId: 0,
Frames: 11,
Seconds: 17,
Minutes: 3,
Hours: 0,
Type: 0,
}
got, err := processorInstance.Process(t.Context(), payload)
if err != nil {
t.Fatalf("artnet.packet.filter processing failed: %s", err)
}
if !reflect.DeepEqual(got, expected) {
t.Fatalf("artnet.packet.filter got %+v, expected %+v", got, expected)
}
}
func TestGoodArtnetPacketFilter(t *testing.T) {
tests := []struct {
name string
params map[string]any
payload any
expected artnet.ArtNetPacket
}{
{
name: "tiemcode packet with matching opCode",
params: map[string]any{
"opCode": float64(artnet.OpTimeCode),
},
payload: &artnet.ArtTimeCode{
ID: []byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
OpCode: artnet.OpTimeCode,
ProtVerHi: 0,
ProtVerLo: 14,
Filler1: 0,
StreamId: 0,
Frames: 11,
Seconds: 17,
Minutes: 3,
Hours: 0,
Type: 0,
},
expected: &artnet.ArtTimeCode{
ID: []byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
OpCode: artnet.OpTimeCode,
ProtVerHi: 0,
ProtVerLo: 14,
Filler1: 0,
StreamId: 0,
Frames: 11,
Seconds: 17,
Minutes: 3,
Hours: 0,
Type: 0,
},
},
{
name: "timecode packet with mismatching opCode",
params: map[string]any{
"opCode": float64(artnet.OpDmx),
},
payload: &artnet.ArtTimeCode{
ID: []byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
OpCode: artnet.OpTimeCode,
ProtVerHi: 0,
ProtVerLo: 14,
Filler1: 0,
StreamId: 0,
Frames: 11,
Seconds: 17,
Minutes: 3,
Hours: 0,
Type: 0,
},
expected: nil,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["artnet.packet.filter"]
if !ok {
t.Fatalf("artnet.packet.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "artnet.packet.filter",
Params: test.params,
})
if err != nil {
t.Fatalf("artnet.packet.filter failed to create processor: %s", err)
}
got, err := processorInstance.Process(t.Context(), test.payload)
if err != nil {
t.Fatalf("artnet.packet.filter processing failed: %s", err)
}
if test.expected == nil {
if got != nil {
t.Fatalf("artnet.packet.filter got %+v, expected nil", got)
}
return
}
gotPacket, ok := got.(artnet.ArtNetPacket)
if !ok {
t.Fatalf("artnet.packet.filter returned a %T payload: %s", got, got)
}
if !reflect.DeepEqual(gotPacket, test.expected) {
t.Fatalf("artnet.packet.filter got %+v, expected %+v", gotPacket, test.expected)
}
})
}
}
func TestBadArtnetPacketFilter(t *testing.T) {
tests := []struct {
name string
params map[string]any
payload any
errorString string
}{
{
name: "non-artnet input",
payload: []byte{0x01},
params: map[string]any{"opCode": float64(artnet.OpTimeCode)},
errorString: "artnet.packet.filter processor only accepts an ArtNetPacket",
},
{
name: "no opCode param",
payload: &artnet.ArtTimeCode{
ID: []byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
OpCode: artnet.OpTimeCode,
ProtVerHi: 0,
ProtVerLo: 14,
Filler1: 0,
StreamId: 0,
Frames: 11,
Seconds: 17,
Minutes: 3,
Hours: 0,
Type: 0,
},
params: map[string]any{},
errorString: "artnet.packet.filter opCode error: not found",
},
{
name: "opCode not a number",
payload: &artnet.ArtTimeCode{
ID: []byte{'A', 'r', 't', '-', 'N', 'e', 't', 0x00},
OpCode: artnet.OpTimeCode,
ProtVerHi: 0,
ProtVerLo: 14,
Filler1: 0,
StreamId: 0,
Frames: 11,
Seconds: 17,
Minutes: 3,
Hours: 0,
Type: 0,
},
params: map[string]any{"opCode": "100"},
errorString: "artnet.packet.filter opCode error: not a number",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["artnet.packet.filter"]
if !ok {
t.Fatalf("artnet.packet.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "artnet.packet.filter",
Params: test.params,
})
if err != nil {
if test.errorString != err.Error() {
t.Fatalf("artnet.packet.filter got error '%s', expected '%s'", err.Error(), test.errorString)
}
return
}
got, err := processorInstance.Process(t.Context(), test.payload)
if err == nil {
t.Fatalf("artnet.packet.filter expected to fail but got payload: %s", got)
}
if err.Error() != test.errorString {
t.Fatalf("artnet.packet.filter got error '%s', expected '%s'", err.Error(), test.errorString)
}
})
}
}

View File

@@ -8,13 +8,13 @@ import (
)
func TestHTTPRequestCreateFromRegistry(t *testing.T) {
registration, ok := processor.ProcessorRegistry["http.request.create"]
registration, ok := processor.ProcessorRegistry["http.request.do"]
if !ok {
t.Fatalf("http.request.create processor not registered")
t.Fatalf("http.request.do processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "http.request.create",
Type: "http.request.do",
Params: map[string]any{
"method": "GET",
"url": "http://example.com",
@@ -22,10 +22,10 @@ func TestHTTPRequestCreateFromRegistry(t *testing.T) {
})
if err != nil {
t.Fatalf("failed to create http.request.create processor: %s", err)
t.Fatalf("failed to create http.request.do processor: %s", err)
}
if processorInstance.Type() != "http.request.create" {
t.Fatalf("http.request.create processor has wrong type: %s", processorInstance.Type())
if processorInstance.Type() != "http.request.do" {
t.Fatalf("http.request.do processor has wrong type: %s", processorInstance.Type())
}
}

View File

@@ -1,31 +0,0 @@
package processor_test
import (
"testing"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/processor"
)
func TestHTTPRequestFilterFromRegistry(t *testing.T) {
registration, ok := processor.ProcessorRegistry["http.request.filter"]
if !ok {
t.Fatalf("http.request.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "http.request.filter",
Params: map[string]any{
"method": "GET",
"path": "/test",
},
})
if err != nil {
t.Fatalf("failed to create http.request.filter processor: %s", err)
}
if processorInstance.Type() != "http.request.filter" {
t.Fatalf("http.request.filter processor has wrong type: %s", processorInstance.Type())
}
}

View File

@@ -99,9 +99,9 @@ func TestBadJsonDecode(t *testing.T) {
errorString string
}{
{
name: "non-string input",
payload: []byte("hello"),
errorString: "json.decode processor only accepts a string",
name: "non-string or byte input",
payload: 123,
errorString: "json.decode can only process a string or []byte",
},
{
name: "invalid json",

View File

@@ -1,157 +0,0 @@
package processor_test
import (
"reflect"
"testing"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/processor"
"gitlab.com/gomidi/midi/v2"
)
func TestMIDIMessageFilterFromRegistry(t *testing.T) {
registration, ok := processor.ProcessorRegistry["midi.message.filter"]
if !ok {
t.Fatalf("midi.message.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "midi.message.filter",
Params: map[string]any{
"type": "NoteOn",
},
})
if err != nil {
t.Fatalf("failed to create midi.message.filter processor: %s", err)
}
if processorInstance.Type() != "midi.message.filter" {
t.Fatalf("midi.message.filter processor has wrong type: %s", processorInstance.Type())
}
}
func TestGoodMIDIMessageFilter(t *testing.T) {
tests := []struct {
name string
params map[string]any
payload midi.Message
expected midi.Message
}{
{
name: "matches pattern",
payload: midi.NoteOn(1, 60, 127),
params: map[string]any{"type": "NoteOn"},
expected: midi.NoteOn(1, 60, 127),
},
{
name: "does not match pattern",
payload: midi.NoteOn(1, 60, 127),
params: map[string]any{"type": "NoteOff"},
expected: nil,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["midi.message.filter"]
if !ok {
t.Fatalf("midi.message.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "midi.message.filter",
Params: test.params,
})
if err != nil {
t.Fatalf("midi.message.filter failed to create processor: %s", err)
}
got, err := processorInstance.Process(t.Context(), test.payload)
if err != nil {
t.Fatalf("midi.message.filter processing failed: %s", err)
}
if test.expected == nil {
if got != nil {
t.Fatalf("midi.message.filter got %+v, expected nil", got)
}
return
}
gotMIDIMessage, ok := got.(midi.Message)
if !ok {
t.Fatalf("midi.message.filter returned a %T payload: %s", got, got)
}
if !reflect.DeepEqual(gotMIDIMessage, test.expected) {
t.Fatalf("midi.message.filter got %+v, expected %+v", gotMIDIMessage, test.expected)
}
})
}
}
func TestBadMIDIMessageFilter(t *testing.T) {
tests := []struct {
name string
params map[string]any
payload any
errorString string
}{
{
name: "no type param",
params: map[string]any{},
payload: midi.NoteOn(1, 60, 127),
errorString: "midi.message.filter type error: not found",
},
{
name: "non-string type param",
params: map[string]any{
"type": 123,
},
payload: "hello",
errorString: "midi.message.filter type error: not a string",
},
{
name: "non-midi message input",
params: map[string]any{
"type": "NoteOn",
},
payload: []byte{0x68, 0x65, 0x6c, 0x6c, 0x6f},
errorString: "midi.message.filter processor only accepts a midi.Message",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["midi.message.filter"]
if !ok {
t.Fatalf("midi.message.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "midi.message.filter",
Params: test.params,
})
if err != nil {
if test.errorString != err.Error() {
t.Fatalf("midi.message.filter got error '%s', expected '%s'", err.Error(), test.errorString)
}
return
}
got, err := processorInstance.Process(t.Context(), test.payload)
if err == nil {
t.Fatalf("midi.message.filter expected to fail but got payload: %s", got)
}
if err.Error() != test.errorString {
t.Fatalf("midi.message.filter got error '%s', expected '%s'", err.Error(), test.errorString)
}
})
}
}

View File

@@ -1,169 +0,0 @@
package processor_test
import (
"reflect"
"testing"
"github.com/jwetzell/osc-go"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/processor"
)
func TestOSCMessageFilterFromRegistry(t *testing.T) {
registration, ok := processor.ProcessorRegistry["osc.message.filter"]
if !ok {
t.Fatalf("osc.message.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "osc.message.filter",
Params: map[string]any{
"address": "/test*",
},
})
if err != nil {
t.Fatalf("failed to filter osc.message.filter processor: %s", err)
}
if processorInstance.Type() != "osc.message.filter" {
t.Fatalf("osc.message.filter processor has wrong type: %s", processorInstance.Type())
}
}
func TestGoodOSCMessageFilter(t *testing.T) {
tests := []struct {
name string
payload osc.OSCMessage
params map[string]any
expected any
}{
{
name: "basic address match",
params: map[string]any{
"address": "/test",
},
payload: osc.OSCMessage{Address: "/test"},
expected: osc.OSCMessage{Address: "/test"},
},
{
name: "basic address no match",
params: map[string]any{
"address": "/test",
},
payload: osc.OSCMessage{Address: "/testing"},
expected: nil,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["osc.message.filter"]
if !ok {
t.Fatalf("osc.message.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "osc.message.filter",
Params: test.params,
})
if err != nil {
t.Fatalf("osc.message.filter failed to create processor: %s", err)
}
got, err := processorInstance.Process(t.Context(), test.payload)
if err != nil {
t.Fatalf("osc.message.filter processing failed: %s", err)
}
if test.expected == nil {
if got != nil {
t.Fatalf("osc.message.filter got %+v, expected nil", got)
}
return
}
gotMessage, ok := got.(osc.OSCMessage)
if !ok {
t.Fatalf("osc.message.filter returned a %T payload: %s", got, got)
}
if !reflect.DeepEqual(gotMessage, test.expected) {
t.Fatalf("osc.message.filter got %+v, expected %+v", gotMessage, test.expected)
}
})
}
}
func TestBadOSCMessageFilter(t *testing.T) {
tests := []struct {
name string
params map[string]any
payload any
errorString string
}{
{
name: "no address parameter",
params: map[string]any{},
payload: osc.OSCMessage{Address: "/test"},
errorString: "osc.message.filter address error: not found",
},
{
name: "non-string address parameter",
params: map[string]any{
"address": 123,
},
payload: osc.OSCMessage{Address: "/test"},
errorString: "osc.message.filter address error: not a string",
},
{
name: "bad address pattern",
params: map[string]any{
"address": "[",
},
payload: osc.OSCMessage{Address: "/test"},
errorString: "error parsing regexp: missing closing ]: `[$`",
},
{
name: "non-osc input",
params: map[string]any{
"address": "/test",
},
payload: []byte("hello"),
errorString: "osc.message.filter can only operate on OSCMessage payloads",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["osc.message.filter"]
if !ok {
t.Fatalf("osc.message.filter processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "osc.message.filter",
Params: test.params,
})
if err != nil {
if test.errorString != err.Error() {
t.Fatalf("string.create got error '%s', expected '%s'", err.Error(), test.errorString)
}
return
}
got, err := processorInstance.Process(t.Context(), test.payload)
if err == nil {
t.Fatalf("osc.message.filter expected to fail but succeeded, got: %v", got)
}
if err.Error() != test.errorString {
t.Fatalf("osc.message.filter got error '%s', expected '%s'", err.Error(), test.errorString)
}
})
}
}

View File

@@ -139,7 +139,7 @@ func TestBadScriptWASM(t *testing.T) {
"enableWasi": true,
},
payload: "hello",
errorString: "script.wasm can only operator on byte array",
errorString: "script.wasm can only process a byte array",
},
{
name: "function not found in module",

View File

@@ -72,6 +72,12 @@ func TestGoodStructFieldGet(t *testing.T) {
payload: TestStruct{Bool: true},
expected: true,
},
{
name: "pointer to struct payload",
params: map[string]any{"name": "Data"},
payload: &TestStruct{Data: "hello"},
expected: "hello",
},
}
for _, test := range tests {

View File

@@ -89,6 +89,14 @@ func TestGoodStructMethodGet(t *testing.T) {
payload: TestStruct{},
expected: nil,
},
{
name: "pointer to struct payload",
params: map[string]any{
"name": "GetData",
},
payload: &TestStruct{Data: "hello"},
expected: "hello",
},
}
for _, test := range tests {

View File

@@ -8,18 +8,18 @@ import (
"github.com/jwetzell/showbridge-go/internal/config"
)
type MetaDelay struct {
type TimeSleep struct {
config config.ProcessorConfig
Duration time.Duration
}
func (md *MetaDelay) Process(ctx context.Context, payload any) (any, error) {
time.Sleep(md.Duration)
func (ts *TimeSleep) Process(ctx context.Context, payload any) (any, error) {
time.Sleep(ts.Duration)
return payload, nil
}
func (md *MetaDelay) Type() string {
return md.config.Type
func (ts *TimeSleep) Type() string {
return ts.config.Type
}
func init() {
@@ -33,7 +33,7 @@ func init() {
return nil, fmt.Errorf("time.sleep duration error: %w", err)
}
return &MetaDelay{config: config, Duration: time.Millisecond * time.Duration(durationNum)}, nil
return &TimeSleep{config: config, Duration: time.Millisecond * time.Duration(durationNum)}, nil
},
})
}

View File

@@ -17,23 +17,9 @@ type RouteError struct {
Config config.RouteConfig
Error error
}
type RouteIOError struct {
Index int
OutputError error
ProcessError error
InputError error
}
type RouteIO interface {
HandleInput(ctx context.Context, sourceId string, payload any) (bool, []RouteIOError)
HandleOutput(ctx context.Context, destinationId string, payload any) error
}
type Route struct {
input string
processors []processor.Processor
output string
}
func NewRoute(config config.RouteConfig) (*Route, error) {
@@ -54,17 +40,13 @@ func NewRoute(config config.RouteConfig) (*Route, error) {
}
}
return &Route{input: config.Input, processors: processors, output: config.Output}, nil
return &Route{input: config.Input, processors: processors}, nil
}
func (r *Route) Input() string {
return r.input
}
func (r *Route) Output() string {
return r.output
}
func (r *Route) ProcessPayload(ctx context.Context, payload any) (any, error) {
tracer := otel.Tracer("route")
processCtx, processSpan := tracer.Start(ctx, "ProcessPayload")

View File

@@ -14,7 +14,6 @@ import (
func TestRouteCreate(t *testing.T) {
routeConfig := config.RouteConfig{
Input: "input",
Output: "output",
}
testRoute, err := route.NewRoute(routeConfig)
@@ -25,15 +24,12 @@ func TestRouteCreate(t *testing.T) {
if testRoute.Input() != routeConfig.Input {
t.Fatalf("route input does not match expected input")
}
if testRoute.Output() != routeConfig.Output {
t.Fatalf("route output does not match expected output")
}
}
type MockRouter struct{}
func (mr *MockRouter) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []route.RouteIOError) {
return false, []route.RouteIOError{}
func (mr *MockRouter) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []common.RouteIOError) {
return false, []common.RouteIOError{}
}
func (mr *MockRouter) HandleOutput(ctx context.Context, destinationId string, payload any) error {
@@ -45,8 +41,13 @@ func TestGoodRouteHandleInput(t *testing.T) {
Input: "input",
Processors: []config.ProcessorConfig{
{Type: "string.encode"},
{
Type: "router.output",
Params: config.Params{
"module": "output",
},
},
},
Output: "output",
}
testRoute, err := route.NewRoute(routeConfig)
@@ -75,8 +76,13 @@ func TestRouteHandleInputWithProcessorError(t *testing.T) {
Input: "input",
Processors: []config.ProcessorConfig{
{Type: "string.create", Params: map[string]any{"template": "{{.invalid}}}"}},
{
Type: "router.output",
Params: config.Params{
"module": "output",
},
},
},
Output: "output",
}
testRoute, err := route.NewRoute(routeConfig)
@@ -94,8 +100,14 @@ func TestRouteHandleInputWithProcessorError(t *testing.T) {
func TestRouteHandleNilPayload(t *testing.T) {
routeConfig := config.RouteConfig{
Input: "input",
Processors: []config.ProcessorConfig{},
Output: "output",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Params: config.Params{
"module": "output",
},
},
},
}
testRoute, err := route.NewRoute(routeConfig)
@@ -118,8 +130,13 @@ func TestRouteHandleNilPayloadFromProcessor(t *testing.T) {
Input: "input",
Processors: []config.ProcessorConfig{
{Type: "script.js", Params: map[string]any{"program": "payload = undefined"}},
{
Type: "router.output",
Params: config.Params{
"module": "output",
},
},
},
Output: "output",
}
testRoute, err := route.NewRoute(routeConfig)
@@ -143,7 +160,6 @@ func TestRouteUnknownProcessor(t *testing.T) {
Processors: []config.ProcessorConfig{
{Type: "asdfasdflkjalkj"},
},
Output: "output",
}
_, err := route.NewRoute(routeConfig)
@@ -158,7 +174,6 @@ func TestRouteBadProcessorConfig(t *testing.T) {
Processors: []config.ProcessorConfig{
{Type: "string.create", Params: map[string]any{}},
},
Output: "output",
}
_, err := route.NewRoute(routeConfig)

View File

@@ -172,10 +172,10 @@ func (r *Router) Stop() {
r.contextCancel()
}
func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []route.RouteIOError) {
func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) (bool, []common.RouteIOError) {
spanCtx, span := otel.Tracer("router").Start(ctx, "input", trace.WithAttributes(attribute.String("source.id", sourceId)), trace.WithNewRoot())
defer span.End()
var routeIOErrors []route.RouteIOError
var routeIOErrors []common.RouteIOError
routeFound := false
var routeWaitGroup sync.WaitGroup
@@ -190,37 +190,22 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any)
routeFound = true
routeContext := context.WithValue(spanCtx, common.SourceContextKey, sourceId)
routeContext = context.WithValue(routeContext, common.RouterContextKey, r)
routeContext = context.WithValue(routeContext, common.ModulesContextKey, r.ModuleInstances)
routeCtx, routeSpan := otel.Tracer("router").Start(routeContext, "route", trace.WithAttributes(attribute.Int("route.index", routeIndex), attribute.String("route.input", routeInstance.Input()), attribute.String("route.output", routeInstance.Output())))
payload, err := routeInstance.ProcessPayload(routeCtx, payload)
routeCtx, routeSpan := otel.Tracer("router").Start(routeContext, "route", trace.WithAttributes(attribute.Int("route.index", routeIndex), attribute.String("route.input", routeInstance.Input())))
_, err := routeInstance.ProcessPayload(routeCtx, payload)
if err != nil {
if routeIOErrors == nil {
routeIOErrors = []route.RouteIOError{}
routeIOErrors = []common.RouteIOError{}
}
r.logger.Error("unable to process input", "route", routeIndex, "source", sourceId, "error", err)
routeIOErrors = append(routeIOErrors, route.RouteIOError{
routeIOErrors = append(routeIOErrors, common.RouteIOError{
Index: routeIndex,
ProcessError: err,
})
return
}
if payload == nil {
r.logger.Debug("no payload after processing, route terminated", "route", routeIndex, "source", sourceId)
return
}
outputError := r.HandleOutput(routeCtx, routeInstance.Output(), payload)
if outputError != nil {
if routeIOErrors == nil {
routeIOErrors = []route.RouteIOError{}
}
routeIOErrors = append(routeIOErrors, route.RouteIOError{
Index: routeIndex,
OutputError: outputError,
})
}
routeSpan.End()
})
}

View File

@@ -12,14 +12,13 @@ import (
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/module"
"github.com/jwetzell/showbridge-go/internal/route"
)
type MockCounterModule struct {
config config.ModuleConfig
ctx context.Context
outputCount int
router route.RouteIO
router common.RouteIO
logger *slog.Logger
cancel context.CancelFunc
}
@@ -34,7 +33,7 @@ func (mcm *MockCounterModule) Output(context.Context, any) error {
}
func (mcm *MockCounterModule) Start(ctx context.Context) error {
router, ok := ctx.Value(common.RouterContextKey).(route.RouteIO)
router, ok := ctx.Value(common.RouterContextKey).(common.RouteIO)
if !ok {
return fmt.Errorf("mock.counter could not get router from context")
@@ -171,7 +170,6 @@ func TestNewRouterRouteWithUnknwonProcessor(t *testing.T) {
Type: "asdfasdf",
},
},
Output: "mock",
},
},
}
@@ -202,7 +200,14 @@ func TestRouterInputUnknownDestinationModule(t *testing.T) {
Routes: []config.RouteConfig{
{
Input: "mock",
Output: "test",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Params: config.Params{
"module": "test",
},
},
},
},
},
}
@@ -238,8 +243,8 @@ func TestRouterInputUnknownDestinationModule(t *testing.T) {
t.Fatalf("router should have returned exactly 1 routing error, got: %d", len(routingErrors))
}
if routingErrors[0].OutputError.Error() != "no module found for destination id" {
t.Fatalf("routing output error did not match expected, got: %s", routingErrors[0].OutputError.Error())
if routingErrors[0].ProcessError.Error() != "router.output failed to send output: no module found for destination id" {
t.Fatalf("routing output error did not match expected, got: %s", routingErrors[0].ProcessError.Error())
}
}
@@ -254,7 +259,14 @@ func TestRouterInputNoMatchingRoute(t *testing.T) {
Routes: []config.RouteConfig{
{
Input: "test",
Output: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Params: config.Params{
"module": "mock",
},
},
},
},
},
}
@@ -298,7 +310,14 @@ func TestRouterInputSingleRoute(t *testing.T) {
Routes: []config.RouteConfig{
{
Input: "mock",
Output: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Params: config.Params{
"module": "mock",
},
},
},
},
},
}
@@ -362,15 +381,36 @@ func TestRouterInputMultipleRoutes(t *testing.T) {
Routes: []config.RouteConfig{
{
Input: "mock",
Output: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Params: config.Params{
"module": "mock",
},
},
},
},
{
Input: "mock",
Output: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Params: config.Params{
"module": "mock",
},
},
},
},
{
Input: "mock",
Output: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Params: config.Params{
"module": "mock",
},
},
},
},
},
}
@@ -437,11 +477,25 @@ func TestRouterInputMultipleModules(t *testing.T) {
Routes: []config.RouteConfig{
{
Input: "mock1",
Output: "mock1",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Params: config.Params{
"module": "mock1",
},
},
},
},
{
Input: "mock2",
Output: "mock2",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Params: config.Params{
"module": "mock2",
},
},
},
},
},
}

View File

@@ -2,7 +2,7 @@
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://showbridge.io/config.schema.json",
"title": "Config",
"description": "showbridge config file",
"description": "showbridge configuration",
"type": "object",
"properties": {
"modules": {

View File

@@ -2,28 +2,13 @@
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://showbridge.io/modules.schema.json",
"title": "Modules",
"description": "showbridge modules array",
"description": "module configurations",
"type": "array",
"items": {
"oneOf": [
{
"type": "object",
"title": "HTTPClientModule",
"properties": {
"id": {
"type": "string",
"minLength": 1
},
"type": {
"const": "http.client"
}
},
"required": ["id", "type"],
"additionalProperties": false
},
{
"type": "object",
"title": "HTTPServerModule",
"title": "HTTP Server",
"properties": {
"id": {
"type": "string",
@@ -36,6 +21,7 @@
"type": "object",
"properties": {
"port": {
"title": "Port",
"type": "integer",
"minimum": 1,
"maximum": 65535
@@ -50,7 +36,7 @@
},
{
"type": "object",
"title": "TimeIntervalModule",
"title": "Interval",
"properties": {
"id": {
"type": "string",
@@ -63,7 +49,9 @@
"type": "object",
"properties": {
"duration": {
"type": "integer"
"title": "Duration",
"type": "integer",
"description": "Interval duration in milliseconds"
}
},
"required": ["duration"],
@@ -75,7 +63,7 @@
},
{
"type": "object",
"title": "TimeTimerModule",
"title": "Timer",
"properties": {
"id": {
"type": "string",
@@ -88,7 +76,9 @@
"type": "object",
"properties": {
"duration": {
"type": "integer"
"title": "Duration",
"type": "integer",
"description": "Timer duration in milliseconds"
}
},
"required": ["duration"],
@@ -100,7 +90,7 @@
},
{
"type": "object",
"title": "MIDIInputModule",
"title": "MIDI Input",
"properties": {
"id": {
"type": "string",
@@ -113,6 +103,7 @@
"type": "object",
"properties": {
"port": {
"title": "Port",
"type": "string"
}
},
@@ -125,7 +116,7 @@
},
{
"type": "object",
"title": "MIDIOutputModule",
"title": "MIDI Output",
"properties": {
"id": {
"type": "string",
@@ -138,6 +129,7 @@
"type": "object",
"properties": {
"port": {
"title": "Port",
"type": "string"
}
},
@@ -150,7 +142,7 @@
},
{
"type": "object",
"title": "MQTTClientModule",
"title": "MQTT Client",
"properties": {
"id": {
"type": "string",
@@ -163,12 +155,15 @@
"type": "object",
"properties": {
"broker": {
"title": "Broker URL",
"type": "string"
},
"topic": {
"title": "Topic",
"type": "string"
},
"clientId": {
"title": "Client ID",
"type": "string"
}
},
@@ -181,7 +176,7 @@
},
{
"type": "object",
"title": "NATSClientModule",
"title": "NATS Client",
"properties": {
"id": {
"type": "string",
@@ -194,9 +189,11 @@
"type": "object",
"properties": {
"url": {
"title": "NATS Server URL",
"type": "string"
},
"subject": {
"title": "Subject",
"type": "string"
}
},
@@ -209,7 +206,7 @@
},
{
"type": "object",
"title": "NATSServerModule",
"title": "NATS Server",
"properties": {
"id": {
"type": "string",
@@ -222,10 +219,12 @@
"type": "object",
"properties": {
"ip": {
"title": "IP",
"type": "string",
"default": "0.0.0.0"
},
"port": {
"title": "Port",
"type": "integer",
"minimum": 1,
"maximum": 65535,
@@ -241,7 +240,7 @@
},
{
"type": "object",
"title": "PSNClientModule",
"title": "PSN Client",
"properties": {
"id": {
"type": "string",
@@ -256,7 +255,7 @@
},
{
"type": "object",
"title": "SerialClientModule",
"title": "Serial Client",
"properties": {
"id": {
"type": "string",
@@ -269,9 +268,11 @@
"type": "object",
"properties": {
"port": {
"title": "Port",
"type": "string"
},
"baudRate": {
"title": "Baud Rate",
"type": "integer"
}
},
@@ -284,7 +285,7 @@
},
{
"type": "object",
"title": "SIPCallServerModule",
"title": "SIP Call Server",
"properties": {
"id": {
"type": "string",
@@ -297,21 +298,25 @@
"type": "object",
"properties": {
"ip": {
"title": "IP",
"type": "string",
"default": "0.0.0.0"
},
"port": {
"title": "Port",
"type": "integer",
"minimum": 1,
"maximum": 65535,
"default": 5060
},
"transport": {
"title": "Transport",
"type": "string",
"enum": ["udp", "tcp", "ws", "udp4", "tcp4"],
"default": "udp"
},
"userAgent": {
"title": "User Agent",
"type": "string",
"default": "showbridge"
}
@@ -325,7 +330,7 @@
},
{
"type": "object",
"title": "SIPDTMFServerModule",
"title": "SIP DTMF Server",
"properties": {
"id": {
"type": "string",
@@ -338,25 +343,30 @@
"type": "object",
"properties": {
"ip": {
"title": "IP",
"type": "string",
"default": "0.0.0.0"
},
"port": {
"title": "Port",
"type": "integer",
"minimum": 1,
"maximum": 65535,
"default": 5060
},
"transport": {
"title": "Transport",
"type": "string",
"enum": ["udp", "tcp", "ws", "udp4", "tcp4"],
"default": "udp"
},
"userAgent": {
"title": "User Agent",
"type": "string",
"default": "showbridge"
},
"separator": {
"title": "DTMF Separator",
"type": "string",
"minLength": 1,
"maxLength": 1
@@ -371,7 +381,7 @@
},
{
"type": "object",
"title": "TCPClientModule",
"title": "TCP Client",
"properties": {
"id": {
"type": "string",
@@ -384,14 +394,17 @@
"type": "object",
"properties": {
"host": {
"title": "Host",
"type": "string"
},
"port": {
"title": "Port",
"type": "integer",
"minimum": 1,
"maximum": 65535
},
"framing": {
"title": "Framing Method",
"type": "string",
"enum": ["LF", "CR", "CRLF", "SLIP", "RAW"]
}
@@ -405,7 +418,7 @@
},
{
"type": "object",
"title": "TCPServerModule",
"title": "TCP Server",
"properties": {
"id": {
"type": "string",
@@ -418,15 +431,18 @@
"type": "object",
"properties": {
"ip": {
"title": "IP",
"type": "string",
"default": "0.0.0.0"
},
"port": {
"title" : "Port",
"type": "integer",
"minimum": 1,
"maximum": 65535
},
"framing": {
"title": "Framing Method",
"type": "string",
"enum": ["LF", "CR", "CRLF", "SLIP", "RAW"]
}
@@ -440,7 +456,7 @@
},
{
"type": "object",
"title": "UDPClientModule",
"title": "UDP Client",
"properties": {
"id": {
"type": "string",
@@ -453,9 +469,11 @@
"type": "object",
"properties": {
"host": {
"title": "Host",
"type": "string"
},
"port": {
"title": "Port",
"type": "integer",
"minimum": 1,
"maximum": 65535
@@ -470,7 +488,7 @@
},
{
"type": "object",
"title": "UDPMulticastModule",
"title": "UDP Multicast",
"properties": {
"id": {
"type": "string",
@@ -483,9 +501,11 @@
"type": "object",
"properties": {
"ip": {
"title": "IP",
"type": "string"
},
"port": {
"title": "Port",
"type": "integer",
"minimum": 1,
"maximum": 65535
@@ -500,7 +520,7 @@
},
{
"type": "object",
"title": "UDPServerModule",
"title": "UDP Server",
"properties": {
"id": {
"type": "string",
@@ -513,10 +533,12 @@
"type": "object",
"properties": {
"ip": {
"title": "IP",
"type": "string",
"default": "0.0.0.0"
},
"port": {
"title": "Port",
"type": "integer",
"minimum": 1,
"maximum": 65535

View File

@@ -2,7 +2,7 @@
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://showbridge.io/processors.schema.json",
"title": "Processors",
"description": "showbridge processors array",
"description": "processor configurations",
"type": "array",
"items": {
"oneOf": [
@@ -30,28 +30,6 @@
"required": ["type"],
"additionalProperties": false
},
{
"type": "object",
"title": "Filter ArtNet Packet",
"properties": {
"type": {
"type": "string",
"const": "artnet.packet.filter"
},
"params": {
"type": "object",
"properties": {
"opCode": {
"type": "integer"
}
},
"required": ["opCode"],
"additionalProperties": false
}
},
"required": ["type", "params"],
"additionalProperties": false
},
{
"type": "object",
"title": "Debug Log",
@@ -76,6 +54,7 @@
"type": "object",
"properties": {
"expression": {
"title": "Expression",
"type": "string"
}
},
@@ -93,6 +72,18 @@
"type": {
"type": "string",
"const": "float.parse"
},
"params": {
"type": "object",
"properties": {
"bitSize": {
"title": "Bit Size",
"type": "integer",
"enum": [32, 64],
"default": 64
}
},
"additionalProperties": false
}
},
"required": ["type"],
@@ -110,30 +101,39 @@
"type": "object",
"properties": {
"id": {
"title": "Camera ID",
"type": "string"
},
"pan": {
"title": "Pan",
"type": "string"
},
"tilt": {
"title": "Tilt",
"type": "string"
},
"roll": {
"title": "Roll",
"type": "string"
},
"posX": {
"title": "Position X",
"type": "string"
},
"posY": {
"title": "Position Y",
"type": "string"
},
"posZ": {
"title": "Position Z",
"type": "string"
},
"zoom": {
"title": "Zoom",
"type": "string"
},
"focus": {
"title": "Focus",
"type": "string"
}
},
@@ -180,20 +180,22 @@
},
{
"type": "object",
"title": "Create HTTP Request",
"title": "Do HTTP Request",
"properties": {
"type": {
"type": "string",
"const": "http.request.create"
"const": "http.request.do"
},
"params": {
"type": "object",
"properties": {
"method": {
"title": "HTTP Method",
"type": "string",
"enum": ["GET", "POST"]
},
"url": {
"title": "URL",
"type": "string"
}
},
@@ -204,32 +206,6 @@
"required": ["type", "params"],
"additionalProperties": false
},
{
"type": "object",
"title": "Filter HTTP Request",
"properties": {
"type": {
"type": "string",
"const": "http.request.filter"
},
"params": {
"type": "object",
"properties": {
"path": {
"type": "string"
},
"method": {
"type": "string",
"enum": ["GET", "POST"]
}
},
"required": ["path"],
"additionalProperties": false
}
},
"required": ["type", "params"],
"additionalProperties": false
},
{
"type": "object",
"title": "Create HTTP Response",
@@ -242,9 +218,11 @@
"type": "object",
"properties": {
"status": {
"title": "Status Code",
"type": "integer"
},
"body": {
"title": "Body",
"type": "string"
}
},
@@ -262,6 +240,24 @@
"type": {
"type": "string",
"const": "int.parse"
},
"params": {
"type": "object",
"properties": {
"base": {
"title": "Base",
"type": "integer",
"enum": [0,2, 8, 10, 16],
"default": 10
},
"bitSize": {
"title": "Bit Size",
"type": "integer",
"enum": [0, 8, 16, 32, 64],
"default": 64
}
},
"additionalProperties": false
}
},
"required": ["type"],
@@ -279,9 +275,11 @@
"type": "object",
"properties": {
"min": {
"title": "Minimum",
"type": "integer"
},
"max": {
"title": "Maximum",
"type": "integer"
}
},
@@ -331,16 +329,20 @@
"type": "object",
"properties": {
"type": {
"title": "MIDI Message Type",
"type": "string",
"enum": ["NoteOn", "noteon", "note_on"]
},
"channel": {
"title": "Channel",
"type": "string"
},
"note": {
"title": "Note",
"type": "string"
},
"velocity": {
"title": "Velocity",
"type": "string"
}
},
@@ -351,16 +353,20 @@
"type": "object",
"properties": {
"type": {
"title": "MIDI Message Type",
"type": "string",
"enum": ["NoteOff", "noteoff", "note_off"]
},
"channel": {
"title": "Channel",
"type": "string"
},
"note": {
"title": "Note",
"type": "string"
},
"velocity": {
"title": "Velocity",
"type": "string"
}
},
@@ -371,16 +377,20 @@
"type": "object",
"properties": {
"type": {
"title": "MIDI Message Type",
"type": "string",
"enum": ["ControlChange", "controlchange", "control_change"]
},
"channel": {
"title": "Channel",
"type": "string"
},
"control": {
"title": "Control",
"type": "string"
},
"value": {
"title": "Value",
"type": "string"
}
},
@@ -391,13 +401,16 @@
"type": "object",
"properties": {
"type": {
"title": "MIDI Message Type",
"type": "string",
"enum": ["ProgramChange", "programchange", "program_change"]
},
"channel": {
"title": "Channel",
"type": "string"
},
"program": {
"title": "Program",
"type": "string"
}
},
@@ -434,29 +447,6 @@
"required": ["type"],
"additionalProperties": false
},
{
"type": "object",
"title": "Filter MIDI Message",
"properties": {
"type": {
"type": "string",
"const": "midi.message.filter"
},
"params": {
"type": "object",
"properties": {
"type": {
"type": "string",
"enum": ["NoteOn", "NoteOff", "ControlChange", "ProgramChange"]
}
},
"required": ["type"],
"additionalProperties": false
}
},
"required": ["type", "params"],
"additionalProperties": false
},
{
"type": "object",
"title": "Unpack MIDI Message",
@@ -481,15 +471,19 @@
"type": "object",
"properties": {
"topic": {
"title": "Topic",
"type": "string"
},
"qos": {
"title": "QoS",
"type": "number"
},
"retained": {
"title": "Retained",
"type": "boolean"
},
"payload": {
"title": "Payload",
"type": "string"
}
},
@@ -524,9 +518,11 @@
"type": "object",
"properties": {
"subject": {
"title": "Subject",
"type": "string"
},
"payload": {
"title": "Payload",
"type": "string"
}
},
@@ -549,15 +545,18 @@
"type": "object",
"properties": {
"address": {
"title": "Address",
"type": "string"
},
"args": {
"title": "Arguments",
"type": "array",
"items": {
"type": "string"
}
},
"types": {
"title": "Argument Types",
"type": "string"
}
},
@@ -594,21 +593,22 @@
},
{
"type": "object",
"title": "Filter OSC Message",
"title": "Router Output",
"properties": {
"type": {
"type": "string",
"const": "osc.message.filter"
"const": "router.output"
},
"params": {
"type": "object",
"properties": {
"address": {
"type": "string"
"module": {
"title": "Module ID",
"type": "string",
"description": "ID of module to send output to"
}
},
"required": ["address"],
"additionalProperties": false
"required": ["module"]
}
},
"required": ["type", "params"],
@@ -626,6 +626,7 @@
"type": "object",
"properties": {
"expression": {
"title": "Expression",
"type": "string"
}
},
@@ -648,6 +649,7 @@
"type": "object",
"properties": {
"program": {
"title": "Program",
"type": "string"
}
},
@@ -670,13 +672,16 @@
"type": "object",
"properties": {
"path": {
"title": "Path",
"type": "string"
},
"function": {
"title": "Function",
"type": "string",
"default": "process"
},
"enableWasi": {
"title": "Enable WASI",
"type": "boolean",
"default": false
}
@@ -700,6 +705,7 @@
"type": "object",
"properties": {
"template": {
"title": "Template",
"type": "string"
}
},
@@ -745,6 +751,7 @@
"type": "object",
"properties": {
"pattern": {
"title": "Pattern",
"type": "string"
}
},
@@ -767,6 +774,7 @@
"type": "object",
"properties": {
"separator": {
"title": "Separator",
"type": "string"
}
},
@@ -788,6 +796,7 @@
"type": "object",
"properties": {
"name": {
"title": "Field Name",
"type": "string"
}
},
@@ -809,6 +818,7 @@
"type": "object",
"properties": {
"name": {
"title": "Method Name",
"type": "string"
}
},
@@ -830,7 +840,9 @@
"type": "object",
"properties": {
"duration": {
"type": "integer"
"title": "Duration",
"type": "integer",
"description": "Duration to sleep in milliseconds"
}
},
"required": ["duration"]

View File

@@ -2,7 +2,7 @@
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://showbridge.io/routes.schema.json",
"title": "Routes",
"description": "showbridge routes array",
"description": "route configurations",
"type": "array",
"items": {
"type": "object",
@@ -13,12 +13,8 @@
},
"processors": {
"$ref": "https://showbridge.io/processors.schema.json"
},
"output": {
"type": "string",
"minLength": 1
}
},
"required": ["input", "output"]
"required": ["input"]
}
}