Fix webflux error handling (#438)

* Minor change to use reactive error handling

* Fix import order

* Update nflow-rest-api-common/src/main/java/io/nflow/rest/v1/ResourceBase.java

Co-authored-by: Edvard Fonsell <[email protected]>

* Update nflow-rest-api-common/src/main/java/io/nflow/rest/v1/ResourceBase.java

Co-authored-by: Edvard Fonsell <[email protected]>

* Change http status magic numbers to constants

* Add fix to changelog

* Add smoke test for checking 404 code

* Refactor netty smoke test a bit and fix indentation

* Fix code formatting

Co-authored-by: Nils Stenholm <[email protected]>
Co-authored-by: Edvard Fonsell <[email protected]>

CHANGELOG.md

@@ -6,6 +6,8 @@
 **Details**
 - `nflow-engine`
   - Add `disabled` state (type `manual`) to `CronWorkflow` to support disabling the work. By default there is no state method for the `disabled` state, but it can added in your workflow definition that extends `CronWorkflow` to execute custom logic when the workflow enters the `disabled` state.
+- `nflow-rest-api-common` and `nflow-rest-api-spring-web`
+  - Fix exception to HTTP status conversion issue when using Netty
 
 ## 7.2.3 (2021-02-22)
 

nflow-netty/src/test/java/io/nflow/netty/StartNflowTest.java

@@ -1,9 +1,12 @@
 package io.nflow.netty;
 
 import static io.nflow.rest.v1.ResourcePaths.NFLOW_WORKFLOW_DEFINITION_PATH;
+import static io.nflow.rest.v1.ResourcePaths.NFLOW_WORKFLOW_INSTANCE_PATH;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.springframework.http.HttpStatus.NOT_FOUND;
 import static org.springframework.http.HttpStatus.OK;
 
 import java.util.HashMap;
@@ -20,8 +23,12 @@
 
 public class StartNflowTest {
 
+  private static final String DEFAULT_LOCALHOST_SERVER_PORT = "7500";
+  private static final String DEFAULT_LOCALHOST_SERVER_ADDRESS = "http://localhost:" + DEFAULT_LOCALHOST_SERVER_PORT;
+
   public class TestApplicationListener implements ApplicationListener<ApplicationContextEvent> {
     public ApplicationContextEvent applicationContextEvent;
+
     @Override
     public void onApplicationEvent(ApplicationContextEvent event) {
       applicationContextEvent = event;
@@ -44,22 +51,35 @@ public void startNflowNetty() throws Exception {
     ApplicationContext ctx = startNflow.startNetty(7500, "local", "", properties);
 
     assertNotNull(testListener.applicationContextEvent);
-    assertEquals("7500", ctx.getEnvironment().getProperty("port"));
+    assertEquals(DEFAULT_LOCALHOST_SERVER_PORT, ctx.getEnvironment().getProperty("port"));
     assertEquals("local", ctx.getEnvironment().getProperty("env"));
     assertEquals("externallyDefinedExecutorGroup", ctx.getEnvironment().getProperty("nflow.executor.group"));
     assertEquals("true", ctx.getEnvironment().getProperty("nflow.db.create_on_startup"));
     assertEquals("false", ctx.getEnvironment().getProperty("nflow.autostart"));
     assertEquals("true", ctx.getEnvironment().getProperty("nflow.autoinit"));
 
     smokeTestRestApi(restApiPrefix);
+    smokeTestRestApiErrorHandling(restApiPrefix);
   }
 
   private void smokeTestRestApi(String restApiPrefix) {
-    WebClient client = WebClient.builder().baseUrl("http://localhost:7500").build();
-    ClientResponse response = client.get().uri(restApiPrefix + NFLOW_WORKFLOW_DEFINITION_PATH).exchange().block();
+    ClientResponse response = getFromDefaultServer(restApiPrefix + NFLOW_WORKFLOW_DEFINITION_PATH);
     assertEquals(OK, response.statusCode());
     JsonNode responseBody = response.bodyToMono(JsonNode.class).block();
     assertTrue(responseBody.isArray());
   }
 
+  // Smoke test for io.nflow.rest.v1.springweb.SpringWebResource#handleExceptions
+  private void smokeTestRestApiErrorHandling(String restApiPrefix) {
+    ClientResponse response = getFromDefaultServer(restApiPrefix + NFLOW_WORKFLOW_INSTANCE_PATH + "/id/0213132");
+    assertEquals(NOT_FOUND, response.statusCode());
+    JsonNode responseBody = response.bodyToMono(JsonNode.class).block();
+    assertNotNull(responseBody);
+    assertFalse(responseBody.isEmpty());
+  }
+
+  private ClientResponse getFromDefaultServer(String url) {
+    WebClient client = WebClient.builder().baseUrl(DEFAULT_LOCALHOST_SERVER_ADDRESS).build();
+    return client.get().uri(url).exchange().block();
+  }
 }

nflow-rest-api-common/src/main/java/io/nflow/rest/v1/ResourceBase.java

@@ -1,6 +1,9 @@
 package io.nflow.rest.v1;
 
 import static io.nflow.engine.workflow.instance.WorkflowInstanceAction.WorkflowActionType.externalChange;
+import static java.net.HttpURLConnection.HTTP_BAD_REQUEST;
+import static java.net.HttpURLConnection.HTTP_INTERNAL_ERROR;
+import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
 import static java.util.Collections.sort;
 import static java.util.Collections.unmodifiableMap;
 import static java.util.stream.Collectors.toCollection;
@@ -183,15 +186,21 @@ public ListWorkflowInstanceResponse fetchWorkflowInstance(final long id, final S
     return listWorkflowConverter.convert(instance, includes);
   }
 
+  protected int resolveExceptionHttpStatus(Throwable t) {
+    if (t instanceof IllegalArgumentException) {
+      return HTTP_BAD_REQUEST;
+    } else if (t instanceof NflowNotFoundException) {
+      return HTTP_NOT_FOUND;
+    }
+    return HTTP_INTERNAL_ERROR;
+  }
+
   protected <T> T handleExceptions(Supplier<T> response, BiFunction<Integer, ErrorResponse, T> error) {
     try {
       return response.get();
-    } catch (IllegalArgumentException e) {
-      return error.apply(400, new ErrorResponse(e.getMessage()));
-    } catch (NflowNotFoundException e) {
-      return error.apply(404, new ErrorResponse(e.getMessage()));
     } catch (Throwable t) {
-      return error.apply(500, new ErrorResponse(t.getMessage()));
+      int code = resolveExceptionHttpStatus(t);
+      return error.apply(code, new ErrorResponse(t.getMessage()));
     }
   }
 }

nflow-rest-api-spring-web/src/main/java/io/nflow/rest/v1/springweb/SpringWebResource.java

@@ -26,7 +26,9 @@ protected Mono<ResponseEntity<?>> wrapBlocking(Callable<ResponseEntity<?>> calla
   }
 
   protected Mono<ResponseEntity<?>> handleExceptions(Supplier<Mono<ResponseEntity<?>>> response) {
-    return handleExceptions(response::get, this::toErrorResponse);
+    return Mono.just(response)
+        .flatMap(Supplier::get)
+        .onErrorResume(ex -> toErrorResponse(resolveExceptionHttpStatus(ex), new ErrorResponse(ex.getMessage())));
   }
 
   private Mono<ResponseEntity<?>> toErrorResponse(int statusCode, ErrorResponse body) {