Committed by
Gerrit Code Review
Supporting Atomix classes for DocumentTree distributed primitive
Change-Id: I754222337401f90f976d4152b6abbdf2e1a4df8e
Showing
12 changed files
with
1449 additions
and
16 deletions
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.store.service; | ||
18 | + | ||
19 | +import java.util.Map; | ||
20 | +import java.util.concurrent.CompletableFuture; | ||
21 | + | ||
22 | +import javax.annotation.concurrent.NotThreadSafe; | ||
23 | + | ||
24 | +/** | ||
25 | + * A hierarchical <a href="https://en.wikipedia.org/wiki/Document_Object_Model">document tree</a> data structure. | ||
26 | + * | ||
27 | + * @param <V> document tree value type | ||
28 | + */ | ||
29 | +@NotThreadSafe | ||
30 | +public interface AsyncDocumentTree<V> extends DistributedPrimitive { | ||
31 | + | ||
32 | + /** | ||
33 | + * Returns the {@link DocumentPath path} to root of the tree. | ||
34 | + * | ||
35 | + * @return path to root of the tree | ||
36 | + */ | ||
37 | + DocumentPath root(); | ||
38 | + | ||
39 | + /** | ||
40 | + * Returns the child values for this node. | ||
41 | + * | ||
42 | + * @param path path to the node | ||
43 | + * @return future for mapping from a child name to its value | ||
44 | + * @throws NoSuchDocumentPathException if the path does not point to a valid node | ||
45 | + */ | ||
46 | + CompletableFuture<Map<String, Versioned<V>>> getChildren(DocumentPath path); | ||
47 | + | ||
48 | + /** | ||
49 | + * Returns a value (and version) of the tree node at specified path. | ||
50 | + * | ||
51 | + * @param path path to node | ||
52 | + * @return future for node value or {@code null} if path does not point to a valid node | ||
53 | + */ | ||
54 | + CompletableFuture<Versioned<V>> get(DocumentPath path); | ||
55 | + | ||
56 | + /** | ||
57 | + * Creates or updates a document tree node. | ||
58 | + * | ||
59 | + * @param path path for the node to create or update | ||
60 | + * @param value the non-null value to be associated with the key | ||
61 | + * @return future for the previous mapping or {@code null} if there was no previous mapping. Future will | ||
62 | + * be completed with a NoSuchDocumentPathException if the parent node (for the node to create/update) does not exist | ||
63 | + */ | ||
64 | + CompletableFuture<Versioned<V>> set(DocumentPath path, V value); | ||
65 | + | ||
66 | + /** | ||
67 | + * Creates a document tree node if one does not exist already. | ||
68 | + * | ||
69 | + * @param path path for the node to create | ||
70 | + * @param value the non-null value to be associated with the key | ||
71 | + * @return future that is completed with {@code true} if the mapping could be added | ||
72 | + * successfully; {@code false} otherwise. Future will be completed with a | ||
73 | + * IllegalDocumentModificationException if the parent node (for the node to create) does not exist | ||
74 | + */ | ||
75 | + CompletableFuture<Boolean> create(DocumentPath path, V value); | ||
76 | + | ||
77 | + /** | ||
78 | + * Conditionally updates a tree node if the current version matches a specified version. | ||
79 | + * | ||
80 | + * @param path path for the node to create | ||
81 | + * @param newValue the non-null value to be associated with the key | ||
82 | + * @param version current version of the value for update to occur | ||
83 | + * @return future that is completed with {@code true} if the update was made and the tree was | ||
84 | + * modified, {@code false} otherwise. | ||
85 | + */ | ||
86 | + CompletableFuture<Boolean> replace(DocumentPath path, V newValue, long version); | ||
87 | + | ||
88 | + /** | ||
89 | + * Conditionally updates a tree node if the current value matches a specified value. | ||
90 | + * | ||
91 | + * @param path path for the node to create | ||
92 | + * @param newValue the non-null value to be associated with the key | ||
93 | + * @param currentValue current value for update to occur | ||
94 | + * @return future that is completed with {@code true} if the update was made and the tree was | ||
95 | + * modified, {@code false} otherwise. | ||
96 | + */ | ||
97 | + CompletableFuture<Boolean> replace(DocumentPath path, V newValue, V currentValue); | ||
98 | + | ||
99 | + /** | ||
100 | + * Removes the node with the specified path. | ||
101 | + * | ||
102 | + * @param path path for the node to remove | ||
103 | + * @return future for the previous value. Future will be completed with a | ||
104 | + * IllegalDocumentModificationException if the node to be removed is either the root | ||
105 | + * node or has one or more children. Future will be completed with a | ||
106 | + * NoSuchDocumentPathException if the node to be removed does not exist | ||
107 | + */ | ||
108 | + CompletableFuture<Versioned<V>> removeNode(DocumentPath path); | ||
109 | + | ||
110 | + /** | ||
111 | + * Registers a listener to be notified when a subtree rooted at the specified path | ||
112 | + * is modified. | ||
113 | + * | ||
114 | + * @param path path to root of subtree to monitor for updates | ||
115 | + * @param listener listener to be notified | ||
116 | + */ | ||
117 | + CompletableFuture<Void> addListener(DocumentPath path, DocumentTreeListener<V> listener); | ||
118 | + | ||
119 | + /** | ||
120 | + * Unregisters a previously added listener. | ||
121 | + * | ||
122 | + * @param listener listener to unregister | ||
123 | + */ | ||
124 | + CompletableFuture<Void> removeListener(DocumentTreeListener<V> listener); | ||
125 | + | ||
126 | + /** | ||
127 | + * Registers a listener to be notified when the tree is modified. | ||
128 | + * | ||
129 | + * @param listener listener to be notified | ||
130 | + */ | ||
131 | + default CompletableFuture<Void> addListener(DocumentTreeListener<V> listener) { | ||
132 | + return addListener(root(), listener); | ||
133 | + } | ||
134 | +} |
... | @@ -72,6 +72,11 @@ public interface DistributedPrimitive { | ... | @@ -72,6 +72,11 @@ public interface DistributedPrimitive { |
72 | WORK_QUEUE, | 72 | WORK_QUEUE, |
73 | 73 | ||
74 | /** | 74 | /** |
75 | + * Document tree. | ||
76 | + */ | ||
77 | + DOCUMENT_TREE, | ||
78 | + | ||
79 | + /** | ||
75 | * Distributed topic. | 80 | * Distributed topic. |
76 | */ | 81 | */ |
77 | TOPIC, | 82 | TOPIC, | ... | ... |
... | @@ -52,6 +52,14 @@ public class DocumentTreeEvent<V> { | ... | @@ -52,6 +52,14 @@ public class DocumentTreeEvent<V> { |
52 | private final Optional<Versioned<V>> newValue; | 52 | private final Optional<Versioned<V>> newValue; |
53 | private final Optional<Versioned<V>> oldValue; | 53 | private final Optional<Versioned<V>> oldValue; |
54 | 54 | ||
55 | + @SuppressWarnings("unused") | ||
56 | + private DocumentTreeEvent() { | ||
57 | + this.path = null; | ||
58 | + this.type = null; | ||
59 | + this.newValue = null; | ||
60 | + this.oldValue = null; | ||
61 | + } | ||
62 | + | ||
55 | /** | 63 | /** |
56 | * Constructs a new {@code DocumentTreeEvent}. | 64 | * Constructs a new {@code DocumentTreeEvent}. |
57 | * | 65 | * | ... | ... |
... | @@ -15,12 +15,13 @@ | ... | @@ -15,12 +15,13 @@ |
15 | */ | 15 | */ |
16 | package org.onosproject.store.primitives.impl; | 16 | package org.onosproject.store.primitives.impl; |
17 | 17 | ||
18 | -import com.google.common.collect.ImmutableList; | ||
19 | -import com.google.common.collect.Maps; | ||
20 | import io.atomix.catalyst.serializer.Serializer; | 18 | import io.atomix.catalyst.serializer.Serializer; |
21 | import io.atomix.catalyst.serializer.TypeSerializerFactory; | 19 | import io.atomix.catalyst.serializer.TypeSerializerFactory; |
22 | import io.atomix.manager.util.ResourceManagerTypeResolver; | 20 | import io.atomix.manager.util.ResourceManagerTypeResolver; |
23 | import io.atomix.variables.internal.LongCommands; | 21 | import io.atomix.variables.internal.LongCommands; |
22 | + | ||
23 | +import java.util.Arrays; | ||
24 | + | ||
24 | import org.onlab.util.Match; | 25 | import org.onlab.util.Match; |
25 | import org.onosproject.cluster.Leader; | 26 | import org.onosproject.cluster.Leader; |
26 | import org.onosproject.cluster.Leadership; | 27 | import org.onosproject.cluster.Leadership; |
... | @@ -34,22 +35,28 @@ import org.onosproject.store.primitives.resources.impl.AtomixConsistentMultimapC | ... | @@ -34,22 +35,28 @@ import org.onosproject.store.primitives.resources.impl.AtomixConsistentMultimapC |
34 | import org.onosproject.store.primitives.resources.impl.AtomixConsistentSetMultimapFactory; | 35 | import org.onosproject.store.primitives.resources.impl.AtomixConsistentSetMultimapFactory; |
35 | import org.onosproject.store.primitives.resources.impl.AtomixConsistentTreeMapCommands; | 36 | import org.onosproject.store.primitives.resources.impl.AtomixConsistentTreeMapCommands; |
36 | import org.onosproject.store.primitives.resources.impl.AtomixConsistentTreeMapFactory; | 37 | import org.onosproject.store.primitives.resources.impl.AtomixConsistentTreeMapFactory; |
38 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands; | ||
39 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeFactory; | ||
37 | import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands; | 40 | import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands; |
38 | import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorFactory; | 41 | import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorFactory; |
39 | import org.onosproject.store.primitives.resources.impl.AtomixWorkQueueCommands; | 42 | import org.onosproject.store.primitives.resources.impl.AtomixWorkQueueCommands; |
40 | import org.onosproject.store.primitives.resources.impl.AtomixWorkQueueFactory; | 43 | import org.onosproject.store.primitives.resources.impl.AtomixWorkQueueFactory; |
41 | import org.onosproject.store.primitives.resources.impl.CommitResult; | 44 | import org.onosproject.store.primitives.resources.impl.CommitResult; |
45 | +import org.onosproject.store.primitives.resources.impl.DocumentTreeUpdateResult; | ||
42 | import org.onosproject.store.primitives.resources.impl.MapEntryUpdateResult; | 46 | import org.onosproject.store.primitives.resources.impl.MapEntryUpdateResult; |
43 | import org.onosproject.store.primitives.resources.impl.PrepareResult; | 47 | import org.onosproject.store.primitives.resources.impl.PrepareResult; |
44 | import org.onosproject.store.primitives.resources.impl.RollbackResult; | 48 | import org.onosproject.store.primitives.resources.impl.RollbackResult; |
45 | import org.onosproject.store.serializers.KryoNamespaces; | 49 | import org.onosproject.store.serializers.KryoNamespaces; |
50 | +import org.onosproject.store.service.DocumentPath; | ||
51 | +import org.onosproject.store.service.DocumentTreeEvent; | ||
46 | import org.onosproject.store.service.MapEvent; | 52 | import org.onosproject.store.service.MapEvent; |
47 | import org.onosproject.store.service.MapTransaction; | 53 | import org.onosproject.store.service.MapTransaction; |
48 | import org.onosproject.store.service.Task; | 54 | import org.onosproject.store.service.Task; |
49 | import org.onosproject.store.service.Versioned; | 55 | import org.onosproject.store.service.Versioned; |
50 | import org.onosproject.store.service.WorkQueueStats; | 56 | import org.onosproject.store.service.WorkQueueStats; |
51 | 57 | ||
52 | -import java.util.Arrays; | 58 | +import com.google.common.collect.ImmutableList; |
59 | +import com.google.common.collect.Maps; | ||
53 | 60 | ||
54 | /** | 61 | /** |
55 | * Serializer utility for Atomix Catalyst. | 62 | * Serializer utility for Atomix Catalyst. |
... | @@ -69,6 +76,11 @@ public final class CatalystSerializers { | ... | @@ -69,6 +76,11 @@ public final class CatalystSerializers { |
69 | Transaction.State.class, | 76 | Transaction.State.class, |
70 | PrepareResult.class, | 77 | PrepareResult.class, |
71 | CommitResult.class, | 78 | CommitResult.class, |
79 | + DocumentPath.class, | ||
80 | + DocumentTreeUpdateResult.class, | ||
81 | + DocumentTreeUpdateResult.Status.class, | ||
82 | + DocumentTreeEvent.class, | ||
83 | + DocumentTreeEvent.Type.class, | ||
72 | RollbackResult.class)); | 84 | RollbackResult.class)); |
73 | // ONOS classes | 85 | // ONOS classes |
74 | serializer.register(Change.class, factory); | 86 | serializer.register(Change.class, factory); |
... | @@ -90,6 +102,11 @@ public final class CatalystSerializers { | ... | @@ -90,6 +102,11 @@ public final class CatalystSerializers { |
90 | serializer.register(MapEvent.class, factory); | 102 | serializer.register(MapEvent.class, factory); |
91 | serializer.register(Task.class, factory); | 103 | serializer.register(Task.class, factory); |
92 | serializer.register(WorkQueueStats.class, factory); | 104 | serializer.register(WorkQueueStats.class, factory); |
105 | + serializer.register(DocumentPath.class, factory); | ||
106 | + serializer.register(DocumentTreeUpdateResult.class, factory); | ||
107 | + serializer.register(DocumentTreeUpdateResult.Status.class, factory); | ||
108 | + serializer.register(DocumentTreeEvent.class, factory); | ||
109 | + serializer.register(DocumentTreeEvent.Type.class, factory); | ||
93 | serializer.register(Maps.immutableEntry("a", "b").getClass(), factory); | 110 | serializer.register(Maps.immutableEntry("a", "b").getClass(), factory); |
94 | serializer.register(ImmutableList.of().getClass(), factory); | 111 | serializer.register(ImmutableList.of().getClass(), factory); |
95 | 112 | ||
... | @@ -97,6 +114,7 @@ public final class CatalystSerializers { | ... | @@ -97,6 +114,7 @@ public final class CatalystSerializers { |
97 | serializer.resolve(new AtomixConsistentMapCommands.TypeResolver()); | 114 | serializer.resolve(new AtomixConsistentMapCommands.TypeResolver()); |
98 | serializer.resolve(new AtomixLeaderElectorCommands.TypeResolver()); | 115 | serializer.resolve(new AtomixLeaderElectorCommands.TypeResolver()); |
99 | serializer.resolve(new AtomixWorkQueueCommands.TypeResolver()); | 116 | serializer.resolve(new AtomixWorkQueueCommands.TypeResolver()); |
117 | + serializer.resolve(new AtomixDocumentTreeCommands.TypeResolver()); | ||
100 | serializer.resolve(new ResourceManagerTypeResolver()); | 118 | serializer.resolve(new ResourceManagerTypeResolver()); |
101 | serializer.resolve(new AtomixConsistentTreeMapCommands.TypeResolver()); | 119 | serializer.resolve(new AtomixConsistentTreeMapCommands.TypeResolver()); |
102 | serializer.resolve(new AtomixConsistentMultimapCommands.TypeResolver()); | 120 | serializer.resolve(new AtomixConsistentMultimapCommands.TypeResolver()); |
... | @@ -104,6 +122,7 @@ public final class CatalystSerializers { | ... | @@ -104,6 +122,7 @@ public final class CatalystSerializers { |
104 | serializer.registerClassLoader(AtomixConsistentMapFactory.class) | 122 | serializer.registerClassLoader(AtomixConsistentMapFactory.class) |
105 | .registerClassLoader(AtomixLeaderElectorFactory.class) | 123 | .registerClassLoader(AtomixLeaderElectorFactory.class) |
106 | .registerClassLoader(AtomixWorkQueueFactory.class) | 124 | .registerClassLoader(AtomixWorkQueueFactory.class) |
125 | + .registerClassLoader(AtomixDocumentTreeFactory.class) | ||
107 | .registerClassLoader(AtomixConsistentTreeMapFactory.class) | 126 | .registerClassLoader(AtomixConsistentTreeMapFactory.class) |
108 | .registerClassLoader(AtomixConsistentSetMultimapFactory.class); | 127 | .registerClassLoader(AtomixConsistentSetMultimapFactory.class); |
109 | 128 | ... | ... |
... | @@ -15,14 +15,15 @@ | ... | @@ -15,14 +15,15 @@ |
15 | */ | 15 | */ |
16 | package org.onosproject.store.primitives.resources.impl; | 16 | package org.onosproject.store.primitives.resources.impl; |
17 | 17 | ||
18 | +import static com.google.common.base.Preconditions.checkState; | ||
18 | import static org.onosproject.store.service.MapEvent.Type.INSERT; | 19 | import static org.onosproject.store.service.MapEvent.Type.INSERT; |
19 | import static org.onosproject.store.service.MapEvent.Type.REMOVE; | 20 | import static org.onosproject.store.service.MapEvent.Type.REMOVE; |
20 | import static org.onosproject.store.service.MapEvent.Type.UPDATE; | 21 | import static org.onosproject.store.service.MapEvent.Type.UPDATE; |
21 | import static org.slf4j.LoggerFactory.getLogger; | 22 | import static org.slf4j.LoggerFactory.getLogger; |
22 | -import io.atomix.copycat.server.session.ServerSession; | ||
23 | import io.atomix.copycat.server.Commit; | 23 | import io.atomix.copycat.server.Commit; |
24 | import io.atomix.copycat.server.Snapshottable; | 24 | import io.atomix.copycat.server.Snapshottable; |
25 | import io.atomix.copycat.server.StateMachineExecutor; | 25 | import io.atomix.copycat.server.StateMachineExecutor; |
26 | +import io.atomix.copycat.server.session.ServerSession; | ||
26 | import io.atomix.copycat.server.session.SessionListener; | 27 | import io.atomix.copycat.server.session.SessionListener; |
27 | import io.atomix.copycat.server.storage.snapshot.SnapshotReader; | 28 | import io.atomix.copycat.server.storage.snapshot.SnapshotReader; |
28 | import io.atomix.copycat.server.storage.snapshot.SnapshotWriter; | 29 | import io.atomix.copycat.server.storage.snapshot.SnapshotWriter; |
... | @@ -68,15 +69,13 @@ import com.google.common.collect.Lists; | ... | @@ -68,15 +69,13 @@ import com.google.common.collect.Lists; |
68 | import com.google.common.collect.Maps; | 69 | import com.google.common.collect.Maps; |
69 | import com.google.common.collect.Sets; | 70 | import com.google.common.collect.Sets; |
70 | 71 | ||
71 | -import static com.google.common.base.Preconditions.checkState; | ||
72 | - | ||
73 | /** | 72 | /** |
74 | * State Machine for {@link AtomixConsistentMap} resource. | 73 | * State Machine for {@link AtomixConsistentMap} resource. |
75 | */ | 74 | */ |
76 | public class AtomixConsistentMapState extends ResourceStateMachine implements SessionListener, Snapshottable { | 75 | public class AtomixConsistentMapState extends ResourceStateMachine implements SessionListener, Snapshottable { |
77 | 76 | ||
78 | private final Logger log = getLogger(getClass()); | 77 | private final Logger log = getLogger(getClass()); |
79 | - private final Map<Long, Commit<? extends AtomixConsistentMapCommands.Listen>> listeners = new HashMap<>(); | 78 | + private final Map<Long, Commit<? extends Listen>> listeners = new HashMap<>(); |
80 | private final Map<String, MapEntryValue> mapEntries = new HashMap<>(); | 79 | private final Map<String, MapEntryValue> mapEntries = new HashMap<>(); |
81 | private final Set<String> preparedKeys = Sets.newHashSet(); | 80 | private final Set<String> preparedKeys = Sets.newHashSet(); |
82 | private final Map<TransactionId, Commit<? extends TransactionPrepare>> pendingTransactions = Maps.newHashMap(); | 81 | private final Map<TransactionId, Commit<? extends TransactionPrepare>> pendingTransactions = Maps.newHashMap(); | ... | ... |
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.store.primitives.resources.impl; | ||
18 | + | ||
19 | +import static com.google.common.base.Preconditions.checkNotNull; | ||
20 | +import io.atomix.copycat.client.CopycatClient; | ||
21 | +import io.atomix.resource.AbstractResource; | ||
22 | +import io.atomix.resource.ResourceTypeInfo; | ||
23 | + | ||
24 | +import java.util.HashMap; | ||
25 | +import java.util.List; | ||
26 | +import java.util.Map; | ||
27 | +import java.util.Properties; | ||
28 | +import java.util.concurrent.CompletableFuture; | ||
29 | +import java.util.concurrent.Executor; | ||
30 | + | ||
31 | +import org.onlab.util.Match; | ||
32 | +import org.onlab.util.Tools; | ||
33 | +import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Unlisten; | ||
34 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Clear; | ||
35 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Get; | ||
36 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.GetChildren; | ||
37 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Listen; | ||
38 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Update; | ||
39 | +import org.onosproject.store.service.AsyncDocumentTree; | ||
40 | +import org.onosproject.store.service.DocumentPath; | ||
41 | +import org.onosproject.store.service.DocumentTreeEvent; | ||
42 | +import org.onosproject.store.service.DocumentTreeListener; | ||
43 | +import org.onosproject.store.service.IllegalDocumentModificationException; | ||
44 | +import org.onosproject.store.service.NoSuchDocumentPathException; | ||
45 | +import org.onosproject.store.service.Versioned; | ||
46 | + | ||
47 | +import com.google.common.util.concurrent.MoreExecutors; | ||
48 | + | ||
49 | +/** | ||
50 | + * Distributed resource providing the {@link AsyncDocumentTree} primitive. | ||
51 | + */ | ||
52 | +@ResourceTypeInfo(id = -156, factory = AtomixDocumentTreeFactory.class) | ||
53 | +public class AtomixDocumentTree extends AbstractResource<AtomixDocumentTree> | ||
54 | + implements AsyncDocumentTree<byte[]> { | ||
55 | + | ||
56 | + private final Map<DocumentTreeListener<byte[]>, Executor> eventListeners = new HashMap<>(); | ||
57 | + public static final String CHANGE_SUBJECT = "changeEvents"; | ||
58 | + | ||
59 | + protected AtomixDocumentTree(CopycatClient client, Properties options) { | ||
60 | + super(client, options); | ||
61 | + } | ||
62 | + | ||
63 | + @Override | ||
64 | + public CompletableFuture<AtomixDocumentTree> open() { | ||
65 | + return super.open().thenApply(result -> { | ||
66 | + client.onStateChange(state -> { | ||
67 | + if (state == CopycatClient.State.CONNECTED && isListening()) { | ||
68 | + client.submit(new Listen()); | ||
69 | + } | ||
70 | + }); | ||
71 | + client.onEvent(CHANGE_SUBJECT, this::processTreeUpdates); | ||
72 | + return result; | ||
73 | + }); | ||
74 | + } | ||
75 | + | ||
76 | + @Override | ||
77 | + public String name() { | ||
78 | + return null; | ||
79 | + } | ||
80 | + | ||
81 | + @Override | ||
82 | + public Type primitiveType() { | ||
83 | + return Type.DOCUMENT_TREE; | ||
84 | + } | ||
85 | + | ||
86 | + @Override | ||
87 | + public CompletableFuture<Void> destroy() { | ||
88 | + return client.submit(new Clear()); | ||
89 | + } | ||
90 | + | ||
91 | + @Override | ||
92 | + public DocumentPath root() { | ||
93 | + return DocumentPath.from("root"); | ||
94 | + } | ||
95 | + | ||
96 | + @Override | ||
97 | + public CompletableFuture<Map<String, Versioned<byte[]>>> getChildren(DocumentPath path) { | ||
98 | + return client.submit(new GetChildren(checkNotNull(path))); | ||
99 | + } | ||
100 | + | ||
101 | + @Override | ||
102 | + public CompletableFuture<Versioned<byte[]>> get(DocumentPath path) { | ||
103 | + return client.submit(new Get(checkNotNull(path))); | ||
104 | + } | ||
105 | + | ||
106 | + @Override | ||
107 | + public CompletableFuture<Versioned<byte[]>> set(DocumentPath path, byte[] value) { | ||
108 | + return client.submit(new Update(checkNotNull(path), checkNotNull(value), Match.any(), Match.any())) | ||
109 | + .thenCompose(result -> { | ||
110 | + if (result.status() == DocumentTreeUpdateResult.Status.INVALID_PATH) { | ||
111 | + return Tools.exceptionalFuture(new NoSuchDocumentPathException()); | ||
112 | + } else if (result.status() == DocumentTreeUpdateResult.Status.ILLEGAL_MODIFICATION) { | ||
113 | + return Tools.exceptionalFuture(new IllegalDocumentModificationException()); | ||
114 | + } else { | ||
115 | + return CompletableFuture.completedFuture(result); | ||
116 | + } | ||
117 | + }).thenApply(result -> result.oldValue()); | ||
118 | + } | ||
119 | + | ||
120 | + @Override | ||
121 | + public CompletableFuture<Boolean> create(DocumentPath path, byte[] value) { | ||
122 | + return client.submit(new Update(checkNotNull(path), checkNotNull(value), Match.ifNull(), Match.any())) | ||
123 | + .thenCompose(result -> { | ||
124 | + if (result.status() == DocumentTreeUpdateResult.Status.INVALID_PATH) { | ||
125 | + return Tools.exceptionalFuture(new NoSuchDocumentPathException()); | ||
126 | + } else if (result.status() == DocumentTreeUpdateResult.Status.ILLEGAL_MODIFICATION) { | ||
127 | + return Tools.exceptionalFuture(new IllegalDocumentModificationException()); | ||
128 | + } else { | ||
129 | + return CompletableFuture.completedFuture(result); | ||
130 | + } | ||
131 | + }).thenApply(result -> result.created()); | ||
132 | + } | ||
133 | + | ||
134 | + @Override | ||
135 | + public CompletableFuture<Boolean> replace(DocumentPath path, byte[] newValue, long version) { | ||
136 | + return client.submit(new Update(checkNotNull(path), newValue, Match.any(), Match.ifValue(version))) | ||
137 | + .thenApply(result -> result.updated()); | ||
138 | + } | ||
139 | + | ||
140 | + @Override | ||
141 | + public CompletableFuture<Boolean> replace(DocumentPath path, byte[] newValue, byte[] currentValue) { | ||
142 | + return client.submit(new Update(checkNotNull(path), newValue, Match.ifValue(currentValue), Match.any())) | ||
143 | + .thenCompose(result -> { | ||
144 | + if (result.status() == DocumentTreeUpdateResult.Status.INVALID_PATH) { | ||
145 | + return Tools.exceptionalFuture(new NoSuchDocumentPathException()); | ||
146 | + } else if (result.status() == DocumentTreeUpdateResult.Status.ILLEGAL_MODIFICATION) { | ||
147 | + return Tools.exceptionalFuture(new IllegalDocumentModificationException()); | ||
148 | + } else { | ||
149 | + return CompletableFuture.completedFuture(result); | ||
150 | + } | ||
151 | + }).thenApply(result -> result.updated()); | ||
152 | + } | ||
153 | + | ||
154 | + @Override | ||
155 | + public CompletableFuture<Versioned<byte[]>> removeNode(DocumentPath path) { | ||
156 | + if (path.equals(DocumentPath.from("root"))) { | ||
157 | + return Tools.exceptionalFuture(new IllegalDocumentModificationException()); | ||
158 | + } | ||
159 | + return client.submit(new Update(checkNotNull(path), null, Match.ifNotNull(), Match.any())) | ||
160 | + .thenCompose(result -> { | ||
161 | + if (result.status() == DocumentTreeUpdateResult.Status.INVALID_PATH) { | ||
162 | + return Tools.exceptionalFuture(new NoSuchDocumentPathException()); | ||
163 | + } else if (result.status() == DocumentTreeUpdateResult.Status.ILLEGAL_MODIFICATION) { | ||
164 | + return Tools.exceptionalFuture(new IllegalDocumentModificationException()); | ||
165 | + } else { | ||
166 | + return CompletableFuture.completedFuture(result); | ||
167 | + } | ||
168 | + }).thenApply(result -> result.oldValue()); | ||
169 | + } | ||
170 | + | ||
171 | + @Override | ||
172 | + public CompletableFuture<Void> addListener(DocumentPath path, DocumentTreeListener<byte[]> listener) { | ||
173 | + checkNotNull(path); | ||
174 | + checkNotNull(listener); | ||
175 | + // TODO: Support API that takes an executor | ||
176 | + if (isListening()) { | ||
177 | + eventListeners.putIfAbsent(listener, MoreExecutors.directExecutor()); | ||
178 | + return CompletableFuture.completedFuture(null); | ||
179 | + } else { | ||
180 | + return client.submit(new Listen(path)) | ||
181 | + .thenRun(() -> eventListeners.put(listener, MoreExecutors.directExecutor())); | ||
182 | + } | ||
183 | + } | ||
184 | + | ||
185 | + @Override | ||
186 | + public CompletableFuture<Void> removeListener(DocumentTreeListener<byte[]> listener) { | ||
187 | + checkNotNull(listener); | ||
188 | + if (eventListeners.remove(listener) != null && eventListeners.isEmpty()) { | ||
189 | + return client.submit(new Unlisten()).thenApply(v -> null); | ||
190 | + } | ||
191 | + return CompletableFuture.completedFuture(null); | ||
192 | + } | ||
193 | + | ||
194 | + private boolean isListening() { | ||
195 | + return !eventListeners.isEmpty(); | ||
196 | + } | ||
197 | + | ||
198 | + private void processTreeUpdates(List<DocumentTreeEvent<byte[]>> events) { | ||
199 | + events.forEach(event -> | ||
200 | + eventListeners.forEach((listener, executor) -> executor.execute(() -> listener.event(event)))); | ||
201 | + } | ||
202 | +} |
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.store.primitives.resources.impl; | ||
18 | + | ||
19 | +import io.atomix.catalyst.buffer.BufferInput; | ||
20 | +import io.atomix.catalyst.buffer.BufferOutput; | ||
21 | +import io.atomix.catalyst.serializer.CatalystSerializable; | ||
22 | +import io.atomix.catalyst.serializer.SerializableTypeResolver; | ||
23 | +import io.atomix.catalyst.serializer.Serializer; | ||
24 | +import io.atomix.catalyst.serializer.SerializerRegistry; | ||
25 | +import io.atomix.copycat.Command; | ||
26 | +import io.atomix.copycat.Query; | ||
27 | + | ||
28 | +import java.util.Map; | ||
29 | + | ||
30 | +import org.onlab.util.Match; | ||
31 | +import org.onosproject.store.service.DocumentPath; | ||
32 | +import org.onosproject.store.service.Versioned; | ||
33 | + | ||
34 | +import com.google.common.base.MoreObjects; | ||
35 | + | ||
36 | +/** | ||
37 | + * {@link AtomixDocumentTree} resource state machine operations. | ||
38 | + */ | ||
39 | +public class AtomixDocumentTreeCommands { | ||
40 | + | ||
41 | + /** | ||
42 | + * Abstract DocumentTree operation. | ||
43 | + */ | ||
44 | + public abstract static class DocumentTreeOperation<V> implements CatalystSerializable { | ||
45 | + | ||
46 | + private DocumentPath path; | ||
47 | + | ||
48 | + DocumentTreeOperation(DocumentPath path) { | ||
49 | + this.path = path; | ||
50 | + } | ||
51 | + | ||
52 | + public DocumentPath path() { | ||
53 | + return path; | ||
54 | + } | ||
55 | + | ||
56 | + @Override | ||
57 | + public void writeObject(BufferOutput<?> buffer, Serializer serializer) { | ||
58 | + serializer.writeObject(path, buffer); | ||
59 | + } | ||
60 | + | ||
61 | + @Override | ||
62 | + public void readObject(BufferInput<?> buffer, Serializer serializer) { | ||
63 | + path = serializer.readObject(buffer); | ||
64 | + } | ||
65 | + } | ||
66 | + | ||
67 | + /** | ||
68 | + * Abstract DocumentTree query. | ||
69 | + */ | ||
70 | + @SuppressWarnings("serial") | ||
71 | + public abstract static class DocumentTreeQuery<V> extends DocumentTreeOperation<V> implements Query<V> { | ||
72 | + | ||
73 | + DocumentTreeQuery(DocumentPath path) { | ||
74 | + super(path); | ||
75 | + } | ||
76 | + | ||
77 | + @Override | ||
78 | + public ConsistencyLevel consistency() { | ||
79 | + return ConsistencyLevel.SEQUENTIAL; | ||
80 | + } | ||
81 | + } | ||
82 | + | ||
83 | + /** | ||
84 | + * Abstract DocumentTree command. | ||
85 | + */ | ||
86 | + @SuppressWarnings("serial") | ||
87 | + public abstract static class DocumentTreeCommand<V> extends DocumentTreeOperation<V> implements Command<V> { | ||
88 | + | ||
89 | + DocumentTreeCommand(DocumentPath path) { | ||
90 | + super(path); | ||
91 | + } | ||
92 | + } | ||
93 | + | ||
94 | + /** | ||
95 | + * DocumentTree#get query. | ||
96 | + */ | ||
97 | + @SuppressWarnings("serial") | ||
98 | + public static class Get extends DocumentTreeQuery<Versioned<byte[]>> { | ||
99 | + public Get() { | ||
100 | + super(null); | ||
101 | + } | ||
102 | + | ||
103 | + public Get(DocumentPath path) { | ||
104 | + super(path); | ||
105 | + } | ||
106 | + | ||
107 | + @Override | ||
108 | + public String toString() { | ||
109 | + return MoreObjects.toStringHelper(getClass()) | ||
110 | + .add("path", path()) | ||
111 | + .toString(); | ||
112 | + } | ||
113 | + } | ||
114 | + | ||
115 | + /** | ||
116 | + * DocumentTree#getChildren query. | ||
117 | + */ | ||
118 | + @SuppressWarnings("serial") | ||
119 | + public static class GetChildren extends DocumentTreeQuery<Map<String, Versioned<byte[]>>> { | ||
120 | + public GetChildren() { | ||
121 | + super(null); | ||
122 | + } | ||
123 | + | ||
124 | + public GetChildren(DocumentPath path) { | ||
125 | + super(path); | ||
126 | + } | ||
127 | + | ||
128 | + @Override | ||
129 | + public String toString() { | ||
130 | + return MoreObjects.toStringHelper(getClass()) | ||
131 | + .add("path", path()) | ||
132 | + .toString(); | ||
133 | + } | ||
134 | + } | ||
135 | + | ||
136 | + /** | ||
137 | + * DocumentTree update command. | ||
138 | + */ | ||
139 | + @SuppressWarnings("serial") | ||
140 | + public static class Update extends DocumentTreeCommand<DocumentTreeUpdateResult<byte[]>> { | ||
141 | + | ||
142 | + private byte[] value; | ||
143 | + private Match<byte[]> valueMatch; | ||
144 | + private Match<Long> versionMatch; | ||
145 | + | ||
146 | + public Update() { | ||
147 | + super(null); | ||
148 | + this.value = null; | ||
149 | + this.valueMatch = null; | ||
150 | + this.versionMatch = null; | ||
151 | + } | ||
152 | + | ||
153 | + public Update(DocumentPath path, byte[] value, Match<byte[]> valueMatch, Match<Long> versionMatch) { | ||
154 | + super(path); | ||
155 | + this.value = value; | ||
156 | + this.valueMatch = valueMatch; | ||
157 | + this.versionMatch = versionMatch; | ||
158 | + } | ||
159 | + | ||
160 | + public byte[] value() { | ||
161 | + return value; | ||
162 | + } | ||
163 | + | ||
164 | + public Match<byte[]> valueMatch() { | ||
165 | + return valueMatch; | ||
166 | + } | ||
167 | + | ||
168 | + public Match<Long> versionMatch() { | ||
169 | + return versionMatch; | ||
170 | + } | ||
171 | + | ||
172 | + @Override | ||
173 | + public void writeObject(BufferOutput<?> buffer, Serializer serializer) { | ||
174 | + super.writeObject(buffer, serializer); | ||
175 | + serializer.writeObject(value, buffer); | ||
176 | + serializer.writeObject(valueMatch, buffer); | ||
177 | + serializer.writeObject(versionMatch, buffer); | ||
178 | + } | ||
179 | + | ||
180 | + @Override | ||
181 | + public void readObject(BufferInput<?> buffer, Serializer serializer) { | ||
182 | + super.readObject(buffer, serializer); | ||
183 | + value = serializer.readObject(buffer); | ||
184 | + valueMatch = serializer.readObject(buffer); | ||
185 | + versionMatch = serializer.readObject(buffer); | ||
186 | + } | ||
187 | + | ||
188 | + @Override | ||
189 | + public String toString() { | ||
190 | + return MoreObjects.toStringHelper(getClass()) | ||
191 | + .add("path", path()) | ||
192 | + .add("value", value) | ||
193 | + .add("valueMatch", valueMatch) | ||
194 | + .add("versionMatch", versionMatch) | ||
195 | + .toString(); | ||
196 | + } | ||
197 | + } | ||
198 | + | ||
199 | + /** | ||
200 | + * Clear command. | ||
201 | + */ | ||
202 | + @SuppressWarnings("serial") | ||
203 | + public static class Clear implements Command<Void>, CatalystSerializable { | ||
204 | + @Override | ||
205 | + public void writeObject(BufferOutput<?> buffer, Serializer serializer) { | ||
206 | + } | ||
207 | + | ||
208 | + @Override | ||
209 | + public void readObject(BufferInput<?> buffer, Serializer serializer) { | ||
210 | + } | ||
211 | + } | ||
212 | + | ||
213 | + /** | ||
214 | + * Change listen. | ||
215 | + */ | ||
216 | + @SuppressWarnings("serial") | ||
217 | + public static class Listen extends DocumentTreeCommand<Void> { | ||
218 | + | ||
219 | + public Listen() { | ||
220 | + this(DocumentPath.from("root")); | ||
221 | + } | ||
222 | + | ||
223 | + public Listen(DocumentPath path) { | ||
224 | + super(path); | ||
225 | + } | ||
226 | + | ||
227 | + @Override | ||
228 | + public void writeObject(BufferOutput<?> buffer, Serializer serializer) { | ||
229 | + } | ||
230 | + | ||
231 | + @Override | ||
232 | + public void readObject(BufferInput<?> buffer, Serializer serializer) { | ||
233 | + } | ||
234 | + } | ||
235 | + | ||
236 | + /** | ||
237 | + * Change unlisten. | ||
238 | + */ | ||
239 | + @SuppressWarnings("serial") | ||
240 | + public static class Unlisten extends DocumentTreeCommand<Void> { | ||
241 | + | ||
242 | + public Unlisten() { | ||
243 | + this(DocumentPath.from("root")); | ||
244 | + } | ||
245 | + | ||
246 | + public Unlisten(DocumentPath path) { | ||
247 | + super(path); | ||
248 | + } | ||
249 | + | ||
250 | + @Override | ||
251 | + public void writeObject(BufferOutput<?> buffer, Serializer serializer) { | ||
252 | + } | ||
253 | + | ||
254 | + @Override | ||
255 | + public void readObject(BufferInput<?> buffer, Serializer serializer) { | ||
256 | + } | ||
257 | + } | ||
258 | + | ||
259 | + /** | ||
260 | + * DocumentTree command type resolver. | ||
261 | + */ | ||
262 | + public static class TypeResolver implements SerializableTypeResolver { | ||
263 | + @Override | ||
264 | + public void resolve(SerializerRegistry registry) { | ||
265 | + registry.register(Get.class, -911); | ||
266 | + registry.register(GetChildren.class, -912); | ||
267 | + registry.register(Update.class, -913); | ||
268 | + registry.register(Listen.class, -914); | ||
269 | + registry.register(Unlisten.class, -915); | ||
270 | + registry.register(Clear.class, -916); | ||
271 | + } | ||
272 | + } | ||
273 | +} |
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | +package org.onosproject.store.primitives.resources.impl; | ||
17 | + | ||
18 | +import io.atomix.catalyst.serializer.SerializableTypeResolver; | ||
19 | +import io.atomix.copycat.client.CopycatClient; | ||
20 | +import io.atomix.resource.ResourceFactory; | ||
21 | +import io.atomix.resource.ResourceStateMachine; | ||
22 | + | ||
23 | +import java.util.Properties; | ||
24 | + | ||
25 | +/** | ||
26 | + * {@link AtomixDocumentTree} resource factory. | ||
27 | + * | ||
28 | + */ | ||
29 | +public class AtomixDocumentTreeFactory implements ResourceFactory<AtomixDocumentTree> { | ||
30 | + | ||
31 | + @Override | ||
32 | + public SerializableTypeResolver createSerializableTypeResolver() { | ||
33 | + return new AtomixDocumentTreeCommands.TypeResolver(); | ||
34 | + } | ||
35 | + | ||
36 | + @Override | ||
37 | + public ResourceStateMachine createStateMachine(Properties config) { | ||
38 | + return new AtomixDocumentTreeState(config); | ||
39 | + } | ||
40 | + | ||
41 | + @Override | ||
42 | + public AtomixDocumentTree createInstance(CopycatClient client, Properties options) { | ||
43 | + return new AtomixDocumentTree(client, options); | ||
44 | + } | ||
45 | + } | ||
... | \ No newline at end of file | ... | \ No newline at end of file |
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.store.primitives.resources.impl; | ||
18 | + | ||
19 | +import static org.slf4j.LoggerFactory.getLogger; | ||
20 | +import io.atomix.copycat.server.Commit; | ||
21 | +import io.atomix.copycat.server.Snapshottable; | ||
22 | +import io.atomix.copycat.server.StateMachineExecutor; | ||
23 | +import io.atomix.copycat.server.session.ServerSession; | ||
24 | +import io.atomix.copycat.server.session.SessionListener; | ||
25 | +import io.atomix.copycat.server.storage.snapshot.SnapshotReader; | ||
26 | +import io.atomix.copycat.server.storage.snapshot.SnapshotWriter; | ||
27 | +import io.atomix.resource.ResourceStateMachine; | ||
28 | + | ||
29 | +import java.util.HashMap; | ||
30 | +import java.util.Map; | ||
31 | +import java.util.Optional; | ||
32 | +import java.util.Properties; | ||
33 | +import java.util.Queue; | ||
34 | +import java.util.concurrent.atomic.AtomicLong; | ||
35 | +import java.util.stream.Collectors; | ||
36 | + | ||
37 | +import org.onlab.util.Match; | ||
38 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Clear; | ||
39 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Get; | ||
40 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.GetChildren; | ||
41 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Listen; | ||
42 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Unlisten; | ||
43 | +import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Update; | ||
44 | +import org.onosproject.store.primitives.resources.impl.DocumentTreeUpdateResult.Status; | ||
45 | +import org.onosproject.store.service.DocumentPath; | ||
46 | +import org.onosproject.store.service.DocumentTree; | ||
47 | +import org.onosproject.store.service.DocumentTreeEvent; | ||
48 | +import org.onosproject.store.service.DocumentTreeEvent.Type; | ||
49 | +import org.onosproject.store.service.IllegalDocumentModificationException; | ||
50 | +import org.onosproject.store.service.NoSuchDocumentPathException; | ||
51 | +import org.onosproject.store.service.Versioned; | ||
52 | +import org.slf4j.Logger; | ||
53 | + | ||
54 | +import com.google.common.base.Throwables; | ||
55 | +import com.google.common.collect.ImmutableList; | ||
56 | +import com.google.common.collect.Maps; | ||
57 | +import com.google.common.collect.Queues; | ||
58 | + | ||
59 | +/** | ||
60 | + * State Machine for {@link AtomixDocumentTree} resource. | ||
61 | + */ | ||
62 | +public class AtomixDocumentTreeState | ||
63 | + extends ResourceStateMachine | ||
64 | + implements SessionListener, Snapshottable { | ||
65 | + | ||
66 | + private final Logger log = getLogger(getClass()); | ||
67 | + private final Map<Long, Commit<? extends Listen>> listeners = new HashMap<>(); | ||
68 | + private final AtomicLong versionCounter = new AtomicLong(0); | ||
69 | + private final DocumentTree<TreeNodeValue> docTree = new DefaultDocumentTree<>(versionCounter::incrementAndGet); | ||
70 | + | ||
71 | + public AtomixDocumentTreeState(Properties properties) { | ||
72 | + super(properties); | ||
73 | + } | ||
74 | + | ||
75 | + @Override | ||
76 | + public void snapshot(SnapshotWriter writer) { | ||
77 | + writer.writeLong(versionCounter.get()); | ||
78 | + } | ||
79 | + | ||
80 | + @Override | ||
81 | + public void install(SnapshotReader reader) { | ||
82 | + versionCounter.set(reader.readLong()); | ||
83 | + } | ||
84 | + | ||
85 | + @Override | ||
86 | + protected void configure(StateMachineExecutor executor) { | ||
87 | + // Listeners | ||
88 | + executor.register(Listen.class, this::listen); | ||
89 | + executor.register(Unlisten.class, this::unlisten); | ||
90 | + // queries | ||
91 | + executor.register(Get.class, this::get); | ||
92 | + executor.register(GetChildren.class, this::getChildren); | ||
93 | + // commands | ||
94 | + executor.register(Update.class, this::update); | ||
95 | + executor.register(Clear.class, this::clear); | ||
96 | + } | ||
97 | + | ||
98 | + protected void listen(Commit<? extends Listen> commit) { | ||
99 | + Long sessionId = commit.session().id(); | ||
100 | + if (listeners.putIfAbsent(sessionId, commit) != null) { | ||
101 | + commit.close(); | ||
102 | + return; | ||
103 | + } | ||
104 | + commit.session() | ||
105 | + .onStateChange( | ||
106 | + state -> { | ||
107 | + if (state == ServerSession.State.CLOSED | ||
108 | + || state == ServerSession.State.EXPIRED) { | ||
109 | + Commit<? extends Listen> listener = listeners.remove(sessionId); | ||
110 | + if (listener != null) { | ||
111 | + listener.close(); | ||
112 | + } | ||
113 | + } | ||
114 | + }); | ||
115 | + } | ||
116 | + | ||
117 | + protected void unlisten(Commit<? extends Unlisten> commit) { | ||
118 | + try { | ||
119 | + closeListener(commit.session().id()); | ||
120 | + } finally { | ||
121 | + commit.close(); | ||
122 | + } | ||
123 | + } | ||
124 | + | ||
125 | + protected Versioned<byte[]> get(Commit<? extends Get> commit) { | ||
126 | + try { | ||
127 | + Versioned<TreeNodeValue> value = docTree.get(commit.operation().path()); | ||
128 | + return value == null ? null : value.map(node -> node == null ? null : node.value()); | ||
129 | + } finally { | ||
130 | + commit.close(); | ||
131 | + } | ||
132 | + } | ||
133 | + | ||
134 | + protected Map<String, Versioned<byte[]>> getChildren(Commit<? extends GetChildren> commit) { | ||
135 | + try { | ||
136 | + Map<String, Versioned<TreeNodeValue>> children = docTree.getChildren(commit.operation().path()); | ||
137 | + return children == null | ||
138 | + ? null : Maps.newHashMap(Maps.transformValues(children, | ||
139 | + value -> value.map(TreeNodeValue::value))); | ||
140 | + } finally { | ||
141 | + commit.close(); | ||
142 | + } | ||
143 | + } | ||
144 | + | ||
145 | + protected DocumentTreeUpdateResult<byte[]> update(Commit<? extends Update> commit) { | ||
146 | + DocumentTreeUpdateResult<byte[]> result = null; | ||
147 | + DocumentPath path = commit.operation().path(); | ||
148 | + boolean updated = false; | ||
149 | + Versioned<TreeNodeValue> currentValue = docTree.get(path); | ||
150 | + try { | ||
151 | + Match<Long> versionMatch = commit.operation().versionMatch(); | ||
152 | + Match<byte[]> valueMatch = commit.operation().valueMatch(); | ||
153 | + | ||
154 | + if (versionMatch.matches(currentValue == null ? null : currentValue.version()) | ||
155 | + && valueMatch.matches(currentValue == null ? null : currentValue.value().value())) { | ||
156 | + if (commit.operation().value() == null) { | ||
157 | + docTree.removeNode(path); | ||
158 | + } else { | ||
159 | + docTree.set(path, new NonTransactionalCommit(commit)); | ||
160 | + } | ||
161 | + updated = true; | ||
162 | + } | ||
163 | + Versioned<TreeNodeValue> newValue = updated ? docTree.get(path) : currentValue; | ||
164 | + Status updateStatus = updated | ||
165 | + ? Status.OK : commit.operation().value() == null ? Status.INVALID_PATH : Status.NOOP; | ||
166 | + result = new DocumentTreeUpdateResult<>(path, | ||
167 | + updateStatus, | ||
168 | + newValue == null | ||
169 | + ? null : newValue.map(TreeNodeValue::value), | ||
170 | + currentValue == null | ||
171 | + ? null : currentValue.map(TreeNodeValue::value)); | ||
172 | + } catch (IllegalDocumentModificationException e) { | ||
173 | + result = DocumentTreeUpdateResult.illegalModification(path); | ||
174 | + } catch (NoSuchDocumentPathException e) { | ||
175 | + result = DocumentTreeUpdateResult.invalidPath(path); | ||
176 | + } catch (Exception e) { | ||
177 | + log.error("Failed to apply {} to state machine", commit.operation(), e); | ||
178 | + throw Throwables.propagate(e); | ||
179 | + } finally { | ||
180 | + if (updated) { | ||
181 | + if (currentValue != null) { | ||
182 | + currentValue.value().discard(); | ||
183 | + } | ||
184 | + } else { | ||
185 | + commit.close(); | ||
186 | + } | ||
187 | + } | ||
188 | + notifyListeners(path, result); | ||
189 | + return result; | ||
190 | + } | ||
191 | + | ||
192 | + protected void clear(Commit<? extends Clear> commit) { | ||
193 | + try { | ||
194 | + Queue<DocumentPath> toClearQueue = Queues.newArrayDeque(); | ||
195 | + Map<String, Versioned<TreeNodeValue>> topLevelChildren = docTree.getChildren(DocumentPath.from("root")); | ||
196 | + toClearQueue.addAll(topLevelChildren.keySet() | ||
197 | + .stream() | ||
198 | + .map(name -> new DocumentPath(name, DocumentPath.from("root"))) | ||
199 | + .collect(Collectors.toList())); | ||
200 | + while (!toClearQueue.isEmpty()) { | ||
201 | + DocumentPath path = toClearQueue.remove(); | ||
202 | + Map<String, Versioned<TreeNodeValue>> children = docTree.getChildren(path); | ||
203 | + if (children.size() == 0) { | ||
204 | + docTree.removeNode(path).value().discard(); | ||
205 | + } else { | ||
206 | + children.keySet() | ||
207 | + .stream() | ||
208 | + .forEach(name -> toClearQueue.add(new DocumentPath(name, path))); | ||
209 | + toClearQueue.add(path); | ||
210 | + } | ||
211 | + } | ||
212 | + } finally { | ||
213 | + commit.close(); | ||
214 | + } | ||
215 | + } | ||
216 | + | ||
217 | + /** | ||
218 | + * Interface implemented by tree node values. | ||
219 | + */ | ||
220 | + private interface TreeNodeValue { | ||
221 | + /** | ||
222 | + * Returns the raw {@code byte[]}. | ||
223 | + * | ||
224 | + * @return raw value | ||
225 | + */ | ||
226 | + byte[] value(); | ||
227 | + | ||
228 | + /** | ||
229 | + * Discards the value by invoke appropriate clean up actions. | ||
230 | + */ | ||
231 | + void discard(); | ||
232 | + } | ||
233 | + | ||
234 | + /** | ||
235 | + * A {@code TreeNodeValue} that is derived from a non-transactional update | ||
236 | + * i.e. via any standard tree update operation. | ||
237 | + */ | ||
238 | + private class NonTransactionalCommit implements TreeNodeValue { | ||
239 | + private final Commit<? extends Update> commit; | ||
240 | + | ||
241 | + public NonTransactionalCommit(Commit<? extends Update> commit) { | ||
242 | + this.commit = commit; | ||
243 | + } | ||
244 | + | ||
245 | + @Override | ||
246 | + public byte[] value() { | ||
247 | + return commit.operation().value(); | ||
248 | + } | ||
249 | + | ||
250 | + @Override | ||
251 | + public void discard() { | ||
252 | + commit.close(); | ||
253 | + } | ||
254 | + } | ||
255 | + | ||
256 | + private void notifyListeners(DocumentPath path, DocumentTreeUpdateResult<byte[]> result) { | ||
257 | + if (result.status() != Status.OK) { | ||
258 | + return; | ||
259 | + } | ||
260 | + DocumentTreeEvent<byte[]> event = | ||
261 | + new DocumentTreeEvent<>(path, | ||
262 | + result.created() ? Type.CREATED : result.newValue() == null ? Type.DELETED : Type.UPDATED, | ||
263 | + Optional.ofNullable(result.newValue()), | ||
264 | + Optional.ofNullable(result.oldValue())); | ||
265 | + Object message = ImmutableList.of(event); | ||
266 | + listeners.values().forEach(commit -> { | ||
267 | + commit.session().publish(AtomixDocumentTree.CHANGE_SUBJECT, message); | ||
268 | + System.out.println("Sent " + message + " to " + commit.session().id()); | ||
269 | + }); | ||
270 | + } | ||
271 | + | ||
272 | + @Override | ||
273 | + public void register(ServerSession session) { | ||
274 | + } | ||
275 | + | ||
276 | + @Override | ||
277 | + public void unregister(ServerSession session) { | ||
278 | + closeListener(session.id()); | ||
279 | + } | ||
280 | + | ||
281 | + @Override | ||
282 | + public void expire(ServerSession session) { | ||
283 | + closeListener(session.id()); | ||
284 | + } | ||
285 | + | ||
286 | + @Override | ||
287 | + public void close(ServerSession session) { | ||
288 | + closeListener(session.id()); | ||
289 | + } | ||
290 | + | ||
291 | + private void closeListener(Long sessionId) { | ||
292 | + Commit<? extends Listen> commit = listeners.remove(sessionId); | ||
293 | + if (commit != null) { | ||
294 | + commit.close(); | ||
295 | + } | ||
296 | + } | ||
297 | +} |
... | @@ -19,7 +19,7 @@ package org.onosproject.store.primitives.resources.impl; | ... | @@ -19,7 +19,7 @@ package org.onosproject.store.primitives.resources.impl; |
19 | import java.util.Iterator; | 19 | import java.util.Iterator; |
20 | import java.util.Map; | 20 | import java.util.Map; |
21 | import java.util.Objects; | 21 | import java.util.Objects; |
22 | -import java.util.concurrent.atomic.AtomicInteger; | 22 | +import java.util.concurrent.atomic.AtomicLong; |
23 | 23 | ||
24 | import org.onosproject.store.service.DocumentPath; | 24 | import org.onosproject.store.service.DocumentPath; |
25 | import org.onosproject.store.service.DocumentTree; | 25 | import org.onosproject.store.service.DocumentTree; |
... | @@ -30,6 +30,7 @@ import org.onosproject.store.service.NoSuchDocumentPathException; | ... | @@ -30,6 +30,7 @@ import org.onosproject.store.service.NoSuchDocumentPathException; |
30 | import org.onosproject.store.service.Versioned; | 30 | import org.onosproject.store.service.Versioned; |
31 | 31 | ||
32 | import com.google.common.base.Preconditions; | 32 | import com.google.common.base.Preconditions; |
33 | +import com.google.common.base.Supplier; | ||
33 | import com.google.common.collect.Maps; | 34 | import com.google.common.collect.Maps; |
34 | 35 | ||
35 | /** | 36 | /** |
... | @@ -41,10 +42,17 @@ public class DefaultDocumentTree<V> implements DocumentTree<V> { | ... | @@ -41,10 +42,17 @@ public class DefaultDocumentTree<V> implements DocumentTree<V> { |
41 | 42 | ||
42 | private static final DocumentPath ROOT_PATH = DocumentPath.from("root"); | 43 | private static final DocumentPath ROOT_PATH = DocumentPath.from("root"); |
43 | private final DefaultDocumentTreeNode<V> root; | 44 | private final DefaultDocumentTreeNode<V> root; |
44 | - private final AtomicInteger versionCounter = new AtomicInteger(0); | 45 | + private final Supplier<Long> versionSupplier; |
45 | 46 | ||
46 | public DefaultDocumentTree() { | 47 | public DefaultDocumentTree() { |
47 | - root = new DefaultDocumentTreeNode<V>(ROOT_PATH, null, nextVersion(), null); | 48 | + AtomicLong versionCounter = new AtomicLong(0); |
49 | + versionSupplier = versionCounter::incrementAndGet; | ||
50 | + root = new DefaultDocumentTreeNode<V>(ROOT_PATH, null, versionSupplier.get(), null); | ||
51 | + } | ||
52 | + | ||
53 | + public DefaultDocumentTree(Supplier<Long> versionSupplier) { | ||
54 | + root = new DefaultDocumentTreeNode<V>(ROOT_PATH, null, versionSupplier.get(), null); | ||
55 | + this.versionSupplier = versionSupplier; | ||
48 | } | 56 | } |
49 | 57 | ||
50 | @Override | 58 | @Override |
... | @@ -74,7 +82,7 @@ public class DefaultDocumentTree<V> implements DocumentTree<V> { | ... | @@ -74,7 +82,7 @@ public class DefaultDocumentTree<V> implements DocumentTree<V> { |
74 | checkRootModification(path); | 82 | checkRootModification(path); |
75 | DefaultDocumentTreeNode<V> node = getNode(path); | 83 | DefaultDocumentTreeNode<V> node = getNode(path); |
76 | if (node != null) { | 84 | if (node != null) { |
77 | - return node.update(value, nextVersion()); | 85 | + return node.update(value, versionSupplier.get()); |
78 | } else { | 86 | } else { |
79 | create(path, value); | 87 | create(path, value); |
80 | return null; | 88 | return null; |
... | @@ -93,7 +101,7 @@ public class DefaultDocumentTree<V> implements DocumentTree<V> { | ... | @@ -93,7 +101,7 @@ public class DefaultDocumentTree<V> implements DocumentTree<V> { |
93 | if (parentNode == null) { | 101 | if (parentNode == null) { |
94 | throw new IllegalDocumentModificationException(); | 102 | throw new IllegalDocumentModificationException(); |
95 | } | 103 | } |
96 | - parentNode.addChild(simpleName(path), value, nextVersion()); | 104 | + parentNode.addChild(simpleName(path), value, versionSupplier.get()); |
97 | return true; | 105 | return true; |
98 | } | 106 | } |
99 | 107 | ||
... | @@ -159,10 +167,6 @@ public class DefaultDocumentTree<V> implements DocumentTree<V> { | ... | @@ -159,10 +167,6 @@ public class DefaultDocumentTree<V> implements DocumentTree<V> { |
159 | return currentNode; | 167 | return currentNode; |
160 | } | 168 | } |
161 | 169 | ||
162 | - private long nextVersion() { | ||
163 | - return versionCounter.incrementAndGet(); | ||
164 | - } | ||
165 | - | ||
166 | private String simpleName(DocumentPath path) { | 170 | private String simpleName(DocumentPath path) { |
167 | return path.pathElements().get(path.pathElements().size() - 1); | 171 | return path.pathElements().get(path.pathElements().size() - 1); |
168 | } | 172 | } | ... | ... |
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.store.primitives.resources.impl; | ||
18 | + | ||
19 | +import org.onosproject.store.service.DocumentPath; | ||
20 | +import org.onosproject.store.service.Versioned; | ||
21 | + | ||
22 | +import com.google.common.base.MoreObjects; | ||
23 | + | ||
24 | +/** | ||
25 | + * Result of a document tree node update operation. | ||
26 | + * <p> | ||
27 | + * Both old and new values are accessible along with a status of update. | ||
28 | + * | ||
29 | + * @param <V> value type | ||
30 | + */ | ||
31 | +public class DocumentTreeUpdateResult<V> { | ||
32 | + | ||
33 | + public enum Status { | ||
34 | + /** | ||
35 | + * Indicates a successful update. | ||
36 | + */ | ||
37 | + OK, | ||
38 | + | ||
39 | + /** | ||
40 | + * Indicates a noop i.e. existing and new value are both same. | ||
41 | + */ | ||
42 | + NOOP, | ||
43 | + | ||
44 | + /** | ||
45 | + * Indicates a failed update due to a write lock. | ||
46 | + */ | ||
47 | + WRITE_LOCK, | ||
48 | + | ||
49 | + /** | ||
50 | + * Indicates a failed update due to a invalid path. | ||
51 | + */ | ||
52 | + INVALID_PATH, | ||
53 | + | ||
54 | + /** | ||
55 | + * Indicates a failed update due to a illegal modification attempt. | ||
56 | + */ | ||
57 | + ILLEGAL_MODIFICATION, | ||
58 | + } | ||
59 | + | ||
60 | + private final DocumentPath path; | ||
61 | + private final Status status; | ||
62 | + private final Versioned<V> oldValue; | ||
63 | + private final Versioned<V> newValue; | ||
64 | + | ||
65 | + public DocumentTreeUpdateResult(DocumentPath path, | ||
66 | + Status status, | ||
67 | + Versioned<V> newValue, | ||
68 | + Versioned<V> oldValue) { | ||
69 | + this.status = status; | ||
70 | + this.path = path; | ||
71 | + this.newValue = newValue; | ||
72 | + this.oldValue = oldValue; | ||
73 | + } | ||
74 | + | ||
75 | + public static <V> DocumentTreeUpdateResult<V> invalidPath(DocumentPath path) { | ||
76 | + return new DocumentTreeUpdateResult<>(path, Status.INVALID_PATH, null, null); | ||
77 | + } | ||
78 | + | ||
79 | + public static <V> DocumentTreeUpdateResult<V> illegalModification(DocumentPath path) { | ||
80 | + return new DocumentTreeUpdateResult<>(path, Status.ILLEGAL_MODIFICATION, null, null); | ||
81 | + } | ||
82 | + | ||
83 | + public Status status() { | ||
84 | + return status; | ||
85 | + } | ||
86 | + | ||
87 | + public DocumentPath path() { | ||
88 | + return path; | ||
89 | + } | ||
90 | + | ||
91 | + public Versioned<V> oldValue() { | ||
92 | + return oldValue; | ||
93 | + } | ||
94 | + | ||
95 | + public Versioned<V> newValue() { | ||
96 | + return this.newValue; | ||
97 | + } | ||
98 | + | ||
99 | + public boolean updated() { | ||
100 | + return status == Status.OK; | ||
101 | + } | ||
102 | + | ||
103 | + public boolean created() { | ||
104 | + return updated() && oldValue == null; | ||
105 | + } | ||
106 | + | ||
107 | + @Override | ||
108 | + public String toString() { | ||
109 | + return MoreObjects.toStringHelper(getClass()) | ||
110 | + .add("path", path) | ||
111 | + .add("status", status) | ||
112 | + .add("newValue", newValue) | ||
113 | + .add("oldValue", oldValue) | ||
114 | + .toString(); | ||
115 | + } | ||
116 | +} |
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.store.primitives.resources.impl; | ||
18 | + | ||
19 | +import static org.junit.Assert.assertArrayEquals; | ||
20 | +import static org.junit.Assert.assertEquals; | ||
21 | +import static org.junit.Assert.assertFalse; | ||
22 | +import static org.junit.Assert.assertNotNull; | ||
23 | +import static org.junit.Assert.assertNull; | ||
24 | +import static org.junit.Assert.assertTrue; | ||
25 | +import static org.junit.Assert.fail; | ||
26 | +import io.atomix.resource.ResourceType; | ||
27 | + | ||
28 | +import java.util.Map; | ||
29 | +import java.util.UUID; | ||
30 | +import java.util.concurrent.ArrayBlockingQueue; | ||
31 | +import java.util.concurrent.BlockingQueue; | ||
32 | + | ||
33 | +import org.junit.AfterClass; | ||
34 | +import org.junit.BeforeClass; | ||
35 | +import org.junit.Ignore; | ||
36 | +import org.junit.Test; | ||
37 | +import org.onosproject.store.service.DocumentPath; | ||
38 | +import org.onosproject.store.service.DocumentTreeEvent; | ||
39 | +import org.onosproject.store.service.DocumentTreeListener; | ||
40 | +import org.onosproject.store.service.IllegalDocumentModificationException; | ||
41 | +import org.onosproject.store.service.NoSuchDocumentPathException; | ||
42 | +import org.onosproject.store.service.Versioned; | ||
43 | + | ||
44 | +import com.google.common.base.Throwables; | ||
45 | + | ||
46 | +/** | ||
47 | + * Unit tests for {@link AtomixDocumentTree}. | ||
48 | + */ | ||
49 | +public class AtomixDocumentTreeTest extends AtomixTestBase { | ||
50 | + @BeforeClass | ||
51 | + public static void preTestSetup() throws Throwable { | ||
52 | + createCopycatServers(3); | ||
53 | + } | ||
54 | + | ||
55 | + @AfterClass | ||
56 | + public static void postTestCleanup() throws Exception { | ||
57 | + clearTests(); | ||
58 | + } | ||
59 | + @Override | ||
60 | + protected ResourceType resourceType() { | ||
61 | + return new ResourceType(AtomixDocumentTree.class); | ||
62 | + } | ||
63 | + /** | ||
64 | + * Tests queries (get and getChildren). | ||
65 | + */ | ||
66 | + @Test | ||
67 | + public void testQueries() throws Throwable { | ||
68 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
69 | + AtomixDocumentTree.class).join(); | ||
70 | + Versioned<byte[]> root = tree.get(DocumentPath.from("root")).join(); | ||
71 | + assertEquals(1, root.version()); | ||
72 | + assertNull(root.value()); | ||
73 | + } | ||
74 | + | ||
75 | + /** | ||
76 | + * Tests create. | ||
77 | + */ | ||
78 | + @Test | ||
79 | + public void testCreate() throws Throwable { | ||
80 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
81 | + AtomixDocumentTree.class).join(); | ||
82 | + tree.create(DocumentPath.from("root.a"), "a".getBytes()).join(); | ||
83 | + tree.create(DocumentPath.from("root.a.b"), "ab".getBytes()).join(); | ||
84 | + tree.create(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
85 | + Versioned<byte[]> a = tree.get(DocumentPath.from("root.a")).join(); | ||
86 | + assertArrayEquals("a".getBytes(), a.value()); | ||
87 | + | ||
88 | + Versioned<byte[]> ab = tree.get(DocumentPath.from("root.a.b")).join(); | ||
89 | + assertArrayEquals("ab".getBytes(), ab.value()); | ||
90 | + | ||
91 | + Versioned<byte[]> ac = tree.get(DocumentPath.from("root.a.c")).join(); | ||
92 | + assertArrayEquals("ac".getBytes(), ac.value()); | ||
93 | + } | ||
94 | + | ||
95 | + /** | ||
96 | + * Tests set. | ||
97 | + */ | ||
98 | + @Test | ||
99 | + public void testSet() throws Throwable { | ||
100 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
101 | + AtomixDocumentTree.class).join(); | ||
102 | + tree.create(DocumentPath.from("root.a"), "a".getBytes()).join(); | ||
103 | + tree.create(DocumentPath.from("root.a.b"), "ab".getBytes()).join(); | ||
104 | + tree.create(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
105 | + | ||
106 | + tree.set(DocumentPath.from("root.a.d"), "ad".getBytes()).join(); | ||
107 | + Versioned<byte[]> ad = tree.get(DocumentPath.from("root.a.d")).join(); | ||
108 | + assertArrayEquals("ad".getBytes(), ad.value()); | ||
109 | + | ||
110 | + tree.set(DocumentPath.from("root.a"), "newA".getBytes()).join(); | ||
111 | + Versioned<byte[]> newA = tree.get(DocumentPath.from("root.a")).join(); | ||
112 | + assertArrayEquals("newA".getBytes(), newA.value()); | ||
113 | + | ||
114 | + tree.set(DocumentPath.from("root.a.b"), "newAB".getBytes()).join(); | ||
115 | + Versioned<byte[]> newAB = tree.get(DocumentPath.from("root.a.b")).join(); | ||
116 | + assertArrayEquals("newAB".getBytes(), newAB.value()); | ||
117 | + } | ||
118 | + | ||
119 | + /** | ||
120 | + * Tests replace if version matches. | ||
121 | + */ | ||
122 | + @Test | ||
123 | + public void testReplaceVersion() throws Throwable { | ||
124 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
125 | + AtomixDocumentTree.class).join(); | ||
126 | + tree.create(DocumentPath.from("root.a"), "a".getBytes()).join(); | ||
127 | + tree.create(DocumentPath.from("root.a.b"), "ab".getBytes()).join(); | ||
128 | + tree.create(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
129 | + | ||
130 | + Versioned<byte[]> ab = tree.get(DocumentPath.from("root.a.b")).join(); | ||
131 | + assertTrue(tree.replace(DocumentPath.from("root.a.b"), "newAB".getBytes(), ab.version()).join()); | ||
132 | + Versioned<byte[]> newAB = tree.get(DocumentPath.from("root.a.b")).join(); | ||
133 | + assertArrayEquals("newAB".getBytes(), newAB.value()); | ||
134 | + | ||
135 | + assertFalse(tree.replace(DocumentPath.from("root.a.b"), "newestAB".getBytes(), ab.version()).join()); | ||
136 | + assertArrayEquals("newAB".getBytes(), tree.get(DocumentPath.from("root.a.b")).join().value()); | ||
137 | + | ||
138 | + assertFalse(tree.replace(DocumentPath.from("root.a.d"), "foo".getBytes(), 1).join()); | ||
139 | + } | ||
140 | + | ||
141 | + /** | ||
142 | + * Tests replace if value matches. | ||
143 | + */ | ||
144 | + @Test | ||
145 | + public void testReplaceValue() throws Throwable { | ||
146 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
147 | + AtomixDocumentTree.class).join(); | ||
148 | + tree.create(DocumentPath.from("root.a"), "a".getBytes()).join(); | ||
149 | + tree.create(DocumentPath.from("root.a.b"), "ab".getBytes()).join(); | ||
150 | + tree.create(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
151 | + | ||
152 | + Versioned<byte[]> ab = tree.get(DocumentPath.from("root.a.b")).join(); | ||
153 | + assertTrue(tree.replace(DocumentPath.from("root.a.b"), "newAB".getBytes(), ab.value()).join()); | ||
154 | + Versioned<byte[]> newAB = tree.get(DocumentPath.from("root.a.b")).join(); | ||
155 | + assertArrayEquals("newAB".getBytes(), newAB.value()); | ||
156 | + | ||
157 | + assertFalse(tree.replace(DocumentPath.from("root.a.b"), "newestAB".getBytes(), ab.value()).join()); | ||
158 | + assertArrayEquals("newAB".getBytes(), tree.get(DocumentPath.from("root.a.b")).join().value()); | ||
159 | + | ||
160 | + assertFalse(tree.replace(DocumentPath.from("root.a.d"), "bar".getBytes(), "foo".getBytes()).join()); | ||
161 | + } | ||
162 | + | ||
163 | + /** | ||
164 | + * Tests remove. | ||
165 | + */ | ||
166 | + @Test | ||
167 | + public void testRemove() throws Throwable { | ||
168 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
169 | + AtomixDocumentTree.class).join(); | ||
170 | + tree.create(DocumentPath.from("root.a"), "a".getBytes()).join(); | ||
171 | + tree.create(DocumentPath.from("root.a.b"), "ab".getBytes()).join(); | ||
172 | + tree.create(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
173 | + | ||
174 | + Versioned<byte[]> ab = tree.removeNode(DocumentPath.from("root.a.b")).join(); | ||
175 | + assertArrayEquals("ab".getBytes(), ab.value()); | ||
176 | + assertNull(tree.get(DocumentPath.from("root.a.b")).join()); | ||
177 | + | ||
178 | + Versioned<byte[]> ac = tree.removeNode(DocumentPath.from("root.a.c")).join(); | ||
179 | + assertArrayEquals("ac".getBytes(), ac.value()); | ||
180 | + assertNull(tree.get(DocumentPath.from("root.a.c")).join()); | ||
181 | + | ||
182 | + Versioned<byte[]> a = tree.removeNode(DocumentPath.from("root.a")).join(); | ||
183 | + assertArrayEquals("a".getBytes(), a.value()); | ||
184 | + assertNull(tree.get(DocumentPath.from("root.a")).join()); | ||
185 | + } | ||
186 | + | ||
187 | + /** | ||
188 | + * Tests invalid removes. | ||
189 | + */ | ||
190 | + @Test | ||
191 | + public void testRemoveFailures() throws Throwable { | ||
192 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
193 | + AtomixDocumentTree.class).join(); | ||
194 | + tree.create(DocumentPath.from("root.a"), "a".getBytes()).join(); | ||
195 | + tree.create(DocumentPath.from("root.a.b"), "ab".getBytes()).join(); | ||
196 | + tree.create(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
197 | + | ||
198 | + try { | ||
199 | + tree.removeNode(DocumentPath.from("root")).join(); | ||
200 | + fail(); | ||
201 | + } catch (Exception e) { | ||
202 | + assertTrue(Throwables.getRootCause(e) instanceof IllegalDocumentModificationException); | ||
203 | + } | ||
204 | + | ||
205 | + try { | ||
206 | + tree.removeNode(DocumentPath.from("root.a")).join(); | ||
207 | + fail(); | ||
208 | + } catch (Exception e) { | ||
209 | + assertTrue(Throwables.getRootCause(e) instanceof IllegalDocumentModificationException); | ||
210 | + } | ||
211 | + | ||
212 | + try { | ||
213 | + tree.removeNode(DocumentPath.from("root.d")).join(); | ||
214 | + fail(); | ||
215 | + } catch (Exception e) { | ||
216 | + assertTrue(Throwables.getRootCause(e) instanceof NoSuchDocumentPathException); | ||
217 | + } | ||
218 | + } | ||
219 | + | ||
220 | + /** | ||
221 | + * Tests invalid create. | ||
222 | + */ | ||
223 | + @Test | ||
224 | + public void testCreateFailures() throws Throwable { | ||
225 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
226 | + AtomixDocumentTree.class).join(); | ||
227 | + try { | ||
228 | + tree.create(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
229 | + fail(); | ||
230 | + } catch (Exception e) { | ||
231 | + assertTrue(Throwables.getRootCause(e) instanceof IllegalDocumentModificationException); | ||
232 | + } | ||
233 | + } | ||
234 | + | ||
235 | + /** | ||
236 | + * Tests invalid set. | ||
237 | + */ | ||
238 | + @Test | ||
239 | + public void testSetFailures() throws Throwable { | ||
240 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
241 | + AtomixDocumentTree.class).join(); | ||
242 | + try { | ||
243 | + tree.set(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
244 | + fail(); | ||
245 | + } catch (Exception e) { | ||
246 | + assertTrue(Throwables.getRootCause(e) instanceof IllegalDocumentModificationException); | ||
247 | + } | ||
248 | + } | ||
249 | + | ||
250 | + /** | ||
251 | + * Tests getChildren. | ||
252 | + */ | ||
253 | + @Test | ||
254 | + public void testGetChildren() throws Throwable { | ||
255 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
256 | + AtomixDocumentTree.class).join(); | ||
257 | + tree.create(DocumentPath.from("root.a"), "a".getBytes()).join(); | ||
258 | + tree.create(DocumentPath.from("root.a.b"), "ab".getBytes()).join(); | ||
259 | + tree.create(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
260 | + | ||
261 | + Map<String, Versioned<byte[]>> rootChildren = tree.getChildren(DocumentPath.from("root")).join(); | ||
262 | + assertEquals(1, rootChildren.size()); | ||
263 | + Versioned<byte[]> a = rootChildren.get("a"); | ||
264 | + assertArrayEquals("a".getBytes(), a.value()); | ||
265 | + | ||
266 | + Map<String, Versioned<byte[]>> children = tree.getChildren(DocumentPath.from("root.a")).join(); | ||
267 | + assertEquals(2, children.size()); | ||
268 | + Versioned<byte[]> ab = children.get("b"); | ||
269 | + assertArrayEquals("ab".getBytes(), ab.value()); | ||
270 | + Versioned<byte[]> ac = children.get("c"); | ||
271 | + assertArrayEquals("ac".getBytes(), ac.value()); | ||
272 | + | ||
273 | + assertEquals(0, tree.getChildren(DocumentPath.from("root.a.b")).join().size()); | ||
274 | + assertEquals(0, tree.getChildren(DocumentPath.from("root.a.c")).join().size()); | ||
275 | + } | ||
276 | + | ||
277 | + /** | ||
278 | + * Tests destroy. | ||
279 | + */ | ||
280 | + @Test | ||
281 | + public void testClear() { | ||
282 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
283 | + AtomixDocumentTree.class).join(); | ||
284 | + tree.create(DocumentPath.from("root.a"), "a".getBytes()).join(); | ||
285 | + tree.create(DocumentPath.from("root.a.b"), "ab".getBytes()).join(); | ||
286 | + tree.create(DocumentPath.from("root.a.c"), "ac".getBytes()).join(); | ||
287 | + | ||
288 | + tree.destroy().join(); | ||
289 | + assertEquals(0, tree.getChildren(DocumentPath.from("root")).join().size()); | ||
290 | + } | ||
291 | + | ||
292 | + /** | ||
293 | + * Tests listeners. | ||
294 | + */ | ||
295 | + @Test | ||
296 | + @Ignore | ||
297 | + public void testNotifications() throws Exception { | ||
298 | + AtomixDocumentTree tree = createAtomixClient().getResource(UUID.randomUUID().toString(), | ||
299 | + AtomixDocumentTree.class).join(); | ||
300 | + TestEventListener listener = new TestEventListener(); | ||
301 | + | ||
302 | + // add listener; create a node in the tree and verify an CREATED event is received. | ||
303 | + tree.addListener(listener).thenCompose(v -> tree.set(DocumentPath.from("root.a"), "a".getBytes())).join(); | ||
304 | + DocumentTreeEvent<byte[]> event = listener.event(); | ||
305 | + assertNotNull(event); | ||
306 | + assertEquals(DocumentTreeEvent.Type.CREATED, event.type()); | ||
307 | + assertArrayEquals("a".getBytes(), event.newValue().get().value()); | ||
308 | + } | ||
309 | + | ||
310 | + private static class TestEventListener implements DocumentTreeListener<byte[]> { | ||
311 | + | ||
312 | + private final BlockingQueue<DocumentTreeEvent<byte[]>> queue = new ArrayBlockingQueue<>(1); | ||
313 | + | ||
314 | + @Override | ||
315 | + public void event(DocumentTreeEvent<byte[]> event) { | ||
316 | + try { | ||
317 | + queue.put(event); | ||
318 | + } catch (InterruptedException e) { | ||
319 | + Throwables.propagate(e); | ||
320 | + } | ||
321 | + } | ||
322 | + | ||
323 | + public boolean eventReceived() { | ||
324 | + return !queue.isEmpty(); | ||
325 | + } | ||
326 | + | ||
327 | + public DocumentTreeEvent<byte[]> event() throws InterruptedException { | ||
328 | + return queue.take(); | ||
329 | + } | ||
330 | + } | ||
331 | +} |
-
Please register or login to post a comment