Merge pull request #320 from buerokratt/add-http-timeout-parameter

Add http timeout parameter & non-blocking HTTP requests

samples/steps/http-get.md

@@ -27,7 +27,12 @@ get_step:
     * `headers`
         * *..desired header values* - Scripts can be used for headers values
 * `result` - name of the variable to store the response of the query in, for use in other steps
+  - __NB!__ if `result` field is missing, the request will be done immediately and
+  non-blockingly, so the DSL execution does not wait for response. Any data that would
+  be sent with response is not handled by Ruuter.
 * `limit` - limit the size of allowed response in kilobytes (default value is configured in application.yaml) 
+* `timeout`- (in milliseconds) overwrites http request timeout set in application properties (or if not
+defined there, 15000ms)
 
 #### How responses are stored with the result field
 

samples/steps/http-post.md

@@ -31,6 +31,9 @@ post_step:
     * `headers`
         * *..desired header values* - Scripts can be used for headers values
     * `result` - name of the variable to store the response of the query in, for use in other steps
+      - __NB!__ if `result` field is missing, the request will be done immediately and
+non-blockingly, so the DSL execution does not wait for response. Any data that would
+be sent with response is not handled by Ruuter.
     * `contentType` - specifies the contenttype to use, currently allowed values:
       * `"plaintext"` - uses field `plaintext` and mediaType 'text/plain'
       * `"formdata"` 
@@ -44,6 +47,8 @@ the key, for example `file:projectdata:Project.csv`, and mediatype "multipart/fo
       * If left empty, `body` is posted as JSON and 'application/json' is used as mediatype.
     * `plaintext` - used instead of `body` if a singular plaintext value is needed to be sent 
 * `limit` - limit the size of allowed response in kilobytes (default value is configured in application.yaml)
+* `timeout`- (in milliseconds) overwrites http request timeout set in application properties (or if not
+  defined there, 15000ms)
 
 * ***Note: POST step responses are stored the same way as [GET step responses](./http-get.md#How-responses-are-stored-with-the-result-field)***
 

src/main/java/ee/buerokratt/ruuter/configuration/ApplicationProperties.java

@@ -32,6 +32,8 @@ public class ApplicationProperties {
 
     private Boolean allowDuplicateRequestKeys;
 
+    private Integer httpRequestTimeout;
+
     @Setter
     @Getter
     public static class FinalResponse {

src/main/java/ee/buerokratt/ruuter/domain/steps/conditional/SwitchStep.java

@@ -25,10 +25,19 @@ public class SwitchStep extends DslStep {
 
     @Override
     protected void executeStepAction(DslInstance di) {
+/*
+        Optional<Condition> correctStatement =
+            conditions.stream()
+                .filter(condition -> condition.getConditionStatement().equals(di.getRequestQuery().get("metric")))
+    .findFirst();
+
+        System.out.println("METRIC: " + di.getRequestQuery().get("metric"));
+*/
         ScriptingHelper scriptingHelper = di.getScriptingHelper();
         Optional<Condition> correctStatement = conditions.stream()
             .filter(condition -> Boolean.TRUE.equals(scriptingHelper.evaluateScripts(condition.getConditionStatement(), di.getContext(), di.getRequestBody(), di.getRequestQuery(), di.getRequestHeaders())))
             .findFirst();
+
         correctStatement.ifPresentOrElse(condition -> this.setNextStepName(condition.getNextStepName()), () -> this.setNextStepName(elseNextStepName));
     }
 

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(), getTimeout() );
     }
 
     @Override

src/main/java/ee/buerokratt/ruuter/domain/steps/http/HttpPostStep.java

@@ -32,7 +32,9 @@ 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(),
+	    getTimeout() );
     }
 
     @Override

src/main/java/ee/buerokratt/ruuter/domain/steps/http/HttpStep.java

@@ -42,6 +42,7 @@ public abstract class HttpStep extends DslStep {
     protected Logging logging;
 
     protected Integer limit;
+    protected Integer timeout;
 
     @JsonAlias("error")
     protected String onErrorStep;
@@ -77,7 +78,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()) {

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, null );
         }
         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, null);
         }
         throw new InvalidHttpMethodTypeException(methodType);
     }

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;
 
@@ -43,27 +43,28 @@ public class HttpHelper {
 
     final private ScriptingHelper scriptingHelper;
 
-    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, DslInstance di, boolean dynamicBody, Integer timeout) {
+        return doPost(url, body, query, headers, this.getClass().getName(), di, dynamicBody, timeout);
     }
-    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);
+
+    public ResponseEntity<Object> doPost(String url, Map<String, Object> body, Map<String, Object> query, Map<String, String> headers, String contentType, DslInstance di, boolean dynamicBody, Integer timeout ) {
+        return doMethod(POST, url, query, body,headers, contentType, null, null, di, dynamicBody, true, timeout);
     }
 
     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, null);
     }
 
-    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);
+    public ResponseEntity<Object> doGet(String url, Map<String, Object> query, Map<String, String> headers, DslInstance di, Integer timeout) {
+        return doMethod(HttpMethod.GET, url, query, null, headers, null, null, null, di, false, true, timeout );
     }
 
-    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, null);
     }
 
     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, null);
     }
 
     public ResponseEntity<Object> doMethod(HttpMethod method,
@@ -75,7 +76,9 @@ public ResponseEntity<Object> doMethod(HttpMethod method,
                                            String plaintextValue,
                                            Integer limit,
                                            DslInstance instance,
-                                           boolean dynamicBody) {
+                                           boolean dynamicBody,
+                                           boolean blockResult,
+                                           Integer timeout) {
         try {
             MultiValueMap<String, String> qp = new LinkedMultiValueMap<>(
                 query.entrySet().stream().collect(Collectors.toMap(e -> e.getKey(), e-> Arrays.asList(e.getValue().toString()))));
@@ -126,24 +129,30 @@ 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())
-                .clientConnector(new ReactorClientHttpConnector(getHttpClient())).build()
+
+                        configurer -> configurer.defaultCodecs().maxInMemorySize(finalLimit * 1024)).build())
+                .clientConnector(new ReactorClientHttpConnector(getHttpClient(timeout))).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());
@@ -156,12 +165,18 @@ private void addHeadersIfNotNull(Map<String, String> headers, HttpHeaders httpHe
         }
     }
 
-    private HttpClient getHttpClient() {
+    private HttpClient getHttpClient(Integer timeout) {
+        final Integer finalTimeout = timeout != null ? timeout :
+            properties.getHttpRequestTimeout() != null ? properties.getHttpRequestTimeout() :
+                15000;
+
+        log.info("HTTP request effective timeout: "+ finalTimeout);
+
         return HttpClient.create()
-            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 15000)
-                    .responseTimeout(Duration.ofMillis(15000))
+            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, finalTimeout)
+                    .responseTimeout(Duration.ofMillis(finalTimeout))
             .doOnConnected(conn ->
-                conn.addHandlerLast(new ReadTimeoutHandler(15000, TimeUnit.MILLISECONDS))
-                    .addHandlerLast(new WriteTimeoutHandler(15000, TimeUnit.MILLISECONDS)));
+                conn.addHandlerLast(new ReadTimeoutHandler(finalTimeout, TimeUnit.MILLISECONDS))
+                    .addHandlerLast(new WriteTimeoutHandler(finalTimeout, TimeUnit.MILLISECONDS)));
     }
 }