Emit CBS config updates only

Add an update method to CbsClient which emits an item only when the
configuration has accentually changed.

Change-Id: I6023fb1cc069b06bd2c4baf94406538965b6534c
Issue-ID: DCAEGEN2-1233
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

1 files changed, 21 insertions, 1 deletions

diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java
index 309bb62f..8a0977d9 100644
--- a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java
@@ -98,7 +98,8 @@ class CbsClientImplIT {
         final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
 
         // when
-        final Flux<JsonObject> result = sut.flatMapMany(cbsClient -> cbsClient.get(diagnosticContext, Duration.ZERO, Duration.ofMillis(10)));
+        final Flux<JsonObject> result = sut
+                .flatMapMany(cbsClient -> cbsClient.get(diagnosticContext, Duration.ZERO, Duration.ofMillis(10)));
 
         // then
         final int itemsToTake = 5;
@@ -108,6 +109,25 @@ class CbsClientImplIT {
                 .verify(Duration.ofSeconds(5));
     }
 
+    @Test
+    void testCbsClientWithUpdatesCall() {
+        // given
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
+        final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
+        final Duration period = Duration.ofMillis(10);
+
+        // when
+        final Flux<JsonObject> result = sut
+                .flatMapMany(cbsClient -> cbsClient.updates(diagnosticContext, Duration.ZERO, period));
+
+        // then
+        final Duration timeToCollectItemsFor = period.multipliedBy(50);
+        StepVerifier.create(result.take(timeToCollectItemsFor).map(this::sampleConfigValue))
+                .expectNext(EXPECTED_CONFIG_VALUE)
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
+
     private String sampleConfigValue(JsonObject obj) {
         return obj.get(SAMPLE_CONFIG_KEY).getAsString();
     }