diff --git a/internal/processor/router-output.go b/internal/processor/module-output.go similarity index 52% rename from internal/processor/router-output.go rename to internal/processor/module-output.go index 3815d07..bb6414a 100644 --- a/internal/processor/router-output.go +++ b/internal/processor/module-output.go @@ -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 }, }) } diff --git a/internal/processor/test/module-output_test.go b/internal/processor/test/module-output_test.go new file mode 100644 index 0000000..e7f7b5f --- /dev/null +++ b/internal/processor/test/module-output_test.go @@ -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) + } + }) + } +} diff --git a/internal/processor/test/router-input_test.go b/internal/processor/test/router-input_test.go index fc6d8a3..b3859bf 100644 --- a/internal/processor/test/router-input_test.go +++ b/internal/processor/test/router-input_test.go @@ -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) } }) } diff --git a/internal/processor/test/router-output_test.go b/internal/processor/test/router-output_test.go deleted file mode 100644 index b3859bf..0000000 --- a/internal/processor/test/router-output_test.go +++ /dev/null @@ -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) - } - }) - } -} diff --git a/internal/route/route_test.go b/internal/route/route_test.go index 3bd6640..738131a 100644 --- a/internal/route/route_test.go +++ b/internal/route/route_test.go @@ -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 { diff --git a/internal/test/module.go b/internal/test/module.go index 1aee83b..27e3ac6 100644 --- a/internal/test/module.go +++ b/internal/test/module.go @@ -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, diff --git a/router_test.go b/router_test.go index a1cca98..982e493 100644 --- a/router_test.go +++ b/router_test.go @@ -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", },