fix: properly handle fire and forget in V4

Author: Guillaume Lamirand

pom.xml

@@ -46,6 +46,8 @@
         <gravitee-reactor-message.version>3.0.0</gravitee-reactor-message.version>
         <gravitee-entrypoint-sse.version>4.1.0</gravitee-entrypoint-sse.version>
 
+        <awaitility.version>4.2.1</awaitility.version>
+
         <maven-assembly-plugin.version>3.7.1</maven-assembly-plugin.version>
         <json-schema-generator-maven-plugin.version>1.3.0</json-schema-generator-maven-plugin.version>
         <json-schema-generator-maven-plugin.outputDirectory>${project.build.directory}/schemas</json-schema-generator-maven-plugin.outputDirectory>
@@ -187,6 +189,12 @@
             <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/io/gravitee/policy/callout/CalloutHttpPolicy.java

@@ -66,6 +66,14 @@ public Completable onResponse(HttpExecutionContext ctx) {
     }
 
     private Completable doCallOut(HttpExecutionContext ctx) {
+        if (configuration.isFireAndForget()) {
+            return Completable.fromRunnable(() -> executeCallOut(ctx).onErrorComplete().subscribe());
+        } else {
+            return executeCallOut(ctx);
+        }
+    }
+
+    private Completable executeCallOut(HttpExecutionContext ctx) {
         var templateEngine = ctx.getTemplateEngine();
         var vertx = ctx.getComponent(Vertx.class);
 

src/test/java/io/gravitee/policy/callout/AbstractV4EngineTest.java

@@ -40,6 +40,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Stream;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
 @GatewayTest
@@ -50,6 +51,11 @@ public class AbstractV4EngineTest extends AbstractPolicyTest<CalloutHttpPolicy,
     @RegisterExtension
     static WireMockExtension calloutServer = WireMockExtension.newInstance().options(wireMockConfig().dynamicPort()).build();
 
+    @BeforeEach
+    public void cleanStub() {
+        calloutServer.resetAll();
+    }
+
     @Override
     public void configureEntrypoints(Map<String, EntrypointConnectorPlugin<?, ?>> entrypoints) {
         entrypoints.putIfAbsent("http-proxy", EntrypointBuilder.build("http-proxy", HttpProxyEntrypointConnectorFactory.class));

src/test/java/io/gravitee/policy/callout/CalloutHttpPolicyV4IntegrationTest.java

@@ -24,6 +24,7 @@
 import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 import io.gravitee.apim.gateway.tests.sdk.annotations.DeployApi;
 import io.gravitee.common.http.MediaType;
@@ -62,7 +63,7 @@ void should_do_callout_and_process_response(String path, HttpClient client) {
                 .test();
 
             obs
-                .awaitDone(30, TimeUnit.SECONDS)
+                .awaitDone(5, TimeUnit.SECONDS)
                 .assertComplete()
                 .assertValue(response -> {
                     assertThat(response).hasToString("new content: response from callout");
@@ -79,9 +80,9 @@ void should_do_callout_and_process_response(String path, HttpClient client) {
             { "/apis/v4/proxy/request-callout-http-fire-and-forget.json", "/apis/v4/proxy/response-callout-http-fire-and-forget.json" }
         )
         @ValueSource(strings = { "/proxy-request-fire-and-forget", "/proxy-response-fire-and-forget" })
-        void should_do_callout_and_do_nothing_with_callout_response(String apiEndpoint, HttpClient client) {
+        void should_do_callout_with_fire_and_forget_and_do_nothing_with_callout_response(String apiEndpoint, HttpClient client) {
             wiremock.stubFor(get("/endpoint").willReturn(ok("response from backend")));
-            calloutServer.stubFor(get("/callout").willReturn(ok("response from callout")));
+            calloutServer.stubFor(get("/callout").willReturn(ok("response from callout").withFixedDelay(8000)));
 
             var obs = client
                 .rxRequest(HttpMethod.GET, apiEndpoint)
@@ -93,7 +94,7 @@ void should_do_callout_and_do_nothing_with_callout_response(String apiEndpoint,
                 .test();
 
             obs
-                .awaitDone(30, TimeUnit.SECONDS)
+                .awaitDone(5, TimeUnit.SECONDS)
                 .assertComplete()
                 .assertValue(response -> {
                     assertThat(response).hasToString("new content");
@@ -102,7 +103,12 @@ void should_do_callout_and_do_nothing_with_callout_response(String apiEndpoint,
                 .assertNoErrors();
 
             wiremock.verify(getRequestedFor(urlPathEqualTo("/endpoint")));
-            calloutServer.verify(getRequestedFor(urlPathEqualTo("/callout")).withHeader("X-Callout", equalTo("calloutHeader")));
+
+            await()
+                .atMost(10, TimeUnit.SECONDS)
+                .untilAsserted(() ->
+                    calloutServer.verify(getRequestedFor(urlPathEqualTo("/callout")).withHeader("X-Callout", equalTo("calloutHeader")))
+                );
         }
 
         @ParameterizedTest
@@ -130,7 +136,7 @@ void should_process_el_before_callout(String apiEndpoint, HttpClient client) {
                 .test();
 
             obs
-                .awaitDone(30, TimeUnit.SECONDS)
+                .awaitDone(5, TimeUnit.SECONDS)
                 .assertComplete()
                 .assertValue(response -> {
                     assertThat(response).hasToString("new content");
@@ -164,7 +170,7 @@ void should_do_callout_and_interrupt_when_callout_fails(String apiEndpoint, Http
                 .test();
 
             obs
-                .awaitDone(30, TimeUnit.SECONDS)
+                .awaitDone(5, TimeUnit.SECONDS)
                 .assertComplete()
                 .assertValue(response -> {
                     assertThat(response).hasToString("Error content with status 502");

src/test/java/io/gravitee/policy/callout/CalloutHttpPolicyV4Test.java

@@ -25,6 +25,7 @@
 import static io.gravitee.policy.v3.callout.CalloutHttpPolicyV3.CALLOUT_EXIT_ON_ERROR;
 import static io.gravitee.policy.v3.callout.CalloutHttpPolicyV3.CALLOUT_HTTP_ERROR;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static test.RequestBuilder.aRequest;
@@ -341,11 +342,10 @@ void should_call_and_do_nothing_when_fire_and_forget_defined(boolean exitOnError
             )
                 .onRequest(ctx)
                 .test()
-                .awaitDone(30, TimeUnit.SECONDS)
                 .assertComplete();
 
             assertThat(ctx.getAttributes()).isEmpty();
-            wiremock.verify(getRequestedFor(urlPathEqualTo("/")));
+            await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> wiremock.verify(getRequestedFor(urlPathEqualTo("/"))));
         }
 
         @Test
@@ -662,11 +662,10 @@ void should_call_and_do_nothing_when_fire_and_forget_defined(boolean exitOnError
             )
                 .onResponse(ctx)
                 .test()
-                .awaitDone(30, TimeUnit.SECONDS)
                 .assertComplete();
 
             assertThat(ctx.getAttributes()).isEmpty();
-            wiremock.verify(getRequestedFor(urlPathEqualTo("/")));
+            await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> wiremock.verify(getRequestedFor(urlPathEqualTo("/"))));
         }
 
         @Test

src/test/resources/apis/v4/proxy/response-callout-http-el-support.json

@@ -57,7 +57,7 @@
                             }
                         ],
                         "body": "{#request.headers['X-Body'][0]}",
-                        "fireAndForget": true
+                        "fireAndForget": false
                     }
                 },
                 {