Support retry in DCAE-SDK DMaaP-Client

Issue-ID: DCAEGEN2-1483
Signed-off-by: tkogut <tomasz.kogut@nokia.com>
Change-Id: Id3f98c0a9367f7c7c2c53ed3eba8805a5a6ab87e

5 files changed, 213 insertions, 17 deletions

diff --git a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java
index 77b842d7..d0bdf414 100644
--- a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java
+++ b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClient.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START====================================
  * DCAEGEN2-SERVICES-SDK
  * =========================================================
- * Copyright (C) 2019-2020 Nokia. All rights reserved.
+ * Copyright (C) 2019-2021 Nokia. All rights reserved.
  * =========================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,18 +17,25 @@
  * limitations under the License.
  * ============LICENSE_END=====================================
  */
+
 package org.onap.dcaegen2.services.sdk.rest.services.adapters.http;
 
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vavr.collection.HashSet;
 import io.vavr.collection.Stream;
 import io.vavr.control.Option;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.config.RetryConfig;
 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.http.client.HttpClient.ResponseReceiver;
 import reactor.netty.http.client.HttpClientRequest;
 import reactor.netty.http.client.HttpClientResponse;
+import reactor.util.retry.Retry;
+import reactor.util.retry.RetryBackoffSpec;
 
 import java.util.stream.Collectors;
 
@@ -39,17 +46,23 @@ public class RxHttpClient {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(RxHttpClient.class);
     private final HttpClient httpClient;
+    private RetryConfig retryConfig;
 
     RxHttpClient(HttpClient httpClient) {
         this.httpClient = httpClient;
     }
 
+    RxHttpClient(HttpClient httpClient, RetryConfig retryConfig) {
+        this(httpClient);
+        this.retryConfig = retryConfig;
+    }
+
     public Mono<HttpResponse> call(HttpRequest request) {
-        return prepareRequest(request)
-                .responseSingle((resp, content) ->
-                        content.asByteArray()
-                                .defaultIfEmpty(new byte[0])
-                                .map(bytes -> new NettyHttpResponse(request.url(), resp.status(), bytes)));
+        Mono<HttpResponse> httpResponseMono = response(request);
+        return Option.of(retryConfig)
+                .map(rc -> retryConfig(rc, request.diagnosticContext()))
+                .map(httpResponseMono::retryWhen)
+                .getOrElse(() -> httpResponseMono);
     }
 
     ResponseReceiver<?> prepareRequest(HttpRequest request) {
@@ -65,6 +78,27 @@ public class RxHttpClient {
         return prepareBody(request, theClient);
     }
 
+    private Mono<HttpResponse> response(HttpRequest request) {
+        return prepareRequest(request)
+                .responseSingle((resp, content) -> mapResponse(request.url(), resp.status(), content));
+    }
+
+    private Mono<HttpResponse> mapResponse(String url, HttpResponseStatus status, reactor.netty.ByteBufMono content) {
+        if (shouldRetry(status.code())) {
+            return Mono.error(new RetryConfig.RetryableException());
+        }
+        return content.asByteArray()
+                .defaultIfEmpty(new byte[0])
+                .map(bytes -> new NettyHttpResponse(url, status, bytes));
+    }
+
+    private boolean shouldRetry(int code) {
+        return Option.of(retryConfig)
+                .map(RetryConfig::retryableHttpResponseCodes)
+                .getOrElse(HashSet::empty)
+                .contains(code);
+    }
+
     private ResponseReceiver<?> prepareBody(HttpRequest request, HttpClient theClient) {
         if (request.body() == null) {
             return prepareBodyWithoutContents(request, theClient);
@@ -79,7 +113,7 @@ public class RxHttpClient {
         return theClient
                 .headers(hdrs -> hdrs.set(HttpHeaders.TRANSFER_ENCODING_TYPE, HttpHeaders.CHUNKED))
                 .request(request.method().asNetty())
-                .send(request.body().contents())
+                .send(Flux.from(request.body().contents()))
                 .uri(request.url());
     }
 
@@ -87,7 +121,7 @@ public class RxHttpClient {
         return theClient
                 .headers(hdrs -> hdrs.set(HttpHeaders.CONTENT_LENGTH, request.body().length().toString()))
                 .request(request.method().asNetty())
-                .send(request.body().contents())
+                .send(Flux.from(request.body().contents()))
                 .uri(request.url());
     }
 
@@ -114,4 +148,22 @@ public class RxHttpClient {
         context.withSlf4jMdc(LOGGER.isDebugEnabled(),
                 () -> LOGGER.debug("Response status: {}", httpClientResponse.status()));
     }
+
+    private RetryBackoffSpec retryConfig(RetryConfig retryConfig, RequestDiagnosticContext context) {
+        RetryBackoffSpec retry = Retry
+                .fixedDelay(retryConfig.retryCount(), retryConfig.retryInterval())
+                .doBeforeRetry(retrySignal -> context.withSlf4jMdc(
+                        LOGGER.isTraceEnabled(), () -> LOGGER.trace("Retry: {}", retrySignal)))
+                .filter(ex -> isRetryable(retryConfig, ex));
+
+        return Option.of(retryConfig.onRetryExhaustedException())
+                .map(ex -> retry.onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> ex))
+                .getOrElse(retry);
+    }
+
+    private boolean isRetryable(RetryConfig retryConfig, Throwable ex) {
+        return retryConfig.retryableExceptions()
+                .toStream()
+                .exists(clazz -> clazz.isAssignableFrom(ex.getClass()));
+    }
 }
diff --git a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientFactory.java b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientFactory.java
index 9b23f1d9..118df52b 100644
--- a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientFactory.java
+++ b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientFactory.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START====================================
  * DCAEGEN2-SERVICES-SDK
  * =========================================================
- * Copyright (C) 2019 Nokia. All rights reserved.
+ * Copyright (C) 2019-2021 Nokia. All rights reserved.
  * =========================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,7 +21,9 @@
 package org.onap.dcaegen2.services.sdk.rest.services.adapters.http;
 
 import io.netty.handler.ssl.SslContext;
+import io.vavr.control.Option;
 import org.jetbrains.annotations.NotNull;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.config.RxHttpClientConfig;
 import org.onap.dcaegen2.services.sdk.security.ssl.SecurityKeys;
 import org.onap.dcaegen2.services.sdk.security.ssl.SslFactory;
 import org.onap.dcaegen2.services.sdk.security.ssl.TrustStoreKeys;
@@ -42,23 +44,53 @@ public final class RxHttpClientFactory {
         return new RxHttpClient(HttpClient.create());
     }
 
+    public static RxHttpClient create(RxHttpClientConfig config) {
+        return createWithConfig(HttpClient.create(), config);
+    }
 
     public static RxHttpClient create(SecurityKeys securityKeys) {
         final SslContext context = SSL_FACTORY.createSecureClientContext(securityKeys);
         return create(context);
     }
 
+    public static RxHttpClient create(SecurityKeys securityKeys, RxHttpClientConfig config) {
+        final SslContext context = SSL_FACTORY.createSecureClientContext(securityKeys);
+        return create(context, config);
+    }
+
     public static RxHttpClient create(TrustStoreKeys trustStoreKeys) {
         final SslContext context = SSL_FACTORY.createSecureClientContext(trustStoreKeys);
         return create(context);
     }
 
+    public static RxHttpClient create(TrustStoreKeys trustStoreKeys, RxHttpClientConfig config) {
+        final SslContext context = SSL_FACTORY.createSecureClientContext(trustStoreKeys);
+        return create(context, config);
+    }
+
     public static RxHttpClient createInsecure() {
         final SslContext context = SSL_FACTORY.createInsecureClientContext();
         return create(context);
     }
 
+    public static RxHttpClient createInsecure(RxHttpClientConfig config) {
+        final SslContext context = SSL_FACTORY.createInsecureClientContext();
+        return create(context, config);
+    }
+
     private static RxHttpClient create(@NotNull SslContext sslContext) {
-        return new RxHttpClient(HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext)));
+        HttpClient secure = HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext));
+        return new RxHttpClient(secure);
+    }
+
+    private static RxHttpClient create(@NotNull SslContext sslContext, RxHttpClientConfig config) {
+        HttpClient secure = HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext));
+        return createWithConfig(secure, config);
+    }
+
+    private static RxHttpClient createWithConfig(HttpClient httpClient, RxHttpClientConfig config) {
+        return Option.of(config.retryConfig())
+                .map(retryConfig -> new RxHttpClient(httpClient, retryConfig))
+                .getOrElse(() -> new RxHttpClient(httpClient));
     }
 }
diff --git a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RetryConfig.java b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RetryConfig.java
new file mode 100644
index 00000000..a0ae1991
--- /dev/null
+++ b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RetryConfig.java
@@ -0,0 +1,59 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2021 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.rest.services.adapters.http.config;
+
+import io.vavr.collection.HashSet;
+import io.vavr.collection.Set;
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
+
+import java.time.Duration;
+
+@Value.Immutable
+public interface RetryConfig {
+
+    int retryCount();
+
+    Duration retryInterval();
+
+    @Value.Default
+    default Set<Integer> retryableHttpResponseCodes() {
+        return HashSet.empty();
+    }
+
+    @Value.Default
+    default Set<Class<? extends Throwable>> customRetryableExceptions() {
+        return HashSet.empty();
+    }
+
+    @Value.Derived
+    default Set<Class<? extends Throwable>> retryableExceptions() {
+        Set<Class<? extends Throwable>> result = customRetryableExceptions();
+        if (retryableHttpResponseCodes().nonEmpty()) {
+            result = result.add(RetryableException.class);
+        }
+        return result;
+    }
+
+    @Nullable RuntimeException onRetryExhaustedException();
+
+    class RetryableException extends RuntimeException {}
+}
diff --git a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RxHttpClientConfig.java b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RxHttpClientConfig.java
new file mode 100644
index 00000000..78a88a47
--- /dev/null
+++ b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/config/RxHttpClientConfig.java
@@ -0,0 +1,29 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2021 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.rest.services.adapters.http.config;
+
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
+
+@Value.Immutable
+public interface RxHttpClientConfig {
+    @Nullable RetryConfig retryConfig();
+}
diff --git a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/test/DummyHttpServer.java b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/test/DummyHttpServer.java
index 4795b00f..8ac0d1d5 100644
--- a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/test/DummyHttpServer.java
+++ b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/test/DummyHttpServer.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START====================================
  * DCAEGEN2-SERVICES-SDK
  * =========================================================
- * Copyright (C) 2019 Nokia. All rights reserved.
+ * Copyright (C) 2019-2021 Nokia. All rights reserved.
  * =========================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,11 +21,8 @@
 package org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test;
 
 import io.vavr.CheckedFunction0;
-import java.net.URL;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Consumer;
+import io.vavr.Tuple3;
+import io.vavr.control.Try;
 import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -35,6 +32,13 @@ import reactor.netty.http.server.HttpServer;
 import reactor.netty.http.server.HttpServerResponse;
 import reactor.netty.http.server.HttpServerRoutes;
 
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+
 /**
  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  * @since February 2019
@@ -63,11 +67,26 @@ public class DummyHttpServer {
         return responses[state.getAndIncrement()];
     }
 
+    public static Publisher<Void> sendInOrderWithDelay(AtomicInteger counter, Tuple3<HttpServerResponse, Integer, Duration>... responses) {
+        Tuple3<HttpServerResponse, Integer, Duration> tuple = responses[counter.get()];
+        HttpServerResponse httpServerResponse = tuple._1;
+        Integer statusCode = tuple._2;
+        long timeout = tuple._3.toMillis();
+        Try.run(() -> Thread.sleep(timeout));
+        counter.incrementAndGet();
+        return sendString(httpServerResponse.status(statusCode), Mono.just("OK"));
+    }
+
+    public static Publisher<Void> sendWithDelay(HttpServerResponse response, int statusCode, Duration timeout) {
+        Try.run(() -> Thread.sleep(timeout.toMillis()));
+        return sendString(response.status(statusCode), Mono.just("OK"));
+    }
+
     public static Publisher<Void> sendResource(HttpServerResponse httpServerResponse, String resourcePath) {
         return sendString(httpServerResponse, Mono.fromCallable(() -> readResource(resourcePath)));
     }
 
-    public static Publisher<Void> sendError(HttpServerResponse httpServerResponse, int statusCode, String message){
+    public static Publisher<Void> sendError(HttpServerResponse httpServerResponse, int statusCode, String message) {
         return sendString(httpServerResponse.status(statusCode), Mono.just(message));
     }
 
@@ -79,6 +98,11 @@ public class DummyHttpServer {
         server.disposeNow();
     }
 
+    public DummyHttpServer closeAndGet() {
+        server.disposeNow();
+        return this;
+    }
+
     public String host() {
         return server.host();
     }