홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Madan Jampani
Committed by Gerrit Code Review
74da78b6 1 parent 1110b2ce

AsyncConsistentMap methods for supporting transactional updates 

Change-Id: Iaeb0aa0abf9f52d514a2c040598599a5b8a55ee8
Inline Side-by-side
Showing 21 changed files with 353 additions and 71 deletions
...@@ -14,12 +14,13 @@ ...@@ -14,12 +14,13 @@
14 * limitations under the License. 14 * limitations under the License.
15 */ 15 */
16 16
17 -package org.onosproject.store.primitives.resources.impl; 17 +package org.onosproject.store.primitives;
18 18
19 -import static com.google.common.base.Preconditions.checkArgument;
20 import static com.google.common.base.Preconditions.checkNotNull; 19 import static com.google.common.base.Preconditions.checkNotNull;
21 import static com.google.common.base.Preconditions.checkState; 20 import static com.google.common.base.Preconditions.checkState;
22 21
22 +import java.util.function.Function;
23 +
23 import com.google.common.base.MoreObjects; 24 import com.google.common.base.MoreObjects;
24 25
25 /** 26 /**
...@@ -126,6 +127,26 @@ public final class MapUpdate<K, V> { ...@@ -126,6 +127,26 @@ public final class MapUpdate<K, V> {
126 return currentVersion; 127 return currentVersion;
127 } 128 }
128 129
130 + /**
131 + * Transforms this instance into an instance of different paramterized types.
132 + *
133 + * @param keyMapper transcoder for key type
134 + * @param valueMapper transcoder to value type
135 + * @return new instance
136 + * @param <S> key type of returned instance
137 + * @param <T> value type of returned instance
138 + */
139 + public <S, T> MapUpdate<S, T> map(Function<K, S> keyMapper, Function<V, T> valueMapper) {
140 + return MapUpdate.<S, T>newBuilder()
141 + .withMapName(mapName)
142 + .withType(type)
143 + .withKey(keyMapper.apply(key))
144 + .withValue(value == null ? null : valueMapper.apply(value))
145 + .withCurrentValue(currentValue == null ? null : valueMapper.apply(currentValue))
146 + .withCurrentVersion(currentVersion)
147 + .build();
148 + }
149 +
129 @Override 150 @Override
130 public String toString() { 151 public String toString() {
131 return MoreObjects.toStringHelper(this) 152 return MoreObjects.toStringHelper(this)
...@@ -180,17 +201,16 @@ public final class MapUpdate<K, V> { ...@@ -180,17 +201,16 @@ public final class MapUpdate<K, V> {
180 } 201 }
181 202
182 public Builder<K, V> withCurrentValue(V value) { 203 public Builder<K, V> withCurrentValue(V value) {
183 - update.currentValue = checkNotNull(value, "currentValue cannot be null"); 204 + update.currentValue = value;
184 return this; 205 return this;
185 } 206 }
186 207
187 public Builder<K, V> withValue(V value) { 208 public Builder<K, V> withValue(V value) {
188 - update.value = checkNotNull(value, "value cannot be null"); 209 + update.value = value;
189 return this; 210 return this;
190 } 211 }
191 212
192 public Builder<K, V> withCurrentVersion(long version) { 213 public Builder<K, V> withCurrentVersion(long version) {
193 - checkArgument(version >= 0, "version cannot be negative");
194 update.currentVersion = version; 214 update.currentVersion = version;
195 return this; 215 return this;
196 } 216 }
......
...@@ -26,6 +26,7 @@ import java.util.function.Function; ...@@ -26,6 +26,7 @@ import java.util.function.Function;
26 import java.util.function.Predicate; 26 import java.util.function.Predicate;
27 27
28 import org.onosproject.store.primitives.DefaultConsistentMap; 28 import org.onosproject.store.primitives.DefaultConsistentMap;
29 +import org.onosproject.store.primitives.TransactionId;
29 30
30 /** 31 /**
31 * A distributed, strongly consistent map whose methods are all executed asynchronously. 32 * A distributed, strongly consistent map whose methods are all executed asynchronously.
...@@ -319,6 +320,37 @@ public interface AsyncConsistentMap<K, V> extends DistributedPrimitive { ...@@ -319,6 +320,37 @@ public interface AsyncConsistentMap<K, V> extends DistributedPrimitive {
319 CompletableFuture<Void> removeListener(MapEventListener<K, V> listener); 320 CompletableFuture<Void> removeListener(MapEventListener<K, V> listener);
320 321
321 /** 322 /**
323 + * Prepares a transaction for commitment.
324 + * @param transaction transaction
325 + * @return {@code true} if prepare is successful and transaction is ready to be committed;
326 + * {@code false} otherwise
327 + */
328 + CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction);
329 +
330 + /**
331 + * Commits a previously prepared transaction.
332 + * @param transactionId transaction identifier
333 + * @return future that will be completed when the operation finishes
334 + */
335 + CompletableFuture<Void> commit(TransactionId transactionId);
336 +
337 + /**
338 + * Aborts a previously prepared transaction.
339 + * @param transactionId transaction identifier
340 + * @return future that will be completed when the operation finishes
341 + */
342 + CompletableFuture<Void> rollback(TransactionId transactionId);
343 +
344 + /**
345 + * Returns a new {@link ConsistentMap} that is backed by this instance.
346 + *
347 + * @return new {@code ConsistentMap} instance
348 + */
349 + default ConsistentMap<K, V> asConsistentMap() {
350 + return asConsistentMap(DistributedPrimitive.DEFAULT_OPERTATION_TIMEOUT_MILLIS);
351 + }
352 +
353 + /**
322 * Returns a new {@link ConsistentMap} that is backed by this instance. 354 * Returns a new {@link ConsistentMap} that is backed by this instance.
323 * 355 *
324 * @param timeoutMillis timeout duration for the returned ConsistentMap operations 356 * @param timeoutMillis timeout duration for the returned ConsistentMap operations
......
1 +/*
2 + * Copyright 2016 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.service;
17 +
18 +import java.util.List;
19 +import java.util.function.Function;
20 +
21 +import org.onosproject.store.primitives.MapUpdate;
22 +import org.onosproject.store.primitives.TransactionId;
23 +
24 +import com.google.common.collect.ImmutableList;
25 +import com.google.common.collect.Lists;
26 +
27 +/**
28 + * Collection of map updates to be committed atomically.
29 + *
30 + * @param <K> key type
31 + * @param <V> value type
32 + */
33 +public class MapTransaction<K, V> {
34 +
35 + private final TransactionId transactionId;
36 + private final List<MapUpdate<K, V>> updates;
37 +
38 + public MapTransaction(TransactionId transactionId, List<MapUpdate<K, V>> updates) {
39 + this.transactionId = transactionId;
40 + this.updates = ImmutableList.copyOf(updates);
41 + }
42 +
43 + /**
44 + * Returns the transaction identifier.
45 + *
46 + * @return transaction id
47 + */
48 + public TransactionId transactionId() {
49 + return transactionId;
50 + }
51 +
52 + /**
53 + * Returns the list of map updates.
54 + *
55 + * @return map updates
56 + */
57 + public List<MapUpdate<K, V>> updates() {
58 + return updates;
59 + }
60 +
61 + /**
62 + * Maps this instance to another {@code MapTransaction} with different key and value types.
63 + *
64 + * @param keyMapper function for mapping key types
65 + * @param valueMapper function for mapping value types
66 + * @return newly typed instance
67 + *
68 + * @param <S> key type of returned instance
69 + * @param <T> value type of returned instance
70 + */
71 + public <S, T> MapTransaction<S, T> map(Function<K, S> keyMapper, Function<V, T> valueMapper) {
72 + return new MapTransaction<>(transactionId, Lists.transform(updates, u -> u.map(keyMapper, valueMapper)));
73 + }
74 +}
...@@ -13,9 +13,10 @@ ...@@ -13,9 +13,10 @@
13 * See the License for the specific language governing permissions and 13 * See the License for the specific language governing permissions and
14 * limitations under the License. 14 * limitations under the License.
15 */ 15 */
16 -package org.onosproject.store.primitives.resources.impl; 16 +package org.onosproject.store.primitives;
17 17
18 import com.google.common.testing.EqualsTester; 18 import com.google.common.testing.EqualsTester;
19 +
19 import org.junit.Test; 20 import org.junit.Test;
20 21
21 import static org.hamcrest.MatcherAssert.assertThat; 22 import static org.hamcrest.MatcherAssert.assertThat;
......
...@@ -32,17 +32,18 @@ import java.util.Scanner; ...@@ -32,17 +32,18 @@ import java.util.Scanner;
32 import org.onlab.util.Match; 32 import org.onlab.util.Match;
33 import org.onosproject.cluster.NodeId; 33 import org.onosproject.cluster.NodeId;
34 import org.onosproject.event.Change; 34 import org.onosproject.event.Change;
35 +import org.onosproject.store.primitives.MapUpdate;
35 import org.onosproject.store.primitives.TransactionId; 36 import org.onosproject.store.primitives.TransactionId;
36 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands; 37 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands;
37 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapState; 38 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapState;
38 import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands; 39 import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands;
39 import org.onosproject.store.primitives.resources.impl.CommitResult; 40 import org.onosproject.store.primitives.resources.impl.CommitResult;
40 import org.onosproject.store.primitives.resources.impl.MapEntryUpdateResult; 41 import org.onosproject.store.primitives.resources.impl.MapEntryUpdateResult;
41 -import org.onosproject.store.primitives.resources.impl.MapUpdate;
42 import org.onosproject.store.primitives.resources.impl.PrepareResult; 42 import org.onosproject.store.primitives.resources.impl.PrepareResult;
43 import org.onosproject.store.primitives.resources.impl.RollbackResult; 43 import org.onosproject.store.primitives.resources.impl.RollbackResult;
44 import org.onosproject.store.serializers.KryoNamespaces; 44 import org.onosproject.store.serializers.KryoNamespaces;
45 import org.onosproject.store.service.MapEvent; 45 import org.onosproject.store.service.MapEvent;
46 +import org.onosproject.store.service.MapTransaction;
46 import org.onosproject.store.service.Versioned; 47 import org.onosproject.store.service.Versioned;
47 48
48 import com.google.common.base.Throwables; 49 import com.google.common.base.Throwables;
...@@ -65,7 +66,7 @@ public final class CatalystSerializers { ...@@ -65,7 +66,7 @@ public final class CatalystSerializers {
65 MapEntryUpdateResult.Status.class, 66 MapEntryUpdateResult.Status.class,
66 MapUpdate.class, 67 MapUpdate.class,
67 MapUpdate.Type.class, 68 MapUpdate.Type.class,
68 - Transaction.class, 69 + MapTransaction.class,
69 Transaction.State.class, 70 Transaction.State.class,
70 TransactionId.class, 71 TransactionId.class,
71 PrepareResult.class, 72 PrepareResult.class,
...@@ -99,7 +100,7 @@ public final class CatalystSerializers { ...@@ -99,7 +100,7 @@ public final class CatalystSerializers {
99 serializer.register(Match.class, factory); 100 serializer.register(Match.class, factory);
100 serializer.register(MapEntryUpdateResult.class, factory); 101 serializer.register(MapEntryUpdateResult.class, factory);
101 serializer.register(MapEntryUpdateResult.Status.class, factory); 102 serializer.register(MapEntryUpdateResult.Status.class, factory);
102 - serializer.register(Transaction.class, factory); 103 + serializer.register(MapTransaction.class, factory);
103 serializer.register(Transaction.State.class, factory); 104 serializer.register(Transaction.State.class, factory);
104 serializer.register(PrepareResult.class, factory); 105 serializer.register(PrepareResult.class, factory);
105 serializer.register(CommitResult.class, factory); 106 serializer.register(CommitResult.class, factory);
......
...@@ -65,8 +65,8 @@ import org.onosproject.cluster.PartitionId; ...@@ -65,8 +65,8 @@ import org.onosproject.cluster.PartitionId;
65 import org.onosproject.core.ApplicationId; 65 import org.onosproject.core.ApplicationId;
66 import org.onosproject.persistence.PersistenceService; 66 import org.onosproject.persistence.PersistenceService;
67 import org.onosproject.store.cluster.messaging.ClusterCommunicationService; 67 import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
68 +import org.onosproject.store.primitives.MapUpdate;
68 import org.onosproject.store.primitives.TransactionId; 69 import org.onosproject.store.primitives.TransactionId;
69 -import org.onosproject.store.primitives.resources.impl.MapUpdate;
70 import org.onosproject.store.serializers.KryoNamespaces; 70 import org.onosproject.store.serializers.KryoNamespaces;
71 import org.onosproject.store.service.AsyncConsistentMap; 71 import org.onosproject.store.service.AsyncConsistentMap;
72 import org.onosproject.store.service.AtomicCounterBuilder; 72 import org.onosproject.store.service.AtomicCounterBuilder;
......
...@@ -21,8 +21,8 @@ import java.nio.ByteBuffer; ...@@ -21,8 +21,8 @@ import java.nio.ByteBuffer;
21 import org.onlab.util.KryoNamespace; 21 import org.onlab.util.KryoNamespace;
22 import org.onlab.util.Match; 22 import org.onlab.util.Match;
23 import org.onosproject.cluster.NodeId; 23 import org.onosproject.cluster.NodeId;
24 +import org.onosproject.store.primitives.MapUpdate;
24 import org.onosproject.store.primitives.TransactionId; 25 import org.onosproject.store.primitives.TransactionId;
25 -import org.onosproject.store.primitives.resources.impl.MapUpdate;
26 import org.onosproject.store.serializers.KryoNamespaces; 26 import org.onosproject.store.serializers.KryoNamespaces;
27 import org.onosproject.store.serializers.KryoSerializer; 27 import org.onosproject.store.serializers.KryoSerializer;
28 import org.onosproject.store.service.Versioned; 28 import org.onosproject.store.service.Versioned;
......
...@@ -40,11 +40,13 @@ import org.onlab.util.Match; ...@@ -40,11 +40,13 @@ import org.onlab.util.Match;
40 import org.onlab.util.SharedExecutors; 40 import org.onlab.util.SharedExecutors;
41 import org.onlab.util.Tools; 41 import org.onlab.util.Tools;
42 import org.onosproject.core.ApplicationId; 42 import org.onosproject.core.ApplicationId;
43 +import org.onosproject.store.primitives.TransactionId;
43 import org.onosproject.store.service.AsyncConsistentMap; 44 import org.onosproject.store.service.AsyncConsistentMap;
44 import org.onosproject.store.service.ConsistentMapException; 45 import org.onosproject.store.service.ConsistentMapException;
45 import org.onosproject.store.service.ConsistentMapException.ConcurrentModification; 46 import org.onosproject.store.service.ConsistentMapException.ConcurrentModification;
46 import org.onosproject.store.service.MapEvent; 47 import org.onosproject.store.service.MapEvent;
47 import org.onosproject.store.service.MapEventListener; 48 import org.onosproject.store.service.MapEventListener;
49 +import org.onosproject.store.service.MapTransaction;
48 import org.onosproject.store.service.Serializer; 50 import org.onosproject.store.service.Serializer;
49 import org.onosproject.store.service.Versioned; 51 import org.onosproject.store.service.Versioned;
50 import org.onosproject.utils.MeteringAgent; 52 import org.onosproject.utils.MeteringAgent;
...@@ -491,6 +493,21 @@ public class DefaultAsyncConsistentMap<K, V> implements AsyncConsistentMap<K, V ...@@ -491,6 +493,21 @@ public class DefaultAsyncConsistentMap<K, V> implements AsyncConsistentMap<K, V
491 return CompletableFuture.completedFuture(null); 493 return CompletableFuture.completedFuture(null);
492 } 494 }
493 495
496 + @Override
497 + public CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction) {
498 + return Tools.exceptionalFuture(new UnsupportedOperationException());
499 + }
500 +
501 + @Override
502 + public CompletableFuture<Void> commit(TransactionId transactionId) {
503 + return Tools.exceptionalFuture(new UnsupportedOperationException());
504 + }
505 +
506 + @Override
507 + public CompletableFuture<Void> rollback(TransactionId transactionId) {
508 + return Tools.exceptionalFuture(new UnsupportedOperationException());
509 + }
510 +
494 protected void notifyListeners(MapEvent<K, V> event) { 511 protected void notifyListeners(MapEvent<K, V> event) {
495 if (event == null) { 512 if (event == null) {
496 return; 513 return;
......
...@@ -26,8 +26,8 @@ import net.kuujo.copycat.state.Initializer; ...@@ -26,8 +26,8 @@ import net.kuujo.copycat.state.Initializer;
26 import net.kuujo.copycat.state.StateContext; 26 import net.kuujo.copycat.state.StateContext;
27 27
28 import org.onlab.util.Match; 28 import org.onlab.util.Match;
29 +import org.onosproject.store.primitives.MapUpdate;
29 import org.onosproject.store.primitives.TransactionId; 30 import org.onosproject.store.primitives.TransactionId;
30 -import org.onosproject.store.primitives.resources.impl.MapUpdate;
31 import org.onosproject.store.service.Versioned; 31 import org.onosproject.store.service.Versioned;
32 32
33 import java.util.Arrays; 33 import java.util.Arrays;
......
...@@ -24,9 +24,9 @@ import java.util.function.Supplier; ...@@ -24,9 +24,9 @@ import java.util.function.Supplier;
24 24
25 import static com.google.common.base.Preconditions.*; 25 import static com.google.common.base.Preconditions.*;
26 26
27 +import org.onosproject.store.primitives.MapUpdate;
27 import org.onosproject.store.primitives.TransactionId; 28 import org.onosproject.store.primitives.TransactionId;
28 import org.onosproject.store.primitives.resources.impl.CommitResult; 29 import org.onosproject.store.primitives.resources.impl.CommitResult;
29 -import org.onosproject.store.primitives.resources.impl.MapUpdate;
30 import org.onosproject.store.service.ConsistentMapBuilder; 30 import org.onosproject.store.service.ConsistentMapBuilder;
31 import org.onosproject.store.service.Serializer; 31 import org.onosproject.store.service.Serializer;
32 import org.onosproject.store.service.TransactionContext; 32 import org.onosproject.store.service.TransactionContext;
...@@ -84,7 +84,10 @@ public class DefaultTransactionContext implements TransactionContext { ...@@ -84,7 +84,10 @@ public class DefaultTransactionContext implements TransactionContext {
84 checkNotNull(serializer); 84 checkNotNull(serializer);
85 return txMaps.computeIfAbsent(mapName, name -> new DefaultTransactionalMap<>( 85 return txMaps.computeIfAbsent(mapName, name -> new DefaultTransactionalMap<>(
86 name, 86 name,
87 - mapBuilderSupplier.get().withName(name).withSerializer(serializer).build(), 87 + mapBuilderSupplier.get()
88 + .withName(name)
89 + .withSerializer(serializer)
90 + .buildAsyncMap(),
88 this, 91 this,
89 serializer)); 92 serializer));
90 } 93 }
...@@ -113,7 +116,7 @@ public class DefaultTransactionContext implements TransactionContext { ...@@ -113,7 +116,7 @@ public class DefaultTransactionContext implements TransactionContext {
113 public void abort() { 116 public void abort() {
114 if (isOpen) { 117 if (isOpen) {
115 try { 118 try {
116 - txMaps.values().forEach(m -> m.rollback()); 119 + txMaps.values().forEach(m -> m.abort());
117 } finally { 120 } finally {
118 isOpen = false; 121 isOpen = false;
119 } 122 }
......
...@@ -19,10 +19,13 @@ package org.onosproject.store.primitives.impl; ...@@ -19,10 +19,13 @@ package org.onosproject.store.primitives.impl;
19 import java.util.List; 19 import java.util.List;
20 import java.util.Map; 20 import java.util.Map;
21 import java.util.Set; 21 import java.util.Set;
22 +import java.util.concurrent.CompletableFuture;
22 23
23 import org.onlab.util.HexString; 24 import org.onlab.util.HexString;
24 -import org.onosproject.store.primitives.resources.impl.MapUpdate; 25 +import org.onosproject.store.primitives.MapUpdate;
26 +import org.onosproject.store.service.AsyncConsistentMap;
25 import org.onosproject.store.service.ConsistentMap; 27 import org.onosproject.store.service.ConsistentMap;
28 +import org.onosproject.store.service.MapTransaction;
26 import org.onosproject.store.service.Serializer; 29 import org.onosproject.store.service.Serializer;
27 import org.onosproject.store.service.TransactionContext; 30 import org.onosproject.store.service.TransactionContext;
28 import org.onosproject.store.service.TransactionalMap; 31 import org.onosproject.store.service.TransactionalMap;
...@@ -46,11 +49,12 @@ import com.google.common.collect.Sets; ...@@ -46,11 +49,12 @@ import com.google.common.collect.Sets;
46 * @param <K> key type 49 * @param <K> key type
47 * @param <V> value type. 50 * @param <V> value type.
48 */ 51 */
49 -public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> { 52 +public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V>, TransactionParticipant {
50 53
51 private final TransactionContext txContext; 54 private final TransactionContext txContext;
52 private static final String TX_CLOSED_ERROR = "Transaction is closed"; 55 private static final String TX_CLOSED_ERROR = "Transaction is closed";
53 - private final ConsistentMap<K, V> backingMap; 56 + private final AsyncConsistentMap<K, V> backingMap;
57 + private final ConsistentMap<K, V> backingConsitentMap;
54 private final String name; 58 private final String name;
55 private final Serializer serializer; 59 private final Serializer serializer;
56 private final Map<K, Versioned<V>> readCache = Maps.newConcurrentMap(); 60 private final Map<K, Versioned<V>> readCache = Maps.newConcurrentMap();
...@@ -76,11 +80,12 @@ public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> { ...@@ -76,11 +80,12 @@ public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> {
76 80
77 public DefaultTransactionalMap( 81 public DefaultTransactionalMap(
78 String name, 82 String name,
79 - ConsistentMap<K, V> backingMap, 83 + AsyncConsistentMap<K, V> backingMap,
80 TransactionContext txContext, 84 TransactionContext txContext,
81 Serializer serializer) { 85 Serializer serializer) {
82 this.name = name; 86 this.name = name;
83 this.backingMap = backingMap; 87 this.backingMap = backingMap;
88 + this.backingConsitentMap = backingMap.asConsistentMap();
84 this.txContext = txContext; 89 this.txContext = txContext;
85 this.serializer = serializer; 90 this.serializer = serializer;
86 } 91 }
...@@ -96,7 +101,7 @@ public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> { ...@@ -96,7 +101,7 @@ public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> {
96 if (latest != null) { 101 if (latest != null) {
97 return latest; 102 return latest;
98 } else { 103 } else {
99 - Versioned<V> v = readCache.computeIfAbsent(key, k -> backingMap.get(k)); 104 + Versioned<V> v = readCache.computeIfAbsent(key, k -> backingConsitentMap.get(k));
100 return v != null ? v.value() : null; 105 return v != null ? v.value() : null;
101 } 106 }
102 } 107 }
...@@ -159,6 +164,62 @@ public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> { ...@@ -159,6 +164,62 @@ public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> {
159 return latest; 164 return latest;
160 } 165 }
161 166
167 + @Override
168 + public CompletableFuture<Boolean> prepare() {
169 + return backingMap.prepare(new MapTransaction<>(txContext.transactionId(), updates()));
170 + }
171 +
172 + @Override
173 + public CompletableFuture<Void> commit() {
174 + return backingMap.commit(txContext.transactionId());
175 + }
176 +
177 + @Override
178 + public CompletableFuture<Void> rollback() {
179 + return backingMap.rollback(txContext.transactionId());
180 + }
181 +
182 + @Override
183 + public boolean hasPendingUpdates() {
184 + return updates().size() > 0;
185 + }
186 +
187 + protected List<MapUpdate<K, V>> updates() {
188 + List<MapUpdate<K, V>> updates = Lists.newLinkedList();
189 + deleteSet.forEach(key -> {
190 + Versioned<V> original = readCache.get(key);
191 + if (original != null) {
192 + updates.add(MapUpdate.<K, V>newBuilder()
193 + .withMapName(name)
194 + .withType(MapUpdate.Type.REMOVE_IF_VERSION_MATCH)
195 + .withKey(key)
196 + .withCurrentVersion(original.version())
197 + .build());
198 + }
199 + });
200 + writeCache.forEach((key, value) -> {
201 + Versioned<V> original = readCache.get(key);
202 + if (original == null) {
203 + updates.add(MapUpdate.<K, V>newBuilder()
204 + .withMapName(name)
205 + .withType(MapUpdate.Type.PUT_IF_ABSENT)
206 + .withKey(key)
207 + .withValue(value)
208 + .build());
209 + } else {
210 + updates.add(MapUpdate.<K, V>newBuilder()
211 + .withMapName(name)
212 + .withType(MapUpdate.Type.PUT_IF_VERSION_MATCH)
213 + .withKey(key)
214 + .withCurrentVersion(original.version())
215 + .withValue(value)
216 + .build());
217 + }
218 + });
219 + return updates;
220 + }
221 +
222 +
162 protected List<MapUpdate<String, byte[]>> toMapUpdates() { 223 protected List<MapUpdate<String, byte[]>> toMapUpdates() {
163 List<MapUpdate<String, byte[]>> updates = Lists.newLinkedList(); 224 List<MapUpdate<String, byte[]>> updates = Lists.newLinkedList();
164 deleteSet.forEach(key -> { 225 deleteSet.forEach(key -> {
...@@ -194,19 +255,18 @@ public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> { ...@@ -194,19 +255,18 @@ public class DefaultTransactionalMap<K, V> implements TransactionalMap<K, V> {
194 return updates; 255 return updates;
195 } 256 }
196 257
197 - // TODO: build expected result Map processing DB updates?
198 @Override 258 @Override
199 public String toString() { 259 public String toString() {
200 return MoreObjects.toStringHelper(this) 260 return MoreObjects.toStringHelper(this)
201 .add("backingMap", backingMap) 261 .add("backingMap", backingMap)
202 - .add("updates", toMapUpdates()) 262 + .add("updates", updates())
203 .toString(); 263 .toString();
204 } 264 }
205 265
206 /** 266 /**
207 * Discards all changes made to this transactional map. 267 * Discards all changes made to this transactional map.
208 */ 268 */
209 - protected void rollback() { 269 + protected void abort() {
210 readCache.clear(); 270 readCache.clear();
211 writeCache.clear(); 271 writeCache.clear();
212 deleteSet.clear(); 272 deleteSet.clear();
......
...@@ -27,8 +27,10 @@ import java.util.function.BiFunction; ...@@ -27,8 +27,10 @@ import java.util.function.BiFunction;
27 import java.util.function.Predicate; 27 import java.util.function.Predicate;
28 28
29 import org.onosproject.core.ApplicationId; 29 import org.onosproject.core.ApplicationId;
30 +import org.onosproject.store.primitives.TransactionId;
30 import org.onosproject.store.service.AsyncConsistentMap; 31 import org.onosproject.store.service.AsyncConsistentMap;
31 import org.onosproject.store.service.MapEventListener; 32 import org.onosproject.store.service.MapEventListener;
33 +import org.onosproject.store.service.MapTransaction;
32 import org.onosproject.store.service.Versioned; 34 import org.onosproject.store.service.Versioned;
33 35
34 import com.google.common.base.MoreObjects; 36 import com.google.common.base.MoreObjects;
...@@ -161,6 +163,21 @@ public class DelegatingAsyncConsistentMap<K, V> implements AsyncConsistentMap<K, ...@@ -161,6 +163,21 @@ public class DelegatingAsyncConsistentMap<K, V> implements AsyncConsistentMap<K,
161 } 163 }
162 164
163 @Override 165 @Override
166 + public CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction) {
167 + return delegateMap.prepare(transaction);
168 + }
169 +
170 + @Override
171 + public CompletableFuture<Void> commit(TransactionId transactionId) {
172 + return delegateMap.commit(transactionId);
173 + }
174 +
175 + @Override
176 + public CompletableFuture<Void> rollback(TransactionId transactionId) {
177 + return delegateMap.rollback(transactionId);
178 + }
179 +
180 + @Override
164 public String toString() { 181 public String toString() {
165 return MoreObjects.toStringHelper(getClass()) 182 return MoreObjects.toStringHelper(getClass())
166 .add("delegateMap", delegateMap) 183 .add("delegateMap", delegateMap)
......
...@@ -28,10 +28,15 @@ import java.util.concurrent.atomic.AtomicBoolean; ...@@ -28,10 +28,15 @@ import java.util.concurrent.atomic.AtomicBoolean;
28 import java.util.concurrent.atomic.AtomicInteger; 28 import java.util.concurrent.atomic.AtomicInteger;
29 import java.util.function.BiFunction; 29 import java.util.function.BiFunction;
30 import java.util.function.Predicate; 30 import java.util.function.Predicate;
31 +import java.util.stream.Collectors;
31 32
33 +import org.onlab.util.Tools;
32 import org.onosproject.cluster.PartitionId; 34 import org.onosproject.cluster.PartitionId;
35 +import org.onosproject.store.primitives.MapUpdate;
36 +import org.onosproject.store.primitives.TransactionId;
33 import org.onosproject.store.service.AsyncConsistentMap; 37 import org.onosproject.store.service.AsyncConsistentMap;
34 import org.onosproject.store.service.MapEventListener; 38 import org.onosproject.store.service.MapEventListener;
39 +import org.onosproject.store.service.MapTransaction;
35 import org.onosproject.store.service.Versioned; 40 import org.onosproject.store.service.Versioned;
36 41
37 import com.google.common.collect.Lists; 42 import com.google.common.collect.Lists;
...@@ -198,6 +203,39 @@ public class PartitionedAsyncConsistentMap<K, V> implements AsyncConsistentMap<K ...@@ -198,6 +203,39 @@ public class PartitionedAsyncConsistentMap<K, V> implements AsyncConsistentMap<K
198 .toArray(CompletableFuture[]::new)); 203 .toArray(CompletableFuture[]::new));
199 } 204 }
200 205
206 + @Override
207 + public CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction) {
208 +
209 + Map<AsyncConsistentMap<K, V>, List<MapUpdate<K, V>>> updatesGroupedByMap = Maps.newIdentityHashMap();
210 + transaction.updates().forEach(update -> {
211 + AsyncConsistentMap<K, V> map = getMap(update.key());
212 + updatesGroupedByMap.computeIfAbsent(map, k -> Lists.newLinkedList()).add(update);
213 + });
214 + Map<AsyncConsistentMap<K, V>, MapTransaction<K, V>> transactionsByMap =
215 + Maps.transformValues(updatesGroupedByMap,
216 + list -> new MapTransaction<>(transaction.transactionId(), list));
217 +
218 + return Tools.allOf(transactionsByMap.entrySet()
219 + .stream()
220 + .map(e -> e.getKey().prepare(e.getValue()))
221 + .collect(Collectors.toList()))
222 + .thenApply(list -> list.stream().reduce(Boolean::logicalAnd).orElse(true));
223 + }
224 +
225 + @Override
226 + public CompletableFuture<Void> commit(TransactionId transactionId) {
227 + return CompletableFuture.allOf(getMaps().stream()
228 + .map(e -> e.commit(transactionId))
229 + .toArray(CompletableFuture[]::new));
230 + }
231 +
232 + @Override
233 + public CompletableFuture<Void> rollback(TransactionId transactionId) {
234 + return CompletableFuture.allOf(getMaps().stream()
235 + .map(e -> e.rollback(transactionId))
236 + .toArray(CompletableFuture[]::new));
237 + }
238 +
201 /** 239 /**
202 * Returns the map (partition) to which the specified key maps. 240 * Returns the map (partition) to which the specified key maps.
203 * @param key key 241 * @param key key
......
...@@ -26,8 +26,8 @@ import net.kuujo.copycat.cluster.Cluster; ...@@ -26,8 +26,8 @@ import net.kuujo.copycat.cluster.Cluster;
26 import net.kuujo.copycat.resource.ResourceState; 26 import net.kuujo.copycat.resource.ResourceState;
27 27
28 import org.onlab.util.Match; 28 import org.onlab.util.Match;
29 +import org.onosproject.store.primitives.MapUpdate;
29 import org.onosproject.store.primitives.resources.impl.CommitResult; 30 import org.onosproject.store.primitives.resources.impl.CommitResult;
30 -import org.onosproject.store.primitives.resources.impl.MapUpdate;
31 import org.onosproject.store.service.Versioned; 31 import org.onosproject.store.service.Versioned;
32 32
33 import java.util.Collection; 33 import java.util.Collection;
......
...@@ -17,8 +17,8 @@ package org.onosproject.store.primitives.impl; ...@@ -17,8 +17,8 @@ package org.onosproject.store.primitives.impl;
17 17
18 import java.util.List; 18 import java.util.List;
19 19
20 +import org.onosproject.store.primitives.MapUpdate;
20 import org.onosproject.store.primitives.TransactionId; 21 import org.onosproject.store.primitives.TransactionId;
21 -import org.onosproject.store.primitives.resources.impl.MapUpdate;
22 22
23 import com.google.common.base.MoreObjects; 23 import com.google.common.base.MoreObjects;
24 import com.google.common.collect.ImmutableList; 24 import com.google.common.collect.ImmutableList;
......
...@@ -17,34 +17,32 @@ package org.onosproject.store.primitives.impl; ...@@ -17,34 +17,32 @@ package org.onosproject.store.primitives.impl;
17 17
18 import java.util.concurrent.CompletableFuture; 18 import java.util.concurrent.CompletableFuture;
19 19
20 -import org.onosproject.store.primitives.TransactionId;
21 -import org.onosproject.store.primitives.resources.impl.CommitResult;
22 -import org.onosproject.store.primitives.resources.impl.PrepareResult;
23 -import org.onosproject.store.primitives.resources.impl.RollbackResult;
24 -
25 /** 20 /**
26 * Participant in a two-phase commit protocol. 21 * Participant in a two-phase commit protocol.
27 */ 22 */
28 public interface TransactionParticipant { 23 public interface TransactionParticipant {
29 24
30 /** 25 /**
31 - * Attempts to execute the prepare phase for the specified {@link Transaction transaction}. 26 + * Returns if this participant has updates that need to be committed.
32 - * @param transaction transaction 27 + * @return {@code true} if yes; {@code false} otherwise
33 - * @return future for prepare result 28 + */
29 + boolean hasPendingUpdates();
30 +
31 + /**
32 + * Executes the prepare phase.
33 + * @return {@code true} is successful; {@code false} otherwise
34 */ 34 */
35 - CompletableFuture<PrepareResult> prepare(Transaction transaction); 35 + CompletableFuture<Boolean> prepare();
36 36
37 /** 37 /**
38 * Attempts to execute the commit phase for previously prepared transaction. 38 * Attempts to execute the commit phase for previously prepared transaction.
39 - * @param transactionId transaction identifier 39 + * @return future that is completed when the operation completes
40 - * @return future for commit result
41 */ 40 */
42 - CompletableFuture<CommitResult> commit(TransactionId transactionId); 41 + CompletableFuture<Void> commit();
43 42
44 /** 43 /**
45 * Attempts to execute the rollback phase for previously prepared transaction. 44 * Attempts to execute the rollback phase for previously prepared transaction.
46 - * @param transactionId transaction identifier 45 + * @return future that is completed when the operation completes
47 - * @return future for rollback result
48 */ 46 */
49 - CompletableFuture<RollbackResult> rollback(TransactionId transactionId); 47 + CompletableFuture<Void> rollback();
50 } 48 }
...\ No newline at end of file ...\ No newline at end of file
......
...@@ -26,9 +26,11 @@ import java.util.function.Function; ...@@ -26,9 +26,11 @@ import java.util.function.Function;
26 import java.util.function.Predicate; 26 import java.util.function.Predicate;
27 import java.util.stream.Collectors; 27 import java.util.stream.Collectors;
28 28
29 +import org.onosproject.store.primitives.TransactionId;
29 import org.onosproject.store.service.AsyncConsistentMap; 30 import org.onosproject.store.service.AsyncConsistentMap;
30 import org.onosproject.store.service.MapEvent; 31 import org.onosproject.store.service.MapEvent;
31 import org.onosproject.store.service.MapEventListener; 32 import org.onosproject.store.service.MapEventListener;
33 +import org.onosproject.store.service.MapTransaction;
32 import org.onosproject.store.service.Versioned; 34 import org.onosproject.store.service.Versioned;
33 35
34 import com.google.common.collect.Maps; 36 import com.google.common.collect.Maps;
...@@ -196,6 +198,21 @@ public class TranscodingAsyncConsistentMap<K1, V1, K2, V2> implements AsyncConsi ...@@ -196,6 +198,21 @@ public class TranscodingAsyncConsistentMap<K1, V1, K2, V2> implements AsyncConsi
196 } 198 }
197 } 199 }
198 200
201 + @Override
202 + public CompletableFuture<Boolean> prepare(MapTransaction<K1, V1> transaction) {
203 + return backingMap.prepare(transaction.map(keyEncoder, valueEncoder));
204 + }
205 +
206 + @Override
207 + public CompletableFuture<Void> commit(TransactionId transactionId) {
208 + return backingMap.commit(transactionId);
209 + }
210 +
211 + @Override
212 + public CompletableFuture<Void> rollback(TransactionId transactionId) {
213 + return backingMap.rollback(transactionId);
214 + }
215 +
199 private class InternalBackingMapEventListener implements MapEventListener<K2, V2> { 216 private class InternalBackingMapEventListener implements MapEventListener<K2, V2> {
200 217
201 private final MapEventListener<K1, V1> listener; 218 private final MapEventListener<K1, V1> listener;
......
...@@ -32,11 +32,10 @@ import java.util.function.Predicate; ...@@ -32,11 +32,10 @@ import java.util.function.Predicate;
32 32
33 import org.onlab.util.Match; 33 import org.onlab.util.Match;
34 import org.onosproject.store.primitives.TransactionId; 34 import org.onosproject.store.primitives.TransactionId;
35 -import org.onosproject.store.primitives.impl.Transaction;
36 -import org.onosproject.store.primitives.impl.TransactionParticipant;
37 import org.onosproject.store.service.AsyncConsistentMap; 35 import org.onosproject.store.service.AsyncConsistentMap;
38 import org.onosproject.store.service.MapEvent; 36 import org.onosproject.store.service.MapEvent;
39 import org.onosproject.store.service.MapEventListener; 37 import org.onosproject.store.service.MapEventListener;
38 +import org.onosproject.store.service.MapTransaction;
40 import org.onosproject.store.service.Versioned; 39 import org.onosproject.store.service.Versioned;
41 40
42 import com.google.common.collect.Sets; 41 import com.google.common.collect.Sets;
...@@ -46,7 +45,7 @@ import com.google.common.collect.Sets; ...@@ -46,7 +45,7 @@ import com.google.common.collect.Sets;
46 */ 45 */
47 @ResourceTypeInfo(id = -151, stateMachine = AtomixConsistentMapState.class) 46 @ResourceTypeInfo(id = -151, stateMachine = AtomixConsistentMapState.class)
48 public class AtomixConsistentMap extends Resource<AtomixConsistentMap, Resource.Options> 47 public class AtomixConsistentMap extends Resource<AtomixConsistentMap, Resource.Options>
49 - implements AsyncConsistentMap<String, byte[]>, TransactionParticipant { 48 + implements AsyncConsistentMap<String, byte[]> {
50 49
51 private final Set<MapEventListener<String, byte[]>> mapEventListeners = Sets.newCopyOnWriteArraySet(); 50 private final Set<MapEventListener<String, byte[]>> mapEventListeners = Sets.newCopyOnWriteArraySet();
52 51
...@@ -266,18 +265,21 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap, Resource. ...@@ -266,18 +265,21 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap, Resource.
266 } 265 }
267 266
268 @Override 267 @Override
269 - public CompletableFuture<PrepareResult> prepare(Transaction transaction) { 268 + public CompletableFuture<Boolean> prepare(MapTransaction<String, byte[]> transaction) {
270 - return submit(new AtomixConsistentMapCommands.TransactionPrepare(transaction)); 269 + return submit(new AtomixConsistentMapCommands.TransactionPrepare(transaction))
270 + .thenApply(v -> v == PrepareResult.OK);
271 } 271 }
272 272
273 @Override 273 @Override
274 - public CompletableFuture<CommitResult> commit(TransactionId transactionId) { 274 + public CompletableFuture<Void> commit(TransactionId transactionId) {
275 - return submit(new AtomixConsistentMapCommands.TransactionCommit(transactionId)); 275 + return submit(new AtomixConsistentMapCommands.TransactionCommit(transactionId))
276 + .thenApply(v -> null);
276 } 277 }
277 278
278 @Override 279 @Override
279 - public CompletableFuture<RollbackResult> rollback(TransactionId transactionId) { 280 + public CompletableFuture<Void> rollback(TransactionId transactionId) {
280 - return submit(new AtomixConsistentMapCommands.TransactionRollback(transactionId)); 281 + return submit(new AtomixConsistentMapCommands.TransactionRollback(transactionId))
282 + .thenApply(v -> null);
281 } 283 }
282 284
283 /** 285 /**
......
...@@ -29,7 +29,7 @@ import java.util.Set; ...@@ -29,7 +29,7 @@ import java.util.Set;
29 29
30 import org.onlab.util.Match; 30 import org.onlab.util.Match;
31 import org.onosproject.store.primitives.TransactionId; 31 import org.onosproject.store.primitives.TransactionId;
32 -import org.onosproject.store.primitives.impl.Transaction; 32 +import org.onosproject.store.service.MapTransaction;
33 import org.onosproject.store.service.Versioned; 33 import org.onosproject.store.service.Versioned;
34 34
35 import com.google.common.base.MoreObjects; 35 import com.google.common.base.MoreObjects;
...@@ -209,35 +209,35 @@ public final class AtomixConsistentMapCommands { ...@@ -209,35 +209,35 @@ public final class AtomixConsistentMapCommands {
209 */ 209 */
210 @SuppressWarnings("serial") 210 @SuppressWarnings("serial")
211 public static class TransactionPrepare extends MapCommand<PrepareResult> { 211 public static class TransactionPrepare extends MapCommand<PrepareResult> {
212 - private Transaction transaction; 212 + private MapTransaction<String, byte[]> mapTransaction;
213 213
214 public TransactionPrepare() { 214 public TransactionPrepare() {
215 } 215 }
216 216
217 - public TransactionPrepare(Transaction transaction) { 217 + public TransactionPrepare(MapTransaction<String, byte[]> mapTransaction) {
218 - this.transaction = transaction; 218 + this.mapTransaction = mapTransaction;
219 } 219 }
220 220
221 - public Transaction transaction() { 221 + public MapTransaction<String, byte[]> transaction() {
222 - return transaction; 222 + return mapTransaction;
223 } 223 }
224 224
225 @Override 225 @Override
226 public void writeObject(BufferOutput<?> buffer, Serializer serializer) { 226 public void writeObject(BufferOutput<?> buffer, Serializer serializer) {
227 super.writeObject(buffer, serializer); 227 super.writeObject(buffer, serializer);
228 - serializer.writeObject(transaction, buffer); 228 + serializer.writeObject(mapTransaction, buffer);
229 } 229 }
230 230
231 @Override 231 @Override
232 public void readObject(BufferInput<?> buffer, Serializer serializer) { 232 public void readObject(BufferInput<?> buffer, Serializer serializer) {
233 super.readObject(buffer, serializer); 233 super.readObject(buffer, serializer);
234 - transaction = serializer.readObject(buffer); 234 + mapTransaction = serializer.readObject(buffer);
235 } 235 }
236 236
237 @Override 237 @Override
238 public String toString() { 238 public String toString() {
239 return MoreObjects.toStringHelper(getClass()) 239 return MoreObjects.toStringHelper(getClass())
240 - .add("transaction", transaction) 240 + .add("mapTransaction", mapTransaction)
241 .toString(); 241 .toString();
242 } 242 }
243 } 243 }
......
...@@ -37,10 +37,11 @@ import java.util.stream.Collectors; ...@@ -37,10 +37,11 @@ import java.util.stream.Collectors;
37 37
38 import org.onlab.util.CountDownCompleter; 38 import org.onlab.util.CountDownCompleter;
39 import org.onlab.util.Match; 39 import org.onlab.util.Match;
40 +import org.onosproject.store.primitives.MapUpdate;
40 import org.onosproject.store.primitives.TransactionId; 41 import org.onosproject.store.primitives.TransactionId;
41 -import org.onosproject.store.primitives.impl.Transaction;
42 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionPrepare; 42 import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionPrepare;
43 import org.onosproject.store.service.MapEvent; 43 import org.onosproject.store.service.MapEvent;
44 +import org.onosproject.store.service.MapTransaction;
44 import org.onosproject.store.service.Versioned; 45 import org.onosproject.store.service.Versioned;
45 46
46 import com.google.common.collect.Maps; 47 import com.google.common.collect.Maps;
...@@ -384,7 +385,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements ...@@ -384,7 +385,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements
384 Commit<? extends AtomixConsistentMapCommands.TransactionPrepare> commit) { 385 Commit<? extends AtomixConsistentMapCommands.TransactionPrepare> commit) {
385 boolean ok = false; 386 boolean ok = false;
386 try { 387 try {
387 - Transaction transaction = commit.operation().transaction(); 388 + MapTransaction<String, byte[]> transaction = commit.operation().transaction();
388 for (MapUpdate<String, byte[]> update : transaction.updates()) { 389 for (MapUpdate<String, byte[]> update : transaction.updates()) {
389 String key = update.key(); 390 String key = update.key();
390 if (preparedKeys.contains(key)) { 391 if (preparedKeys.contains(key)) {
...@@ -404,7 +405,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements ...@@ -404,7 +405,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements
404 // No violations detected. Add to pendingTranctions and mark 405 // No violations detected. Add to pendingTranctions and mark
405 // modified keys as 406 // modified keys as
406 // currently locked to updates. 407 // currently locked to updates.
407 - pendingTransactions.put(transaction.id(), commit); 408 + pendingTransactions.put(transaction.transactionId(), commit);
408 transaction.updates().forEach(u -> preparedKeys.add(u.key())); 409 transaction.updates().forEach(u -> preparedKeys.add(u.key()));
409 ok = true; 410 ok = true;
410 return PrepareResult.OK; 411 return PrepareResult.OK;
...@@ -430,7 +431,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements ...@@ -430,7 +431,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements
430 if (prepareCommit == null) { 431 if (prepareCommit == null) {
431 return CommitResult.UNKNOWN_TRANSACTION_ID; 432 return CommitResult.UNKNOWN_TRANSACTION_ID;
432 } 433 }
433 - Transaction transaction = prepareCommit.operation().transaction(); 434 + MapTransaction<String, byte[]> transaction = prepareCommit.operation().transaction();
434 long totalReferencesToCommit = transaction 435 long totalReferencesToCommit = transaction
435 .updates() 436 .updates()
436 .stream() 437 .stream()
...@@ -610,7 +611,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements ...@@ -610,7 +611,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements
610 611
611 @Override 612 @Override
612 public byte[] value() { 613 public byte[] value() {
613 - Transaction transaction = completer.object().operation().transaction(); 614 + MapTransaction<String, byte[]> transaction = completer.object().operation().transaction();
614 return valueForKey(key, transaction); 615 return valueForKey(key, transaction);
615 } 616 }
616 617
...@@ -624,7 +625,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements ...@@ -624,7 +625,7 @@ public class AtomixConsistentMapState extends ResourceStateMachine implements
624 completer.countDown(); 625 completer.countDown();
625 } 626 }
626 627
627 - private byte[] valueForKey(String key, Transaction transaction) { 628 + private byte[] valueForKey(String key, MapTransaction<String, byte[]> transaction) {
628 MapUpdate<String, byte[]> update = transaction.updates() 629 MapUpdate<String, byte[]> update = transaction.updates()
629 .stream() 630 .stream()
630 .filter(u -> u.key().equals(key)) 631 .filter(u -> u.key().equals(key))
......
...@@ -27,10 +27,11 @@ import java.util.stream.Collectors; ...@@ -27,10 +27,11 @@ import java.util.stream.Collectors;
27 import org.junit.Ignore; 27 import org.junit.Ignore;
28 import org.junit.Test; 28 import org.junit.Test;
29 import org.onlab.util.Tools; 29 import org.onlab.util.Tools;
30 +import org.onosproject.store.primitives.MapUpdate;
30 import org.onosproject.store.primitives.TransactionId; 31 import org.onosproject.store.primitives.TransactionId;
31 -import org.onosproject.store.primitives.impl.Transaction;
32 import org.onosproject.store.service.MapEvent; 32 import org.onosproject.store.service.MapEvent;
33 import org.onosproject.store.service.MapEventListener; 33 import org.onosproject.store.service.MapEventListener;
34 +import org.onosproject.store.service.MapTransaction;
34 import org.onosproject.store.service.Versioned; 35 import org.onosproject.store.service.Versioned;
35 36
36 import com.google.common.collect.Sets; 37 import com.google.common.collect.Sets;
...@@ -353,10 +354,10 @@ public class AtomixConsistentMapTest extends AtomixTestBase { ...@@ -353,10 +354,10 @@ public class AtomixConsistentMapTest extends AtomixTestBase {
353 .withValue(value1) 354 .withValue(value1)
354 .build(); 355 .build();
355 356
356 - Transaction tx = new Transaction(TransactionId.from("tx1"), Arrays.asList(update1)); 357 + MapTransaction<String, byte[]> tx = new MapTransaction<>(TransactionId.from("tx1"), Arrays.asList(update1));
357 358
358 map.prepare(tx).thenAccept(result -> { 359 map.prepare(tx).thenAccept(result -> {
359 - assertEquals(PrepareResult.OK, result); 360 + assertEquals(true, result);
360 }).join(); 361 }).join();
361 assertNull(listener.event()); 362 assertNull(listener.event());
362 363
...@@ -377,7 +378,7 @@ public class AtomixConsistentMapTest extends AtomixTestBase { ...@@ -377,7 +378,7 @@ public class AtomixConsistentMapTest extends AtomixTestBase {
377 378
378 assertNull(listener.event()); 379 assertNull(listener.event());
379 380
380 - map.commit(tx.id()).join(); 381 + map.commit(tx.transactionId()).join();
381 assertNotNull(listener.event()); 382 assertNotNull(listener.event());
382 assertEquals(MapEvent.Type.INSERT, listener.event().type()); 383 assertEquals(MapEvent.Type.INSERT, listener.event().type());
383 assertTrue(Arrays.equals(value1, listener.event().newValue().value())); 384 assertTrue(Arrays.equals(value1, listener.event().newValue().value()));
...@@ -407,13 +408,13 @@ public class AtomixConsistentMapTest extends AtomixTestBase { ...@@ -407,13 +408,13 @@ public class AtomixConsistentMapTest extends AtomixTestBase {
407 .withKey("foo") 408 .withKey("foo")
408 .withValue(value1) 409 .withValue(value1)
409 .build(); 410 .build();
410 - Transaction tx = new Transaction(TransactionId.from("tx1"), Arrays.asList(update1)); 411 + MapTransaction<String, byte[]> tx = new MapTransaction<>(TransactionId.from("tx1"), Arrays.asList(update1));
411 map.prepare(tx).thenAccept(result -> { 412 map.prepare(tx).thenAccept(result -> {
412 - assertEquals(PrepareResult.OK, result); 413 + assertEquals(true, result);
413 }).join(); 414 }).join();
414 assertNull(listener.event()); 415 assertNull(listener.event());
415 416
416 - map.rollback(tx.id()).join(); 417 + map.rollback(tx.transactionId()).join();
417 assertNull(listener.event()); 418 assertNull(listener.event());
418 419
419 map.get("foo").thenAccept(result -> { 420 map.get("foo").thenAccept(result -> {
......