Merge pull request #30 from jwetzell/fix/multi-out-errors

fix error handling/short-circuiting in multi route matching

internal/module/http-server.go

@@ -20,7 +20,15 @@ type HTTPServer struct {
 	logger *slog.Logger
 }
 
+type ResponseIOError struct {
+	Index        int      `json:"index"`
+	OutputErrors []string `json:"outputErrors"`
+	ProcessError *string  `json:"processError"`
+	InputError   *string  `json:"inputError"`
+}
+
 type ResponseData struct {
+	IOErrors []ResponseIOError `json:"ioErrors"`
 	Message  string            `json:"message"`
 	Status   string            `json:"status"`
 }
@@ -43,6 +51,8 @@ func init() {
 
 			router, ok := ctx.Value(route.RouterContextKey).(route.RouteIO)
 
+			fmt.Printf("%+T", ctx.Value(route.RouterContextKey))
+
 			if !ok {
 				return nil, errors.New("http.server unable to get router from context")
 			}
@@ -69,15 +79,51 @@ func (hs *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if hs.router != nil {
-		routingErrors := hs.router.HandleInput(hs.Id(), r)
+		aRouteFound, routingErrors := hs.router.HandleInput(hs.Id(), r)
+		if aRouteFound {
 			if routingErrors != nil {
 				w.WriteHeader(http.StatusInternalServerError)
 				response.Status = "error"
 				response.Message = "routing failed"
+
+				response.IOErrors = []ResponseIOError{}
+				for _, responseIOError := range routingErrors {
+					errorToAdd := ResponseIOError{
+						Index: responseIOError.Index,
+					}
+
+					if responseIOError.InputError != nil {
+						errorMsg := responseIOError.InputError.Error()
+						errorToAdd.InputError = &errorMsg
+					}
+
+					if responseIOError.ProcessError != nil {
+						errorMsg := responseIOError.ProcessError.Error()
+						errorToAdd.ProcessError = &errorMsg
+					}
+
+					if responseIOError.OutputErrors != nil {
+						outputErrorMsgs := []string{}
+
+						for _, outputError := range responseIOError.OutputErrors {
+							outputErrorMsgs = append(outputErrorMsgs, outputError.Error())
+						}
+
+						errorToAdd.OutputErrors = outputErrorMsgs
+					}
+
+					response.IOErrors = append(response.IOErrors, errorToAdd)
+
+				}
 			} else {
 				w.WriteHeader(http.StatusOK)
 				response.Message = "routing successful"
 			}
+		} else {
+			w.WriteHeader(http.StatusNotFound)
+			response.Status = "error"
+			response.Message = "no matching routes found"
+		}
 	} else {
 		w.WriteHeader(http.StatusInternalServerError)
 		response.Message = "no router registered"

internal/route/route.go

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

router.go

@@ -129,34 +129,58 @@ func (r *Router) Stop() {
 	r.contextCancel()
 }
 
-func (r *Router) HandleInput(sourceId string, payload any) []route.RouteIOError {
+func (r *Router) HandleInput(sourceId string, payload any) (bool, []route.RouteIOError) {
-	var routingErrors []route.RouteIOError
+	var routeIOErrors []route.RouteIOError
+	routeFound := false
 	for routeIndex, routeInstance := range r.RouteInstances {
 		if routeInstance.Input() == sourceId {
+			routeFound = true
 			routeContext := context.WithValue(r.Context, route.SourceContextKey, sourceId)
 
 			payload, err := routeInstance.ProcessPayload(routeContext, payload)
 			if err != nil {
-				if routingErrors == nil {
+				if routeIOErrors == nil {
-					routingErrors = []route.RouteIOError{}
+					routeIOErrors = []route.RouteIOError{}
 				}
-				routingErrors = append(routingErrors, route.RouteIOError{
+				r.logger.Error("unable to process input", "route", routeIndex, "source", sourceId, "error", err)
+				routeIOErrors = append(routeIOErrors, route.RouteIOError{
 					Index:        routeIndex,
-					Error: err,
+					ProcessError: err,
 				})
-				r.logger.Error("unable to route input", "route", routeIndex, "source", sourceId, "error", err)
+				continue
-			}
-			r.HandleOutput(routeContext, routeInstance.Output(), payload)
-		}
-	}
-	return routingErrors
 			}
 
-func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload any) error {
+			outputErrors := r.HandleOutput(routeContext, routeInstance.Output(), payload)
+			if outputErrors != nil {
+				if routeIOErrors == nil {
+					routeIOErrors = []route.RouteIOError{}
+				}
+				routeIOErrors = append(routeIOErrors, route.RouteIOError{
+					Index:        routeIndex,
+					OutputErrors: outputErrors,
+				})
+			}
+
+		}
+	}
+	return routeFound, routeIOErrors
+}
+
+func (r *Router) HandleOutput(ctx context.Context, destinationId string, payload any) []error {
+
+	var outputErrors []error
 	for _, moduleInstance := range r.ModuleInstances {
 		if moduleInstance.Id() == destinationId {
-			return moduleInstance.Output(ctx, payload)
+			err := moduleInstance.Output(ctx, payload)
+			if err != nil {
+				if outputErrors == nil {
+					outputErrors = []error{}
+				}
+				outputErrors = append(outputErrors, err)
+			}
+			// r.logger.Error("unable to route output", "module", moduleInstance.Id(), "error", err)
 		}
 	}
-	return fmt.Errorf("router could not find module instance for destination %s", destinationId)
+	fmt.Println(len(outputErrors))
+	return outputErrors
 }