PoC of change-based configuration using MerkleTree

The goal of this MR is to check if storing application configuration
fetched from CBS in Merkle Tree would be a good approach.

CoS:
- it should be possible to refetch configuration at some intervals
- any new update should trigger changes in affected modules, ie. change
  of dmaap topics should trigger changes in dmaap client but not in aai
  client

In order to satisfy this requirement a MerkleTree-based solution was
proposed. The structure keeps track of hashes of each node and its
subnodes. In case of a change in one branch the hash of other sibling
branch will not change.

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

2 files changed, 235 insertions, 0 deletions

diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ListenableCbsConfigTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ListenableCbsConfigTest.java
new file mode 100644
index 00000000..3e77251d
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ListenableCbsConfigTest.java
@@ -0,0 +1,138 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 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.cbs.client.api.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.vavr.collection.List;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.listener.ListenableCbsConfig;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.listener.MerkleTree;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.ReplayProcessor;
+import reactor.test.StepVerifier;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since February 2019
+ */
+class ListenableCbsConfigTest {
+
+    @Test
+    void listen_shouldCallListenerAfterEachChange() {
+        ListenableCbsConfig cut = new ListenableCbsConfig();
+
+        final List<String> expectedChanges = List.of("1", "2", "3");
+        final AtomicReference<List<String>> actualChanges = new AtomicReference<>(List.empty());
+
+        cut.listen(List.of("some-key"), subtreeOption ->
+                actualChanges.updateAndGet(
+                        changes -> changes.append(subtreeOption.flatMap(subtree -> subtree.get()).getOrElse("[None]")))
+
+        );
+
+        final MerkleTree<String> initialConfig = MerkleTree.<String>emptyWithDefaultDigest(String::getBytes)
+                .add(List.of("some-key"), "1");
+
+        final MerkleTree<String> updatedConfig1 = initialConfig
+                .add(List.of("some-key"), "2");
+
+        final MerkleTree<String> updatedConfig2 = updatedConfig1
+                .add(List.of("some-key"), "3");
+
+        cut.update(initialConfig);
+        cut.update(updatedConfig1);
+        cut.update(updatedConfig2);
+
+        assertThat(actualChanges.get()).isEqualTo(expectedChanges);
+
+    }
+
+
+    @Test
+    void subtreeChanges_shouldEmitItemOnEachChange() {
+        ListenableCbsConfig cut = new ListenableCbsConfig();
+
+        final ReplayProcessor<String> replayProcessor = ReplayProcessor.create();
+        final List<String> expectedChanges = List.of("1", "2", "3");
+
+        cut.subtreeChanges(List.of("some-key"))
+                .map(subtreeOption ->
+                        subtreeOption.flatMap(subtree -> subtree.get()).getOrElse("[None]")
+                )
+                .subscribe(replayProcessor);
+
+        final MerkleTree<String> initialConfig = MerkleTree.<String>emptyWithDefaultDigest(String::getBytes)
+                .add(List.of("some-key"), "1");
+
+        final MerkleTree<String> updatedConfig1 = initialConfig
+                .add(List.of("some-key"), "2");
+
+        final MerkleTree<String> updatedConfig2 = updatedConfig1
+                .add(List.of("some-key"), "3");
+
+        cut.subscribeForUpdates(Flux.just(initialConfig, updatedConfig1, updatedConfig2));
+
+        StepVerifier.create(replayProcessor.take(expectedChanges.size()))
+                .expectNextSequence(expectedChanges)
+                .verifyComplete();
+
+    }
+
+    @Test
+    void subtreeChanges_shouldEmitItemOnEachActualChangeAndWhenNodeHasBeenRemoved() {
+        ListenableCbsConfig cut = new ListenableCbsConfig();
+
+        final ReplayProcessor<String> actualChanges = ReplayProcessor.create();
+        final List<String> expectedChanges = List.of("http://dmaap/topic1", "http://dmaap/topic1-updated", "[None]");
+
+        cut.subtreeChanges(List.of("streams", "publishes"))
+                .map(subtreeOption ->
+                        subtreeOption.flatMap(subtree -> subtree.get("topic1", "dmaap-url")).getOrElse("[None]")
+                )
+                .subscribe(actualChanges);
+
+        final MerkleTree<String> initialConfig = MerkleTree.<String>emptyWithDefaultDigest(String::getBytes)
+                .add(List.of("collector", "treshold"), "145")
+                .add(List.of("collector", "listenPort"), "8080");
+
+        final MerkleTree<String> updatedConfig1 = initialConfig
+                .add(List.of("streams", "publishes", "topic1", "type"), "message-bus")
+                .add(List.of("streams", "publishes", "topic1", "dmaap-url"), "http://dmaap/topic1");
+
+        final MerkleTree<String> updatedConfig2 = updatedConfig1
+                .add(List.of("streams", "publishes", "topic1", "type"), "message-bus")
+                .add(List.of("streams", "publishes", "topic1", "dmaap-url"), "http://dmaap/topic1-updated");
+
+        final MerkleTree<String> updatedConfig3 = updatedConfig2
+                .add(List.of("collector", "treshold"), "1410");
+
+        final MerkleTree<String> updatedConfig4 = initialConfig;
+
+        cut.subscribeForUpdates(Flux.just(initialConfig, updatedConfig1, updatedConfig2, updatedConfig3, updatedConfig4));
+
+        StepVerifier.create(actualChanges.take(expectedChanges.size()))
+                .expectNextSequence(expectedChanges)
+                .verifyComplete();
+
+    }
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/MerkleTreeTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/MerkleTreeTest.java
new file mode 100644
index 00000000..2a9a7fcd
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/MerkleTreeTest.java
@@ -0,0 +1,97 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 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.cbs.client.api.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.listener.MerkleTree;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since February 2019
+ */
+class MerkleTreeTest {
+
+    @Test
+    void shouldBeAbleToGetEntries() {
+        MerkleTree<String> cut = emptyTree()
+                .add("value1", "ala", "ma", "kota")
+                .add("value2", "ala", "ma", "psa");
+
+        assertThat(cut.get("ala", "ma", "kota")).contains("value1");
+        assertThat(cut.get("ala", "ma", "psa")).contains("value2");
+    }
+
+    @Test
+    void shouldReturnNoneForNonExistingPaths() {
+        MerkleTree<String> cut = emptyTree()
+                .add("value1", "ala", "ma", "kota")
+                .add("value2", "ala", "ma", "psa");
+
+        assertThat(cut.get("ala", "je", "obiad")).isEmpty();
+    }
+
+    @Test
+    void shouldReturnNoneWhenNodeDoesntContainValue() {
+        MerkleTree<String> cut = emptyTree()
+                .add("value1", "ala", "ma", "kota")
+                .add("value2", "ala", "ma", "psa");
+
+        assertThat(cut.get("ala", "ma")).isEmpty();
+    }
+
+
+    @Test
+    void shouldNotCreateNewObjectWhenNothingChanged() {
+        MerkleTree<String> cut = emptyTree()
+                .add("some value", "ala", "ma", "kota");
+
+        final MerkleTree<String> result = cut.add("some value", "ala", "ma", "kota");
+
+        assertThat(result).isSameAs(cut);
+    }
+
+    @Test
+    void shouldRecalculateHashesAfterAddingNewNode() {
+        MerkleTree<String> cut = emptyTree()
+                .add("value1", "ala", "ma", "kota")
+                .add("value2", "ala", "ma", "psa")
+                .add("value3", "ala", "name");
+
+        final MerkleTree<String> modified = cut.add("value4", "ala", "surname");
+
+        assertThat(modified).isNotSameAs(cut);
+
+        assertThat(modified.hashOf("ala", "ma")).isEqualTo(cut.hashOf("ala", "ma"));
+        assertThat(modified.hashOf("ala", "ma", "kota")).isEqualTo(cut.hashOf("ala", "ma", "kota"));
+        assertThat(modified.hashOf("ala", "ma", "psa")).isEqualTo(cut.hashOf("ala", "ma", "psa"));
+        assertThat(modified.hashOf("ala", "name")).isEqualTo(cut.hashOf("ala", "name"));
+
+        assertThat(modified.hashOf("ala", "surname")).isNotEqualTo(cut.hashOf("ala", "surname"));
+        assertThat(modified.hashOf("ala")).isNotEqualTo(cut.hashOf("ala"));
+        assertThat(modified.hash()).isNotEqualTo(cut.hash());
+    }
+
+    private MerkleTree<String> emptyTree() {
+        return MerkleTree.emptyWithDefaultDigest(String::getBytes);
+    }
+}
\ No newline at end of file