jwetzell/showbridge-go
1
0
Fork 0
mirror of https://github.com/jwetzell/showbridge-go.git synced 2026-07-26 09:58:40 +00:00
Code Issues Packages Projects Releases 10 Wiki Activity

rename router.output to module.output

Browse Source
This commit is contained in:
Joel Wetzell
2026-05-19 21:31:14 -05:00
parent b2e07d7452
commit 99132c90c6
7 changed files with 264 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/processor/router-output.go → internal/processor/module-output.go
+28 -12
View File
@@ -11,37 +11,53 @@ import (
"github.com/jwetzell/showbridge-go/internal/config"
)
type RouterOutput struct {
type ModuleOutput struct {
config config.ProcessorConfig
ModuleId string
logger *slog.Logger
module common.OutputModule
}
func (ro *RouterOutput) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
func (ro *ModuleOutput) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) {
if wrappedPayload.Router == nil {
wrappedPayload.End = true
return wrappedPayload, errors.New("router.output no router found")
if ro.module == nil {
if wrappedPayload.Modules == nil {
wrappedPayload.End = true
return wrappedPayload, errors.New("module.output wrapped payload has no modules")
}
module, ok := wrappedPayload.Modules[ro.ModuleId]
if !ok {
wrappedPayload.End = true
return wrappedPayload, fmt.Errorf("module.output unable to find module with id: %s", ro.ModuleId)
}
outputModule, ok := module.(common.OutputModule)
if !ok {
wrappedPayload.End = true
return wrappedPayload, fmt.Errorf("module.output module with id %s is not an OutputModule", ro.ModuleId)
}
ro.module = outputModule
}
err := wrappedPayload.Router.HandleOutput(ctx, ro.ModuleId, wrappedPayload.Payload)
err := ro.module.Output(ctx, wrappedPayload.Payload)
if err != nil {
wrappedPayload.End = true
return wrappedPayload, fmt.Errorf("router.output failed to send output: %w", err)
return wrappedPayload, fmt.Errorf("module.output failed to send output: %w", err)
}
return wrappedPayload, nil
}
func (ro *RouterOutput) Type() string {
func (ro *ModuleOutput) Type() string {
return ro.config.Type
}
func init() {
RegisterProcessor(ProcessorRegistration{
Type: "router.output",
Title: "Router Output",
Type: "module.output",
Title: "Module Output",
ParamsSchema: &jsonschema.Schema{
Type: "object",
Properties: map[string]*jsonschema.Schema{
@@ -61,10 +77,10 @@ func init() {
moduleId, err := params.GetString("module")
if err != nil {
return nil, fmt.Errorf("router.output module error: %w", err)
return nil, fmt.Errorf("module.output module error: %w", err)
}
return &RouterOutput{config: config, ModuleId: moduleId, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
return &ModuleOutput{config: config, ModuleId: moduleId, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
},
})
}
internal/processor/test/module-output_test.go
+155
View File
@@ -0,0 +1,155 @@
package processor_test
import (
"reflect"
"testing"
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/processor"
"github.com/jwetzell/showbridge-go/internal/test"
)
func TestModuleOutputFromRegistry(t *testing.T) {
registration, ok := processor.ProcessorRegistry["module.output"]
if !ok {
t.Fatalf("module.output processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "module.output",
Params: config.Params{
"module": "test",
},
})
if err != nil {
t.Fatalf("failed to create module.output processor: %s", err)
}
if processorInstance.Type() != "module.output" {
t.Fatalf("module.output processor has wrong type: %s", processorInstance.Type())
}
payload := "test"
expected := "test"
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{
Router: test.GetNewTestRouter(),
Modules: map[string]common.Module{"test": &test.TestOutputModule{}},
Payload: payload,
})
if err != nil {
t.Fatalf("module.output processing failed: %s", err)
}
if got.Payload != expected {
t.Fatalf("module.output got %+v, expected %+v", got, expected)
}
}
func TestGoodModuleOutput(t *testing.T) {
testCases := []struct {
name string
params map[string]any
payload any
expected any
}{}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["module.output"]
if !ok {
t.Fatalf("module.output processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "module.output",
Params: testCase.params,
})
if err != nil {
t.Fatalf("module.output failed to create processor: %s", err)
}
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Payload: testCase.payload})
if err != nil {
t.Fatalf("module.output processing failed: %s", err)
}
if !reflect.DeepEqual(got.Payload, testCase.expected) {
t.Fatalf("module.output got %+v (%T), expected %+v (%T)", got.Payload, got.Payload, testCase.expected, testCase.expected)
}
})
}
}
func TestBadModuleOutput(t *testing.T) {
testCases := []struct {
name string
params map[string]any
payload any
modules map[string]common.Module
errorString string
}{
{
name: "no module param",
params: map[string]any{},
payload: "test",
modules: map[string]common.Module{"test": &test.TestModule{}},
errorString: "module.output module error: not found",
},
{
name: "non-string module",
params: map[string]any{
"module": 123,
},
payload: "test",
modules: map[string]common.Module{"test": &test.TestModule{}},
errorString: "module.output module error: not a string",
},
{
name: "modules not found in context",
params: map[string]any{
"module": "test",
},
payload: "test",
modules: nil,
errorString: "module.output wrapped payload has no modules",
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["module.output"]
if !ok {
t.Fatalf("module.output processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "module.output",
Params: testCase.params,
})
if err != nil {
if testCase.errorString != err.Error() {
t.Fatalf("module.output got error '%s', expected '%s'", err.Error(), testCase.errorString)
}
return
}
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Modules: testCase.modules, Payload: testCase.payload})
if err == nil {
t.Fatalf("module.output expected to fail but succeeded, got: %v", got)
}
if err.Error() != testCase.errorString {
t.Fatalf("module.output got error '%s', expected '%s'", err.Error(), testCase.errorString)
}
})
}
}
internal/processor/test/router-input_test.go
+33 -34
View File
@@ -10,25 +10,25 @@ import (
"github.com/jwetzell/showbridge-go/internal/test"
)
func TestRouterOutputFromRegistry(t *testing.T) {
registration, ok := processor.ProcessorRegistry["router.output"]
func TestRouterInputFromRegistry(t *testing.T) {
registration, ok := processor.ProcessorRegistry["router.input"]
if !ok {
t.Fatalf("router.output processor not registered")
t.Fatalf("router.input processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "router.output",
Type: "router.input",
Params: config.Params{
"module": "test",
"source": "test",
},
})
if err != nil {
t.Fatalf("failed to create router.output processor: %s", err)
t.Fatalf("failed to create router.input processor: %s", err)
}
if processorInstance.Type() != "router.output" {
t.Fatalf("router.output processor has wrong type: %s", processorInstance.Type())
if processorInstance.Type() != "router.input" {
t.Fatalf("router.input processor has wrong type: %s", processorInstance.Type())
}
payload := "test"
@@ -39,15 +39,15 @@ func TestRouterOutputFromRegistry(t *testing.T) {
Payload: payload,
})
if err != nil {
t.Fatalf("router.output processing failed: %s", err)
t.Fatalf("router.input processing failed: %s", err)
}
if got.Payload != expected {
t.Fatalf("router.output got %+v, expected %+v", got, expected)
t.Fatalf("router.input got %+v, expected %+v", got, expected)
}
}
func TestGoodRouterOutput(t *testing.T) {
func TestGoodRouterInput(t *testing.T) {
testCases := []struct {
name string
@@ -59,33 +59,33 @@ func TestGoodRouterOutput(t *testing.T) {
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["router.output"]
registration, ok := processor.ProcessorRegistry["router.input"]
if !ok {
t.Fatalf("router.output processor not registered")
t.Fatalf("router.input processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "router.output",
Type: "router.input",
Params: testCase.params,
})
if err != nil {
t.Fatalf("router.output failed to create processor: %s", err)
t.Fatalf("router.input failed to create processor: %s", err)
}
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Payload: testCase.payload})
if err != nil {
t.Fatalf("router.output processing failed: %s", err)
t.Fatalf("router.input processing failed: %s", err)
}
if !reflect.DeepEqual(got.Payload, testCase.expected) {
t.Fatalf("router.output got %+v (%T), expected %+v (%T)", got.Payload, got.Payload, testCase.expected, testCase.expected)
t.Fatalf("router.input got %+v (%T), expected %+v (%T)", got.Payload, got.Payload, testCase.expected, testCase.expected)
}
})
}
}
func TestBadRouterOutput(t *testing.T) {
func TestBadRouterInput(t *testing.T) {
testCases := []struct {
name string
params map[string]any
@@ -94,49 +94,48 @@ func TestBadRouterOutput(t *testing.T) {
errorString string
}{
{
name: "no module param",
name: "no source param",
params: map[string]any{},
payload: "test",
router: test.GetNewTestRouter(),
errorString: "router.output module error: not found",
errorString: "router.input source error: not found",
},
{
name: "non-string module",
name: "non-string source",
params: map[string]any{
"module": 123,
"source": 123,
},
payload: "test",
router: test.GetNewTestRouter(),
errorString: "router.output module error: not a string",
payload: "test",
router: test.GetNewTestRouter(),
errorString: "router.input source error: not a string",
},
{
name: "router not found in context",
params: map[string]any{
"module": "test",
"source": "test",
},
payload: "test",
router: nil,
errorString: "router.output no router found",
errorString: "router.input no router found",
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["router.output"]
registration, ok := processor.ProcessorRegistry["router.input"]
if !ok {
t.Fatalf("router.output processor not registered")
t.Fatalf("router.input processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "router.output",
Type: "router.input",
Params: testCase.params,
})
if err != nil {
if testCase.errorString != err.Error() {
t.Fatalf("router.output got error '%s', expected '%s'", err.Error(), testCase.errorString)
t.Fatalf("router.input got error '%s', expected '%s'", err.Error(), testCase.errorString)
}
return
}
@@ -144,11 +143,11 @@ func TestBadRouterOutput(t *testing.T) {
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Router: testCase.router, Payload: testCase.payload})
if err == nil {
t.Fatalf("router.output expected to fail but succeeded, got: %v", got)
t.Fatalf("router.input expected to fail but succeeded, got: %v", got)
}
if err.Error() != testCase.errorString {
t.Fatalf("router.output got error '%s', expected '%s'", err.Error(), testCase.errorString)
t.Fatalf("router.input got error '%s', expected '%s'", err.Error(), testCase.errorString)
}
})
}
internal/processor/test/router-output_test.go
-154
View File
@@ -1,154 +0,0 @@
package processor_test
import (
"reflect"
"testing"
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/processor"
"github.com/jwetzell/showbridge-go/internal/test"
)
func TestRouterInputFromRegistry(t *testing.T) {
registration, ok := processor.ProcessorRegistry["router.input"]
if !ok {
t.Fatalf("router.input processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "router.input",
Params: config.Params{
"source": "test",
},
})
if err != nil {
t.Fatalf("failed to create router.input processor: %s", err)
}
if processorInstance.Type() != "router.input" {
t.Fatalf("router.input processor has wrong type: %s", processorInstance.Type())
}
payload := "test"
expected := "test"
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{
Router: test.GetNewTestRouter(),
Payload: payload,
})
if err != nil {
t.Fatalf("router.input processing failed: %s", err)
}
if got.Payload != expected {
t.Fatalf("router.input got %+v, expected %+v", got, expected)
}
}
func TestGoodRouterInput(t *testing.T) {
testCases := []struct {
name string
params map[string]any
payload any
expected any
}{}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["router.input"]
if !ok {
t.Fatalf("router.input processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "router.input",
Params: testCase.params,
})
if err != nil {
t.Fatalf("router.input failed to create processor: %s", err)
}
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Payload: testCase.payload})
if err != nil {
t.Fatalf("router.input processing failed: %s", err)
}
if !reflect.DeepEqual(got.Payload, testCase.expected) {
t.Fatalf("router.input got %+v (%T), expected %+v (%T)", got.Payload, got.Payload, testCase.expected, testCase.expected)
}
})
}
}
func TestBadRouterInput(t *testing.T) {
testCases := []struct {
name string
params map[string]any
payload any
router common.RouteIO
errorString string
}{
{
name: "no source param",
params: map[string]any{},
payload: "test",
router: test.GetNewTestRouter(),
errorString: "router.input source error: not found",
},
{
name: "non-string source",
params: map[string]any{
"source": 123,
},
payload: "test",
router: test.GetNewTestRouter(),
errorString: "router.input source error: not a string",
},
{
name: "router not found in context",
params: map[string]any{
"source": "test",
},
payload: "test",
router: nil,
errorString: "router.input no router found",
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
registration, ok := processor.ProcessorRegistry["router.input"]
if !ok {
t.Fatalf("router.input processor not registered")
}
processorInstance, err := registration.New(config.ProcessorConfig{
Type: "router.input",
Params: testCase.params,
})
if err != nil {
if testCase.errorString != err.Error() {
t.Fatalf("router.input got error '%s', expected '%s'", err.Error(), testCase.errorString)
}
return
}
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Router: testCase.router, Payload: testCase.payload})
if err == nil {
t.Fatalf("router.input expected to fail but succeeded, got: %v", got)
}
if err.Error() != testCase.errorString {
t.Fatalf("router.input got error '%s', expected '%s'", err.Error(), testCase.errorString)
}
})
}
}
internal/route/route_test.go
+9 -4
View File
@@ -8,6 +8,7 @@ import (
"github.com/jwetzell/showbridge-go/internal/common"
"github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/route"
"github.com/jwetzell/showbridge-go/internal/test"
)
func TestRouteCreate(t *testing.T) {
@@ -41,7 +42,7 @@ func TestGoodRouteHandleInput(t *testing.T) {
Processors: []config.ProcessorConfig{
{Type: "string.encode"},
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "output",
},
@@ -57,6 +58,7 @@ func TestGoodRouteHandleInput(t *testing.T) {
inputData := "test input data"
payload, err := testRoute.ProcessPayload(t.Context(), common.WrappedPayload{
Router: &MockRouter{},
Modules: map[string]common.Module{"output": &test.TestOutputModule{}},
Payload: inputData,
})
if err != nil {
@@ -79,7 +81,7 @@ func TestRouteHandleInputWithProcessorError(t *testing.T) {
Processors: []config.ProcessorConfig{
{Type: "string.create", Params: map[string]any{"template": "{{.invalid}}}"}},
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "output",
},
@@ -95,6 +97,7 @@ func TestRouteHandleInputWithProcessorError(t *testing.T) {
inputData := "test input data"
_, err = testRoute.ProcessPayload(t.Context(), common.WrappedPayload{
Router: &MockRouter{},
Modules: map[string]common.Module{"output": &test.TestOutputModule{}},
Payload: inputData,
})
if err == nil {
@@ -107,7 +110,7 @@ func TestRouteHandleNilPayload(t *testing.T) {
Input: "input",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "output",
},
@@ -123,6 +126,7 @@ func TestRouteHandleNilPayload(t *testing.T) {
payload, err := testRoute.ProcessPayload(t.Context(), common.WrappedPayload{
Router: &MockRouter{},
Modules: map[string]common.Module{"output": &test.TestOutputModule{}},
Payload: nil,
})
if err != nil {
@@ -139,7 +143,7 @@ func TestRouteHandleNilPayloadFromProcessor(t *testing.T) {
Processors: []config.ProcessorConfig{
{Type: "script.js", Params: map[string]any{"program": "payload = undefined"}},
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "output",
},
@@ -154,6 +158,7 @@ func TestRouteHandleNilPayloadFromProcessor(t *testing.T) {
_, err = testRoute.ProcessPayload(t.Context(), common.WrappedPayload{
Router: &MockRouter{},
Modules: map[string]common.Module{"output": &test.TestOutputModule{}},
Payload: "test",
})
if err != nil {
internal/test/module.go
+30
View File
@@ -26,6 +26,35 @@ func (m *TestModule) Id() string {
return "test"
}
func NewTestOutputModule(id string) *TestOutputModule {
return &TestOutputModule{
id: id,
}
}
type TestOutputModule struct {
id string
}
func (m *TestOutputModule) Start(ctx context.Context, router common.RouteIO) error {
<-ctx.Done()
return nil
}
func (m *TestOutputModule) Output(ctx context.Context, payload any) error {
return nil
}
func (m *TestOutputModule) Stop() {}
func (m *TestOutputModule) Type() string {
return "test.output"
}
func (m *TestOutputModule) Id() string {
return m.id
}
func NewTestKVModule(id string) *TestKVModule {
return &TestKVModule{
id: id,
@@ -63,6 +92,7 @@ func (m *TestKVModule) Set(key string, value any) error {
m.kvData[key] = value
return nil
}
func NewTestDBModule(id string) *TestDBModule {
return &TestDBModule{
id: id,
router_test.go
+9 -9
View File
@@ -203,7 +203,7 @@ func TestRouterInputUnknownDestinationModule(t *testing.T) {
Input: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "test",
},
@@ -239,7 +239,7 @@ func TestRouterInputUnknownDestinationModule(t *testing.T) {
t.Fatalf("router should have returned exactly 1 routing error, got: %d", len(routingErrors))
}
if routingErrors[0].ProcessError.Error() != "router.output failed to send output: no module found for destination id" {
if routingErrors[0].ProcessError.Error() != "module.output unable to find module with id: test" {
t.Fatalf("routing output error did not match expected, got: %s", routingErrors[0].ProcessError.Error())
}
}
@@ -257,7 +257,7 @@ func TestRouterInputNoMatchingRoute(t *testing.T) {
Input: "test",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "mock",
},
@@ -303,7 +303,7 @@ func TestRouterInputSingleRoute(t *testing.T) {
Input: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "mock",
},
@@ -369,7 +369,7 @@ func TestRouterInputMultipleRoutes(t *testing.T) {
Input: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "mock",
},
@@ -380,7 +380,7 @@ func TestRouterInputMultipleRoutes(t *testing.T) {
Input: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "mock",
},
@@ -391,7 +391,7 @@ func TestRouterInputMultipleRoutes(t *testing.T) {
Input: "mock",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "mock",
},
@@ -461,7 +461,7 @@ func TestRouterInputMultipleModules(t *testing.T) {
Input: "mock1",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "mock1",
},
@@ -472,7 +472,7 @@ func TestRouterInputMultipleModules(t *testing.T) {
Input: "mock2",
Processors: []config.ProcessorConfig{
{
Type: "router.output",
Type: "module.output",
Params: config.Params{
"module": "mock2",
},