Implement distributed multiset state machine.

core/src/main/java/io/atomix/core/Atomix.java

@@ -29,6 +29,7 @@
 import io.atomix.core.atomic.treemap.AtomicTreeMap;
 import io.atomix.core.atomic.value.AtomicValue;
 import io.atomix.core.collection.list.DistributedList;
+import io.atomix.core.collection.multiset.DistributedMultiset;
 import io.atomix.core.collection.queue.DistributedQueue;
 import io.atomix.core.collection.set.DistributedSet;
 import io.atomix.core.concurrent.lock.DistributedLock;
@@ -372,6 +373,11 @@ public <E> DistributedList<E> getList(String name) {
     return primitives.getList(name);
   }
 
+  @Override
+  public <E> DistributedMultiset<E> getMultiset(String name) {
+    return primitives.getMultiset(name);
+  }
+
   @Override
   public AtomicCounter getAtomicCounter(String name) {
     return primitives.getAtomicCounter(name);

core/src/main/java/io/atomix/core/PrimitivesService.java

@@ -42,6 +42,9 @@
 import io.atomix.core.collection.list.DistributedList;
 import io.atomix.core.collection.list.DistributedListBuilder;
 import io.atomix.core.collection.list.DistributedListType;
+import io.atomix.core.collection.multiset.DistributedMultiset;
+import io.atomix.core.collection.multiset.DistributedMultisetBuilder;
+import io.atomix.core.collection.multiset.DistributedMultisetType;
 import io.atomix.core.collection.queue.DistributedQueue;
 import io.atomix.core.collection.queue.DistributedQueueBuilder;
 import io.atomix.core.collection.queue.DistributedQueueType;
@@ -266,6 +269,29 @@ default <E> DistributedListBuilder<E> listBuilder(String name, PrimitiveProtocol
     return primitiveBuilder(name, DistributedListType.instance(), protocol);
   }
 
+  /**
+   * Creates a new DistributedMultisetBuilder.
+   *
+   * @param name the primitive name
+   * @param <E>  multiset element type
+   * @return builder for a distributed multiset
+   */
+  default <E> DistributedMultisetBuilder<E> multisetBuilder(String name) {
+    return primitiveBuilder(name, DistributedMultisetType.instance());
+  }
+
+  /**
+   * Creates a new DistributedMultisetBuilder.
+   *
+   * @param name     the primitive name
+   * @param protocol the primitive protocol
+   * @param <E>      multiset element type
+   * @return builder for a distributed multiset
+   */
+  default <E> DistributedMultisetBuilder<E> multisetBuilder(String name, PrimitiveProtocol protocol) {
+    return primitiveBuilder(name, DistributedMultisetType.instance(), protocol);
+  }
+
   /**
    * Creates a new AtomicCounterBuilder.
    *
@@ -529,6 +555,15 @@ default TransactionBuilder transactionBuilder() {
    */
   <E> DistributedList<E> getList(String name);
 
+  /**
+   * Creates a new DistributedMultiset.
+   *
+   * @param name the primitive name
+   * @param <E>  multiset element type
+   * @return a multiton instance of a distributed multiset
+   */
+  <E> DistributedMultiset<E> getMultiset(String name);
+
   /**
    * Creates a new AtomicCounterBuilder.
    *

core/src/main/java/io/atomix/core/atomic/multimap/impl/AtomicMultimapProxy.java

@@ -18,6 +18,7 @@
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Multiset;
 import io.atomix.core.atomic.multimap.AsyncAtomicMultimap;
 import io.atomix.core.atomic.multimap.AtomicMultimap;
 import io.atomix.core.atomic.multimap.AtomicMultimapEvent;
@@ -393,6 +394,41 @@ public CompletableFuture<Boolean> contains(String element) {
       return containsKey(element);
     }
 
+    @Override
+    public CompletableFuture<Integer> count(Object element) {
+      return get((String) element).thenApply(value -> value == null ? 0 : value.value().size());
+    }
+
+    @Override
+    public CompletableFuture<Integer> add(String element, int occurrences) {
+      return Futures.exceptionalFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public CompletableFuture<Integer> remove(Object element, int occurrences) {
+      return Futures.exceptionalFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public CompletableFuture<Integer> setCount(String element, int count) {
+      return Futures.exceptionalFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public CompletableFuture<Boolean> setCount(String element, int oldCount, int newCount) {
+      return Futures.exceptionalFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public AsyncDistributedSet<String> elementSet() {
+      return new AtomicMultimapKeySet();
+    }
+
+    @Override
+    public AsyncDistributedSet<Multiset.Entry<String>> entrySet() {
+      throw new UnsupportedOperationException();
+    }
+
     @Override
     public CompletableFuture<Boolean> addAll(Collection<? extends String> c) {
       return Futures.exceptionalFuture(new UnsupportedOperationException());
@@ -512,6 +548,41 @@ public CompletableFuture<Boolean> contains(byte[] element) {
       return containsValue(element);
     }
 
+    @Override
+    public CompletableFuture<Integer> count(Object element) {
+      return Futures.exceptionalFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public CompletableFuture<Integer> add(byte[] element, int occurrences) {
+      return Futures.exceptionalFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public CompletableFuture<Integer> remove(Object element, int occurrences) {
+      return Futures.exceptionalFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public CompletableFuture<Integer> setCount(byte[] element, int count) {
+      return Futures.exceptionalFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public CompletableFuture<Boolean> setCount(byte[] element, int oldCount, int newCount) {
+      return Futures.exceptionalFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public AsyncDistributedSet<byte[]> elementSet() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public AsyncDistributedSet<Multiset.Entry<byte[]>> entrySet() {
+      throw new UnsupportedOperationException();
+    }
+
     @Override
     public CompletableFuture<Boolean> addAll(Collection<? extends byte[]> c) {
       return Futures.exceptionalFuture(new UnsupportedOperationException());

core/src/main/java/io/atomix/core/collection/impl/DistributedCollectionProxy.java

@@ -164,13 +164,13 @@ public CompletableFuture<A> connect() {
    */
   private class DistributedCollectionIterator implements AsyncIterator<String> {
     private final long id;
-    private volatile CompletableFuture<DistributedCollectionService.Batch> batch;
+    private volatile CompletableFuture<DistributedCollectionService.Batch<String>> batch;
     private volatile CompletableFuture<Void> closeFuture;
 
     DistributedCollectionIterator(long id) {
       this.id = id;
       this.batch = CompletableFuture.completedFuture(
-          new DistributedCollectionService.Batch(0, Collections.emptyList()));
+          new DistributedCollectionService.Batch<>(0, Collections.emptyList()));
     }
 
     /**
@@ -194,7 +194,7 @@ private CompletableFuture<Iterator<String>> batch() {
      * @param position the position from which to fetch the next batch
      * @return the next batch of entries from the cluster
      */
-    private CompletableFuture<DistributedCollectionService.Batch> fetch(int position) {
+    private CompletableFuture<DistributedCollectionService.Batch<String>> fetch(int position) {
       return getProxyClient().applyBy(name(), service -> service.next(id, position))
           .thenCompose(batch -> {
             if (batch == null) {

core/src/main/java/io/atomix/core/collection/impl/DistributedCollectionService.java

@@ -259,7 +259,7 @@ public interface DistributedCollectionService {
    * @return the next batch of entries for the iterator or {@code null} if the iterator is complete
    */
   @Query
-  Batch next(long iteratorId, int position);
+  Batch<String> next(long iteratorId, int position);
 
   /**
    * Closes an iterator.
@@ -272,12 +272,12 @@ public interface DistributedCollectionService {
   /**
    * Iterator batch.
    */
-  final class Batch implements Iterator<String> {
+  final class Batch<T> implements Iterator<T> {
     private final int position;
-    private final Collection<String> elements;
-    private transient volatile Iterator<String> iterator;
+    private final Collection<T> elements;
+    private transient volatile Iterator<T> iterator;
 
-    Batch(int position, Collection<String> elements) {
+    public Batch(int position, Collection<T> elements) {
       this.position = position;
       this.elements = elements;
     }
@@ -296,11 +296,11 @@ public int position() {
      *
      * @return the batch of elements
      */
-    public Collection<String> elements() {
+    public Collection<T> elements() {
       return elements;
     }
 
-    private Iterator<String> iterator() {
+    private Iterator<T> iterator() {
       if (iterator == null) {
         synchronized (this) {
           if (iterator == null) {
@@ -317,7 +317,7 @@ public boolean hasNext() {
     }
 
     @Override
-    public String next() {
+    public T next() {
       return iterator().next();
     }
   }

core/src/main/java/io/atomix/core/collection/impl/PartitionedDistributedCollectionProxy.java

@@ -237,13 +237,14 @@ public CompletableFuture<String> next() {
   private class DistributedCollectionPartitionIterator implements AsyncIterator<String> {
     private final PartitionId partitionId;
     private final long iteratorId;
-    private volatile CompletableFuture<DistributedCollectionService.Batch> batch;
+    private volatile CompletableFuture<DistributedCollectionService.Batch<String>> batch;
     private volatile CompletableFuture<Void> closeFuture;
 
     DistributedCollectionPartitionIterator(PartitionId partitionId, long iteratorId) {
       this.partitionId = partitionId;
       this.iteratorId = iteratorId;
-      this.batch = CompletableFuture.completedFuture(new DistributedCollectionService.Batch(0, Collections.emptyList()));
+      this.batch = CompletableFuture.completedFuture(
+          new DistributedCollectionService.Batch<>(0, Collections.emptyList()));
     }
 
     /**
@@ -267,7 +268,7 @@ private CompletableFuture<Iterator<String>> batch() {
      * @param position the position from which to fetch the next batch
      * @return the next batch of entries from the cluster
      */
-    private CompletableFuture<DistributedCollectionService.Batch> fetch(int position) {
+    private CompletableFuture<DistributedCollectionService.Batch<String>> fetch(int position) {
       return getProxyClient().applyOn(partitionId, service -> service.next(iteratorId, position))
           .thenCompose(batch -> {
             if (batch == null) {