From 5020b150b594e4bb972a136156f744398eff8fd1 Mon Sep 17 00:00:00 2001 From: Piotr Jaszczyk Date: Thu, 14 Feb 2019 13:06:33 +0100 Subject: 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 --- rest-services/cbs-client/pom.xml | 19 +- .../cbs/client/api/listener/HashAlgorithm.java | 39 +++ .../client/api/listener/ListenableCbsConfig.java | 107 +++++++ .../cbs/client/api/listener/MerkleTree.java | 355 +++++++++++++++++++++ .../client/api/listener/TreeChangeListener.java | 54 ++++ .../cbs/client/api/listener/ValueSerializer.java | 32 ++ .../api/listener/ListenableCbsConfigTest.java | 138 ++++++++ .../cbs/client/api/listener/MerkleTreeTest.java | 97 ++++++ 8 files changed, 836 insertions(+), 5 deletions(-) create mode 100644 rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/HashAlgorithm.java create mode 100644 rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ListenableCbsConfig.java create mode 100644 rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/MerkleTree.java create mode 100644 rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/TreeChangeListener.java create mode 100644 rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ValueSerializer.java create mode 100644 rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ListenableCbsConfigTest.java create mode 100644 rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/MerkleTreeTest.java (limited to 'rest-services/cbs-client') diff --git a/rest-services/cbs-client/pom.xml b/rest-services/cbs-client/pom.xml index 2580762d..a3424b63 100644 --- a/rest-services/cbs-client/pom.xml +++ b/rest-services/cbs-client/pom.xml @@ -36,6 +36,11 @@ org.immutables gson + + io.vavr + vavr + + io.projectreactor reactor-core @@ -62,11 +67,15 @@ mockito-core test - - org.junit.jupiter - junit-jupiter-engine - test - + + org.junit.jupiter + junit-jupiter-engine + test + + + org.assertj + assertj-core + io.projectreactor reactor-test diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/HashAlgorithm.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/HashAlgorithm.java new file mode 100644 index 00000000..7b47b127 --- /dev/null +++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/HashAlgorithm.java @@ -0,0 +1,39 @@ +/* + * ============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 io.vavr.Function1; +import io.vavr.collection.List; + +/** + * @author Piotr Jaszczyk + * @since 1.1.2 + */ +@FunctionalInterface +public interface HashAlgorithm extends Function1 { + + @Override + default byte[] apply(byte[] bytes) { + return apply(List.of(bytes)); + } + + byte[] apply(Iterable bytes); +} diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ListenableCbsConfig.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ListenableCbsConfig.java new file mode 100644 index 00000000..b27c718e --- /dev/null +++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ListenableCbsConfig.java @@ -0,0 +1,107 @@ +/* + * ============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 io.vavr.collection.List; +import io.vavr.control.Option; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import reactor.core.Disposable; +import reactor.core.publisher.Flux; + +/** + * @author Piotr Jaszczyk + * @since 1.1.2 + */ +public class ListenableCbsConfig { + + private MerkleTree tree = MerkleTree.emptyWithDefaultDigest(String::getBytes); + private final Map, CompositeTreeChangeListener> pathListeners = new HashMap<>(); + private final Object listenersUpdateMonitor = new Object(); + private final Object treeUpdateMonitor = new Object(); + + public void listen(List path, TreeChangeListener listener) { + synchronized (listenersUpdateMonitor) { + CompositeTreeChangeListener compositeListener = pathListeners + .computeIfAbsent(path, p -> new CompositeTreeChangeListener<>()); + compositeListener.addListener(listener); + } + } + + public void cancel(List path, TreeChangeListener listener) { + synchronized (listenersUpdateMonitor) { + CompositeTreeChangeListener compositeListener = pathListeners.get(path); + if (compositeListener != null) { + compositeListener.removeListener(listener); + } + } + } + + public Flux>> subtreeChanges(List path) { + return Flux.create(sink -> { + final TreeChangeListener listener = sink::next; + sink.onDispose(() -> cancel(path, listener)); + listen(path, listener); + }); + } + + public Disposable subscribeForUpdates(Flux> updates) { + return updates.subscribe(this::update); + } + + public void update(MerkleTree newTree) { + final MerkleTree oldTree; + synchronized (treeUpdateMonitor) { + oldTree = tree; + tree = newTree; + } + + for (Map.Entry, CompositeTreeChangeListener> entry : pathListeners.entrySet()) { + final List path = entry.getKey(); + final CompositeTreeChangeListener listeners = entry.getValue(); + if (!newTree.isSame(path, oldTree)) { + listeners.accept(newTree, path); + } + } + } + + private static class CompositeTreeChangeListener implements TreeChangeListener { + + private final Collection> listeners = new HashSet<>(); + + void addListener(TreeChangeListener listener) { + listeners.add(listener); + } + + void removeListener(TreeChangeListener listener) { + listeners.remove(listener); + } + + @Override + public void accept(Option> updatedSubtree) { + for (TreeChangeListener listener : listeners) { + listener.accept(updatedSubtree); + } + } + } +} diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/MerkleTree.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/MerkleTree.java new file mode 100644 index 00000000..7f24b36e --- /dev/null +++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/MerkleTree.java @@ -0,0 +1,355 @@ +/* + * ============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 io.vavr.Function1; +import io.vavr.collection.HashMap; +import io.vavr.collection.List; +import io.vavr.collection.Map; +import io.vavr.control.Option; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.Arrays; +import java.util.Objects; +import org.jetbrains.annotations.NotNull; + +/** + * An immutable Merkle Tree implementation. + * + * Each node is labelled with a {@code String} label. A path of a node is defined as a list of labels from root to a + * given node. + * + * @author Piotr Jaszczyk + * @since 1.1.2 + */ +public final class MerkleTree { + + private static final String DEFAULT_DIGEST_ALGORITHM = "SHA-256"; + private final ValueSerializer valueSerializer; + private final HashAlgorithm hashAlgorithm; + private final MTNode root; + private final Function1 hashForValue; + + private MerkleTree( + @NotNull ValueSerializer valueSerializer, + @NotNull HashAlgorithm hashAlgorithm, + @NotNull MTNode root) { + this.valueSerializer = Objects.requireNonNull(valueSerializer); + this.hashAlgorithm = Objects.requireNonNull(hashAlgorithm); + this.root = Objects.requireNonNull(root); + hashForValue = valueSerializer.andThen(hashAlgorithm); + } + + /** + * Create an empty tree with given serializer and using default digest algorithm as a hash function. + * + * @param serializer a way of serializing a value to array of bytes + * @param type of values kept in a tree + * @return empty tree + */ + public static @NotNull MerkleTree emptyWithDefaultDigest(@NotNull ValueSerializer serializer) { + return emptyWithDigest(DEFAULT_DIGEST_ALGORITHM, serializer); + } + + /** + * Create an empty tree with given serializer and given digest algorithm used as a hash function. + * + * @param digestAlgorithmName name of a digest algorithm as used by {@link MessageDigest#getInstance(String)} + * @param serializer a way of serializing a value to array of bytes + * @param type of values kept in a tree + * @return empty tree + */ + public static @NotNull MerkleTree emptyWithDigest( + @NotNull String digestAlgorithmName, + @NotNull ValueSerializer serializer) { + return emptyWithHashProvider(serializer, messages -> { + final MessageDigest messageDigest = messageDigest(digestAlgorithmName); + messages.forEach(messageDigest::update); + return messageDigest.digest(); + }); + } + + /** + * Create an empty tree with given hash function. + * + * @param serializer a function which serializes values to a byte array + * @param hashAlgorithm a function which calculates a hash of a serialized value + * @param type of values kept in a tree + * @return empty tree + */ + public static MerkleTree emptyWithHashProvider(ValueSerializer serializer, HashAlgorithm hashAlgorithm) { + return new MerkleTree<>(serializer, hashAlgorithm, MTNode.empty(hashAlgorithm)); + } + + private static MessageDigest messageDigest(String digestAlgorithmName) { + try { + return MessageDigest.getInstance(digestAlgorithmName); + } catch (NoSuchAlgorithmException e) { + throw new IllegalArgumentException("Unsupported hash algorithm " + digestAlgorithmName, e); + } + } + + /** + * Assigns a value to a given path. + * + * Overrides current value if already exists. + * + * @param value a value to assign + * @param path path of labels from root + * @return an updated tree instance or this if hashes are the same + */ + public MerkleTree add(V value, String... path) { + return add(List.of(path), value); + } + + /** + * Assigns a value to a given path. + * + * Overrides current value if already exists. + * + * @param path path of labels from root + * @param value a value to assign + * @return an updated tree instance or this if hashes are the same + */ + public MerkleTree add(List path, V value) { + final MTNode result = root.addChild(path, MTNode.leaf(hashAlgorithm, hashForValue.apply(value), value)); + return Arrays.equals(result.hash(), root.hash()) + ? this + : new MerkleTree<>(valueSerializer, hashAlgorithm, result); + } + + + /** + * Gets a value assigned to a given path. + * + * @param path to search for + * @return Some(value) if path exists and contains a value, None otherwise + */ + public Option get(String... path) { + return get(List.of(path)); + } + + /** + * Gets a value assigned to a given path. + * + * @param path to search for + * @return Some(value) if path exists and contains a value, None otherwise + */ + public Option get(List path) { + return root.findNode(path).flatMap(MTNode::value); + } + + /** + * Checks if nodes under given path are the same in {@code this} and {@code other} tree. + * + * @param other a tree to compare with + * @param path a path to a subtree to compare + * @return true if hashes are the same, false otherwise + */ + public boolean isSame(MerkleTree other, String... path) { + return isSame(List.of(path), other); + } + + /** + * Checks if nodes under given path are the same in {@code this} and {@code other} tree. + * + * @param other a tree to compare with + * @param path a path to a subtree to compare + * @return true if hashes are the same, false otherwise + */ + public boolean isSame(List path, MerkleTree other) { + final byte[] oldHash = other.hashOf(path); + final byte[] curHash = hashOf(path); + return Arrays.equals(oldHash, curHash); + } + + /** + * Returns a hash of a node under given path. + * + * @param path a path of a node to check + * @return a hash or empty array if node does not exist + */ + public byte[] hashOf(List path) { + return root + .findNode(path) + .map(node -> node.hash().clone()) + .getOrElse(() -> new byte[0]); + } + + /** + * Returns a hash of a node under given path. + * + * @param path a path of a node to check + * @return a hash or empty array if node does not exist + */ + public byte[] hashOf(String... path) { + return hashOf(List.of(path)); + } + + /** + * Returns a subtree with given node as a root. + * + * @param path a path of a node to be a subtree root + * @return Some(subtree) if path exists, None otherwise + */ + public Option> subtree(List path) { + return root.findNode(path).map(node -> new MerkleTree<>(valueSerializer, hashAlgorithm, node)); + } + + /** + * Returns a subtree with given node as a root. + * + * @param path a path of a node to be a subtree root + * @return Some(subtree) if path exists, None otherwise + */ + public Option> subtree(String... path) { + return subtree(List.of(path)); + } + + /** + * Hash of a root node. + * + * @return a copy of root node's hash + */ + public byte[] hash() { + return root.hash().clone(); + } + + @Override + public String toString() { + return root.toString(); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + MerkleTree that = (MerkleTree) o; + return Objects.equals(root, that.root); + } + + @Override + public int hashCode() { + return Objects.hash(root); + } +} + +final class MTNode { + + private final byte[] hash; + private final Option value; + private final Map> children; + private final HashAlgorithm hashAlgorithm; + + static MTNode empty(HashAlgorithm hashAlgorithm) { + return new MTNode<>(hashAlgorithm, new byte[0], Option.none(), HashMap.empty()); + } + + static MTNode leaf(HashAlgorithm hashAlgorithm, byte[] hash, V value) { + return new MTNode<>(hashAlgorithm, hash, Option.of(value), HashMap.empty()); + } + + private MTNode( + HashAlgorithm hashAlgorithm, + byte[] hash, + Option value, + Map> children) { + this.hashAlgorithm = hashAlgorithm; + this.hash = hash.clone(); + this.value = value; + this.children = children; + } + + MTNode addChild(final List path, final MTNode child) { + if (path.isEmpty()) { + return child; + } else { + String label = path.head(); + MTNode newChild = children.get(label).fold( + () -> MTNode.empty(hashAlgorithm).addChild(path.tail(), child), + node -> node.addChild(path.tail(), child) + ); + return addChild(label, newChild); + } + } + + Option value() { + return value; + } + + Option> findNode(List path) { + return path.headOption().fold( + () -> Option.of(this), + head -> children.get(head).flatMap(child -> child.findNode(path.tail())) + ); + } + + byte[] hash() { + return hash; + } + + private MTNode addChild(String label, MTNode child) { + final Map> newChildren = children.put(label, child); + byte[] newHash = composeHashes(newChildren.iterator(this::hashForChild)); + return Arrays.equals(newHash, hash) ? this : new MTNode<>( + hashAlgorithm, + newHash, + value, + newChildren + ); + } + + private byte[] hashForChild(String label, MTNode child) { + return composeHashes(List.of(label.getBytes(), child.hash())); + } + + private byte[] composeHashes(Iterable hashes) { + return hashAlgorithm.apply(hashes); + } + + @Override + public String toString() { + return "(\"" + value.map(Object::toString).getOrElse("") + "\" [" + + children.map(entry -> entry._1 + "=" + entry._2).mkString(", ") + + "])"; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + MTNode mtNode = (MTNode) o; + return Arrays.equals(hash, mtNode.hash); + } + + @Override + public int hashCode() { + return Arrays.hashCode(hash); + } +} diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/TreeChangeListener.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/TreeChangeListener.java new file mode 100644 index 00000000..0368c8ad --- /dev/null +++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/TreeChangeListener.java @@ -0,0 +1,54 @@ +/* + * ============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 io.vavr.collection.List; +import io.vavr.control.Option; +import java.util.function.Consumer; + +/** + * The listener for changes of the {@link MerkleTree} subtree. + * + * @author Piotr Jaszczyk + * @since 1.1.2 + */ +@FunctionalInterface +public interface TreeChangeListener extends Consumer>> { + + /** + * Will be called when a change in a subtree has been detected. Default implementation will extract the changed subtree + * from the root and call {@link #accept(Option)}. + * + * @param updatedTree new, updated root tree + * @param path a changed path + */ + default void accept(MerkleTree updatedTree, List path) { + accept(updatedTree.subtree(path)); + } + + /** + * Will be called when a change in a subtree has been detected. + * + * @param updatedSubtree new, updated subtree or None when branch has been removed + */ + @Override + void accept(Option> updatedSubtree); +} diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ValueSerializer.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ValueSerializer.java new file mode 100644 index 00000000..9ef9fe63 --- /dev/null +++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/listener/ValueSerializer.java @@ -0,0 +1,32 @@ +/* + * ============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 io.vavr.Function1; + +/** + * @author Piotr Jaszczyk + * @since 1.1.2 + */ +@FunctionalInterface +public interface ValueSerializer extends Function1 { + +} 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 Piotr Jaszczyk + * @since February 2019 + */ +class ListenableCbsConfigTest { + + @Test + void listen_shouldCallListenerAfterEachChange() { + ListenableCbsConfig cut = new ListenableCbsConfig(); + + final List expectedChanges = List.of("1", "2", "3"); + final AtomicReference> 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 initialConfig = MerkleTree.emptyWithDefaultDigest(String::getBytes) + .add(List.of("some-key"), "1"); + + final MerkleTree updatedConfig1 = initialConfig + .add(List.of("some-key"), "2"); + + final MerkleTree 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 replayProcessor = ReplayProcessor.create(); + final List 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 initialConfig = MerkleTree.emptyWithDefaultDigest(String::getBytes) + .add(List.of("some-key"), "1"); + + final MerkleTree updatedConfig1 = initialConfig + .add(List.of("some-key"), "2"); + + final MerkleTree 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 actualChanges = ReplayProcessor.create(); + final List 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 initialConfig = MerkleTree.emptyWithDefaultDigest(String::getBytes) + .add(List.of("collector", "treshold"), "145") + .add(List.of("collector", "listenPort"), "8080"); + + final MerkleTree updatedConfig1 = initialConfig + .add(List.of("streams", "publishes", "topic1", "type"), "message-bus") + .add(List.of("streams", "publishes", "topic1", "dmaap-url"), "http://dmaap/topic1"); + + final MerkleTree 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 updatedConfig3 = updatedConfig2 + .add(List.of("collector", "treshold"), "1410"); + + final MerkleTree 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 Piotr Jaszczyk + * @since February 2019 + */ +class MerkleTreeTest { + + @Test + void shouldBeAbleToGetEntries() { + MerkleTree 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 cut = emptyTree() + .add("value1", "ala", "ma", "kota") + .add("value2", "ala", "ma", "psa"); + + assertThat(cut.get("ala", "je", "obiad")).isEmpty(); + } + + @Test + void shouldReturnNoneWhenNodeDoesntContainValue() { + MerkleTree cut = emptyTree() + .add("value1", "ala", "ma", "kota") + .add("value2", "ala", "ma", "psa"); + + assertThat(cut.get("ala", "ma")).isEmpty(); + } + + + @Test + void shouldNotCreateNewObjectWhenNothingChanged() { + MerkleTree cut = emptyTree() + .add("some value", "ala", "ma", "kota"); + + final MerkleTree result = cut.add("some value", "ala", "ma", "kota"); + + assertThat(result).isSameAs(cut); + } + + @Test + void shouldRecalculateHashesAfterAddingNewNode() { + MerkleTree cut = emptyTree() + .add("value1", "ala", "ma", "kota") + .add("value2", "ala", "ma", "psa") + .add("value3", "ala", "name"); + + final MerkleTree 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 emptyTree() { + return MerkleTree.emptyWithDefaultDigest(String::getBytes); + } +} \ No newline at end of file -- cgit 1.2.3-korg