diff --git a/internal/common/routing.go b/internal/common/routing.go index 8e6d317..d470cfa 100644 --- a/internal/common/routing.go +++ b/internal/common/routing.go @@ -6,12 +6,9 @@ import ( 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 `json:"index"` - OutputError error `json:"outputError"` ProcessError error `json:"processError"` - InputError error `json:"inputError"` } diff --git a/internal/module/http-server.go b/internal/module/http-server.go index 5f23861..ef5bedb 100644 --- a/internal/module/http-server.go +++ b/internal/module/http-server.go @@ -29,9 +29,7 @@ type HTTPServer struct { type ResponseIOError struct { Index int `json:"index"` - OutputError *string `json:"outputError"` ProcessError *string `json:"processError"` - InputError *string `json:"inputError"` } type IOResponseData struct { @@ -116,21 +114,11 @@ func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { Index: responseIOError.Index, } - if responseIOError.InputError != nil { - errorMsg := responseIOError.InputError.Error() - errorToAdd.InputError = &errorMsg - } - if responseIOError.ProcessError != nil { errorMsg := responseIOError.ProcessError.Error() errorToAdd.ProcessError = &errorMsg } - if responseIOError.OutputError != nil { - errorMsg := responseIOError.OutputError.Error() - errorToAdd.OutputError = &errorMsg - } - response.IOErrors = append(response.IOErrors, errorToAdd) } diff --git a/internal/route/route_test.go b/internal/route/route_test.go index 738131a..9075126 100644 --- a/internal/route/route_test.go +++ b/internal/route/route_test.go @@ -32,10 +32,6 @@ func (mr *MockRouter) HandleInput(ctx context.Context, sourceId string, payload return false, []common.RouteIOError{} } -func (mr *MockRouter) HandleOutput(ctx context.Context, destinationId string, payload any) error { - return nil -} - func TestGoodRouteHandleInput(t *testing.T) { routeConfig := config.RouteConfig{ Input: "input", @@ -101,7 +97,7 @@ func TestRouteHandleInputWithProcessorError(t *testing.T) { Payload: inputData, }) if err == nil { - t.Fatalf("route HandleOutput did not return error for bad processor") + t.Fatalf("route did not return error for bad processor") } } @@ -162,7 +158,7 @@ func TestRouteHandleNilPayloadFromProcessor(t *testing.T) { Payload: "test", }) if err != nil { - t.Fatalf("route HandleOutput returned error for nil payload: %v", err) + t.Fatalf("route returned error for nil payload: %v", err) } } diff --git a/router.go b/router.go index 20b98b2..b973273 100644 --- a/router.go +++ b/router.go @@ -15,7 +15,6 @@ import ( "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/trace" ) @@ -254,55 +253,6 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any) return routeFound.Load(), routeIOErrors } -func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload any) error { - spanCtx, span := otel.Tracer("router").Start(ctx, "output", trace.WithAttributes(attribute.String("destination.id", destinationId))) - defer span.End() - outputEvent := common.Event{ - Type: "output", - Data: map[string]any{ - "destination": destinationId, - }, - } - destinationModule := r.getModule(destinationId) - - if destinationModule == nil { - err := errors.New("no module found for destination id") - span.SetStatus(codes.Error, err.Error()) - span.RecordError(err) - r.logger.Error("no module found for destination id", "destinationId", destinationId) - outputEvent.Error = err.Error() - r.broadcastEvent(outputEvent) - return err - } - - outputModule, ok := destinationModule.(common.OutputModule) - if !ok { - err := errors.New("module does not support output") - span.SetStatus(codes.Error, err.Error()) - span.RecordError(err) - r.logger.Error("module does not support output", "destinationId", destinationId) - outputEvent.Error = err.Error() - r.broadcastEvent(outputEvent) - return err - } - - moduleOutputCtx, moduleOutputSpan := otel.Tracer("module").Start(spanCtx, "output", trace.WithAttributes(attribute.String("module.id", destinationModule.Id()), attribute.String("module.type", destinationModule.Type()))) - defer moduleOutputSpan.End() - err := outputModule.Output(moduleOutputCtx, payload) - if err != nil { - moduleOutputSpan.SetStatus(codes.Error, err.Error()) - moduleOutputSpan.RecordError(err) - r.logger.ErrorContext(moduleOutputCtx, "module output encountered error", "module", destinationModule.Id(), "error", err) - outputEvent.Error = err.Error() - r.broadcastEvent(outputEvent) - return err - } else { - moduleOutputSpan.SetStatus(codes.Ok, "module output successful") - } - r.broadcastEvent(outputEvent) - return nil -} - func (r *Router) startModules() { for moduleId := range r.ModuleInstances { // TODO(jwetzell): handle module run errors