#318 Added functionality to do non-blocking HTTP requests

buerokratt · Oct 16, 2024 · 6260c15 · 6260c15
1 parent ffeec3c
commit 6260c15
Show file tree

Hide file tree

Showing 5 changed files with 27 additions and 21 deletions.
diff --git a/src/main/java/ee/buerokratt/ruuter/domain/steps/http/HttpGetStep.java b/src/main/java/ee/buerokratt/ruuter/domain/steps/http/HttpGetStep.java
@@ -22,7 +22,7 @@ public ResponseEntity<Object> getRequestResponse(DslInstance di) {
         Map<String, Object> evaluatedQuery = di.getScriptingHelper().evaluateScripts(args.getQuery(), di.getContext(), di.getRequestBody(), di.getRequestQuery(), di.getRequestHeaders());
         Map<String, Object> evaluatedHeaders = di.getScriptingHelper().evaluateScripts(args.getHeaders(), di.getContext(), di.getRequestBody(), di.getRequestQuery(), di.getRequestHeaders());
         Map<String, String> mappedHeaders = di.getMappingHelper().convertMapObjectValuesToString(evaluatedHeaders);
-        return di.getHttpHelper().doMethod(HttpMethod.GET, evaluatedURL, evaluatedQuery, null, mappedHeaders, args.getContentType() , null, getLimit(), di, args.isDynamicParameters());
+        return di.getHttpHelper().doMethod(HttpMethod.GET, evaluatedURL, evaluatedQuery, null, mappedHeaders, args.getContentType() , null, getLimit(), di, args.isDynamicParameters(), resultName != null && !resultName.isEmpty() );
     }
 
     @Override

diff --git a/src/main/java/ee/buerokratt/ruuter/domain/steps/http/HttpPostStep.java b/src/main/java/ee/buerokratt/ruuter/domain/steps/http/HttpPostStep.java
@@ -32,7 +32,7 @@ protected ResponseEntity<Object> getRequestResponse(DslInstance di) {
             args.getContentType(),
             "plaintext".equals(args.getContentType()) ? args.getPlaintext() : null,
             getLimit(), di,
-            args.isDynamicParameters());
+            args.isDynamicParameters(),resultName != null && !resultName.isEmpty() );
     }
 
     @Override

diff --git a/src/main/java/ee/buerokratt/ruuter/domain/steps/http/HttpStep.java b/src/main/java/ee/buerokratt/ruuter/domain/steps/http/HttpStep.java
@@ -77,7 +77,7 @@ protected void executeStepAction(DslInstance di) {
     @Override
     public void handleFailedResult(DslInstance di) {
         super.handleFailedResult(di);
-        HttpStepResult stepResult = (HttpStepResult) di.getContext().get(resultName);
+        HttpStepResult stepResult = (HttpStepResult) di.getContext().get(                   resultName);
         if (stepResult != null && !isAllowedHttpStatusCode(di, stepResult.getResponse().getStatusCodeValue())) {
             DefaultHttpDsl globalHttpExceptionDsl = di.getProperties().getDefaultDslInCaseOfException();
             if (localHttpExceptionDslExists()) {

diff --git a/src/main/java/ee/buerokratt/ruuter/helper/ExternalForwardingHelper.java b/src/main/java/ee/buerokratt/ruuter/helper/ExternalForwardingHelper.java
@@ -58,10 +58,10 @@ public ResponseEntity<Object> forwardRequest(String dsl, Map<String, Object> req
             .toUriString();
 
         if (methodType.equals(HttpMethod.POST.name())) {
-            return httpHelper.doMethod(HttpMethod.POST,forwardingUrl, query, body, headers, contentType, null, null, di, false);
+            return httpHelper.doMethod(HttpMethod.POST,forwardingUrl, query, body, headers, contentType, null, null, di, false, true );
         }
         if (methodType.equals(HttpMethod.GET.name())) {
-            return httpHelper.doMethod(HttpMethod.GET, forwardingUrl,  query,null, headers, null, null, null, di, false);
+            return httpHelper.doMethod(HttpMethod.GET, forwardingUrl,  query,null, headers, null, null, null, di, false, true                        );
         }
         throw new InvalidHttpMethodTypeException(methodType);
     }

diff --git a/src/main/java/ee/buerokratt/ruuter/helper/HttpHelper.java b/src/main/java/ee/buerokratt/ruuter/helper/HttpHelper.java
@@ -2,7 +2,6 @@
 
 import ee.buerokratt.ruuter.configuration.ApplicationProperties;
 import ee.buerokratt.ruuter.domain.DslInstance;
-import ee.buerokratt.ruuter.util.LoggingUtils;
 import io.netty.channel.ChannelOption;
 import io.netty.handler.timeout.ReadTimeoutHandler;
 import io.netty.handler.timeout.WriteTimeoutHandler;
@@ -22,6 +21,7 @@
 import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.reactive.function.client.WebClientResponseException;
 
+import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
 import reactor.netty.http.client.HttpClient;
 
@@ -46,24 +46,25 @@ public class HttpHelper {
     public ResponseEntity<Object> doPost(String url, Map<String, Object> body, Map<String, Object> query, Map<String, String> headers, DslInstance di, boolean dynamicBody) {
         return doPost(url, body, query, headers, this.getClass().getName(), di, dynamicBody);
     }
+
     public ResponseEntity<Object> doPost(String url, Map<String, Object> body, Map<String, Object> query, Map<String, String> headers, String contentType, DslInstance di, boolean dynamicBody) {
-        return doMethod(POST, url, query, body,headers, contentType, null, null, di, dynamicBody);
+        return doMethod(POST, url, query, body,headers, contentType, null, null, di, dynamicBody, true);
     }
 
     public ResponseEntity<Object> doPostPlaintext(String url, Map<String, Object> body, Map<String, Object> query, Map<String, String> headers, String plaintext, DslInstance di) {
-        return doMethod(POST, url, body, query, headers, "plaintext", plaintext, null, di, false);
+        return doMethod(POST, url, body, query, headers, "plaintext", plaintext, null, di, false, true);
     }
 
     public ResponseEntity<Object> doGet(String url, Map<String, Object> query, Map<String, String> headers, DslInstance di) {
-        return doMethod(HttpMethod.GET, url, query, null, headers, null, null, null, di, false);
+        return doMethod(HttpMethod.GET, url, query, null, headers, null, null, null, di, false, true );
     }
 
-    public ResponseEntity<Object> doPut(String url, Map<String, Object> body, Map<String, Object> query, Map<String, String> headers, String contentType, DslInstance di, boolean dynamicBody) {
-        return doMethod(HttpMethod.PUT, url, query, body,headers, contentType, null, null, di, dynamicBody);
+    public ResponseEntity<Object> doPut(String url, Map<String, Object> body, Map<String, Object> query, Map<String, String> headers, String contentType, DslInstance di, boolean dynamicBody, boolean blockResult) {
+        return doMethod(HttpMethod.PUT, url, query, body,headers, contentType, null, null, di, dynamicBody, true);
     }
 
     public ResponseEntity<Object> doDelete(String url, Map<String, Object> body, Map<String, Object> query, Map<String, String> headers, String contentType, DslInstance di) {
-        return doMethod(HttpMethod.DELETE, url, query, body, headers, contentType, null, null, di, false);
+        return doMethod(HttpMethod.DELETE, url, query, body, headers, contentType, null, null, di, false, true);
     }
 
     public ResponseEntity<Object> doMethod(HttpMethod method,
@@ -75,7 +76,7 @@ public ResponseEntity<Object> doMethod(HttpMethod method,
                                            String plaintextValue,
                                            Integer limit,
                                            DslInstance instance,
-                                           boolean dynamicBody) {
+                                           boolean dynamicBody, boolean blockResult) {
         try {
             MultiValueMap<String, String> qp = new LinkedMultiValueMap<>(
                 query.entrySet().stream().collect(Collectors.toMap(e -> e.getKey(), e-> Arrays.asList(e.getValue().toString()))));
@@ -126,24 +127,29 @@ public ResponseEntity<Object> doMethod(HttpMethod method,
             }
 
             Integer finalLimit = limit == null ? properties.getHttpResponseSizeLimit() : limit;
-            return WebClient.builder()
+            Mono<ResponseEntity<Object>> retrieve = WebClient.builder()
                 .filter((request, next) -> !("json_override".equals(contentType)) ? next.exchange(request)
-                     :next.exchange(request)
-                        .flatMap(response -> Mono.just(response.mutate()
-                            .headers(httpHeaders -> httpHeaders.set(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE))
-                            .build())))
+                    : next.exchange(request)
+                    .flatMap(response -> Mono.just(response.mutate()
+                        .headers(httpHeaders -> httpHeaders.set(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE))
+                        .build())))
                 .exchangeStrategies(
                     ExchangeStrategies.builder().codecs(
-                        configurer -> configurer.defaultCodecs().maxInMemorySize(finalLimit * 1024 )).build())
+                        configurer -> configurer.defaultCodecs().maxInMemorySize(finalLimit * 1024)).build())
                 .clientConnector(new ReactorClientHttpConnector(getHttpClient())).build()
                 .method(method)
                 .uri(url, uriBuilder -> uriBuilder.queryParams(qp).build())
                 .headers(httpHeaders -> addHeadersIfNotNull(headers, httpHeaders))
                 .body(bodyValue)
                 .header(HttpHeaders.CONTENT_TYPE, mediaType)
                 .retrieve()
-                .toEntity(Object.class)
-                .block();
+                .toEntity(Object.class);
+            if (blockResult)
+                return retrieve.block();
+            else {
+                Disposable dis = retrieve.subscribe();
+                return ResponseEntity.ok(null);
+            }
         } catch (WebClientResponseException e) {
             log.error("Failed HTTP request: ", e);
             return new ResponseEntity<>(e.getStatusText(), e.getStatusCode());