Aaron Kruglikov
Committed by Gerrit Code Review

Adding commands for the distributed multimap.

Change-Id: Ieecfea9aaabefa1f95918d4f95c489a0e94b01a9
...@@ -109,6 +109,28 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { ...@@ -109,6 +109,28 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive {
109 CompletableFuture<Boolean> remove(K key, V value); 109 CompletableFuture<Boolean> remove(K key, V value);
110 110
111 /** 111 /**
112 + * Removes the key-value pairs with the specified key and values if they
113 + * exist. In implementations that allow duplicates each instance of a key
114 + * will remove one matching entry, which one is not defined. Equivalent to
115 + * repeated calls to {@code remove()} for each key value pair but more
116 + * efficient.
117 + * @param key the key of the pair to be removed
118 + * @param values the set of values to be removed
119 + * @return a future whose value will be true if the map changes because of
120 + * this call, false otherwise.
121 + */
122 + CompletableFuture<Boolean> removeAll(K key, Iterable<? extends V> values);
123 +
124 + /**
125 + * Removes all values associated with the specified key as well as the key
126 + * itself.
127 + * @param key the key whose key-value pairs will be removed
128 + * @return a future whose value is the set of values that were removed,
129 + * which may be empty
130 + */
131 + CompletableFuture<Versioned<Collection<byte[]>>> removeAll(K key);
132 +
133 + /**
112 * Adds the set of key-value pairs of the specified key with each of the 134 * Adds the set of key-value pairs of the specified key with each of the
113 * values in the iterable if each key-value pair does not already exist, 135 * values in the iterable if each key-value pair does not already exist,
114 * if the pair does exist the behavior is implementation specific. 136 * if the pair does exist the behavior is implementation specific.
...@@ -142,15 +164,6 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive { ...@@ -142,15 +164,6 @@ public interface AsyncConsistentMultimap<K, V> extends DistributedPrimitive {
142 CompletableFuture<Collection<V>> replaceValues(K key, Iterable<V> values); 164 CompletableFuture<Collection<V>> replaceValues(K key, Iterable<V> values);
143 165
144 /** 166 /**
145 - * Removes all values associated with the specified key as well as the key
146 - * itself.
147 - * @param key the key whose key-value pairs will be removed
148 - * @return a future whose value is the set of values that were removed,
149 - * which may be empty
150 - */
151 - CompletableFuture<Collection<V>> removeAll(K key);
152 -
153 - /**
154 * Removes all key-value pairs, after which it will be empty. 167 * Removes all key-value pairs, after which it will be empty.
155 * @return a future whose value is irrelevant, simply used to determine if 168 * @return a future whose value is irrelevant, simply used to determine if
156 * the call has completed 169 * the call has completed
......
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 +
17 +package org.onosproject.store.primitives.resources.impl;
18 +
19 +import com.google.common.collect.Lists;
20 +import com.google.common.collect.Multimap;
21 +import com.google.common.collect.Multiset;
22 +import io.atomix.copycat.client.CopycatClient;
23 +import io.atomix.resource.AbstractResource;
24 +import org.onlab.util.Match;
25 +import org.onosproject.store.service.AsyncConsistentMultimap;
26 +import org.onosproject.store.service.Versioned;
27 +
28 +import java.util.Collection;
29 +import java.util.ConcurrentModificationException;
30 +import java.util.Map;
31 +import java.util.Properties;
32 +import java.util.Set;
33 +import java.util.concurrent.CompletableFuture;
34 +
35 +import static org.onosproject.store.primitives.resources.impl.AsyncConsistentMultimapCommands.*;
36 +
37 +/**
38 + * Set based implementation of the {@link AsyncConsistentMultimap}.
39 + * <p>
40 + * Note: this implementation does not allow null entries or duplicate entries.
41 + */
42 +public class AsyncConsistentSetMultimap
43 + extends AbstractResource<AsyncConsistentSetMultimap>
44 + implements AsyncConsistentMultimap<String, byte[]> {
45 +
46 + public AsyncConsistentSetMultimap(CopycatClient client,
47 + Properties properties) {
48 + super(client, properties);
49 + }
50 +
51 + @Override
52 + public CompletableFuture<AsyncConsistentSetMultimap> open() {
53 + return super.open();
54 + //TODO
55 + }
56 +
57 + @Override
58 + public CompletableFuture<Integer> size() {
59 + return submit(new Size());
60 + }
61 +
62 + @Override
63 + public CompletableFuture<Boolean> isEmpty() {
64 + return submit(new IsEmpty());
65 + }
66 +
67 + @Override
68 + public CompletableFuture<Boolean> containsKey(String key) {
69 + return submit(new ContainsKey(key));
70 + }
71 +
72 + @Override
73 + public CompletableFuture<Boolean> containsValue(byte[] value) {
74 + return submit(new ContainsValue(value));
75 + }
76 +
77 + @Override
78 + public CompletableFuture<Boolean> containsEntry(String key, byte[] value) {
79 + return submit(new ContainsEntry(key, value));
80 + }
81 +
82 + @Override
83 + public CompletableFuture<Boolean> put(String key, byte[] value) {
84 + return submit(new UpdateAndGet(key, Lists.newArrayList(value),
85 + Lists.newArrayList(Match.NULL),
86 + Lists.newArrayList(Match.NULL)))
87 + .whenComplete((result, e) -> throwIfLocked(result.status()))
88 + .thenApply(result ->
89 + result.status() == MapEntryUpdateResult.Status.OK);
90 + }
91 +
92 + @Override
93 + public CompletableFuture<Boolean> remove(String key, byte[] value) {
94 + return submit(new UpdateAndGet(key, Lists.newArrayList(value),
95 + Lists.newArrayList(Match.ifValue(value)),
96 + Lists.newArrayList(Match.NULL)))
97 + .whenComplete((result, e) -> throwIfLocked(result.status()))
98 + .thenApply(result ->
99 + result.status() == MapEntryUpdateResult.Status.OK);
100 + }
101 +
102 + @Override
103 + public CompletableFuture<Boolean> removeAll(String key, Iterable<? extends byte[]> values) {
104 +
105 + throw new UnsupportedOperationException("This operation cannot be " +
106 + "used without support for " +
107 + "transactions.");
108 + }
109 +
110 + @Override
111 + public CompletableFuture<Versioned<Collection<byte[]>>> removeAll(String key) {
112 + return submit(new UpdateAndGet(key, null, null, null))
113 + .whenComplete((result, e) -> throwIfLocked(result.status()))
114 + .thenApply(result -> result.oldValue());
115 + }
116 +
117 + @Override
118 + public CompletableFuture<Boolean> putAll(String key, Iterable<? extends byte[]> values) {
119 + throw new UnsupportedOperationException("This operation cannot be " +
120 + "used without support for " +
121 + "transactions.");
122 + }
123 +
124 + @Override
125 + public CompletableFuture<Boolean> putAll(Multimap<? extends String, ? extends byte[]> multiMap) {
126 + throw new UnsupportedOperationException("This operation cannot be " +
127 + "used without support for " +
128 + "transactions.");
129 + }
130 +
131 + @Override
132 + public CompletableFuture<Collection<byte[]>> replaceValues(String key, Iterable<byte[]> values) {
133 + throw new UnsupportedOperationException("This operation cannot be " +
134 + "used without support for " +
135 + "transactions.");
136 + }
137 +
138 + @Override
139 + public CompletableFuture<Void> clear() {
140 + return submit(new AsyncConsistentMultimapCommands.Clear());
141 + }
142 +
143 + @Override
144 + public CompletableFuture<Collection<byte[]>> get(String key) {
145 + return submit(new Get());
146 + }
147 +
148 + @Override
149 + public CompletableFuture<Set<String>> keySet() {
150 + return submit(new KeySet());
151 + }
152 +
153 + @Override
154 + public CompletableFuture<Multiset<String>> keys() {
155 + return submit(new Keys());
156 + }
157 +
158 + @Override
159 + public CompletableFuture<Collection<byte[]>> values() {
160 + return submit(new Values());
161 + }
162 +
163 + @Override
164 + public CompletableFuture<Collection<Map.Entry<String, byte[]>>> entries() {
165 + return submit(new Entries());
166 + }
167 +
168 + @Override
169 + public CompletableFuture<Map<String, Collection<byte[]>>> asMap() {
170 + //TODO
171 + throw new UnsupportedOperationException("Expensive operation.");
172 + }
173 +
174 + @Override
175 + public String name() {
176 + return null;
177 + }
178 +
179 + /**
180 + * Helper to check if there was a lock based issue.
181 + * @param status the status of an update result
182 + */
183 + private void throwIfLocked(MapEntryUpdateResult.Status status) {
184 + if (status == MapEntryUpdateResult.Status.WRITE_LOCK) {
185 + throw new ConcurrentModificationException("Cannot update map: Another transaction in progress");
186 + }
187 + }
188 +}