Added support for createIndexes

Author: rozza

driver/src/main/com/mongodb/reactivestreams/client/MongoCollection.java

@@ -20,7 +20,6 @@
 import com.mongodb.ReadPreference;
 import com.mongodb.WriteConcern;
 import com.mongodb.annotations.ThreadSafe;
-import com.mongodb.async.SingleResultCallback;
 import com.mongodb.bulk.BulkWriteResult;
 import com.mongodb.client.model.BulkWriteOptions;
 import com.mongodb.client.model.CountOptions;
@@ -448,7 +447,7 @@ public interface MongoCollection<TDocument> {
      * @return a publisher with a single element indicating when the operation has completed
      * @mongodb.driver.manual reference/method/db.collection.ensureIndex Ensure Index
      */
-    Publisher<Success> createIndex(Bson key);
+    Publisher<String> createIndex(Bson key);
 
     /**
      * Creates an index.
@@ -458,7 +457,7 @@ public interface MongoCollection<TDocument> {
      * @return a publisher with a single element indicating when the operation has completed
      * @mongodb.driver.manual reference/method/db.collection.ensureIndex Ensure Index
      */
-    Publisher<Success> createIndex(Bson key, IndexOptions options);
+    Publisher<String> createIndex(Bson key, IndexOptions options);
 
 
     /**
@@ -469,7 +468,7 @@ public interface MongoCollection<TDocument> {
      * @return a publisher with a single element indicating when the operation has completed
      * @mongodb.server.release 2.6
      */
-    Publisher<Success> createIndexes(List<IndexModel> indexes);
+    Publisher<String> createIndexes(List<IndexModel> indexes);
 
     /**
      * Get all the indexes in this collection.

driver/src/main/com/mongodb/reactivestreams/client/MongoCollectionImpl.java

@@ -326,26 +326,26 @@ void execute(final SingleResultCallback<Success> callback) {
     }
 
     @Override
-    public Publisher<Success> createIndex(final Bson key) {
+    public Publisher<String> createIndex(final Bson key) {
         return createIndex(key, new IndexOptions());
     }
 
     @Override
-    public Publisher<Success> createIndex(final Bson key, final IndexOptions options) {
-        return new SingleResultPublisher<Success>() {
+    public Publisher<String> createIndex(final Bson key, final IndexOptions options) {
+        return new SingleResultPublisher<String>() {
             @Override
-            void execute(final SingleResultCallback<Success> callback) {
-                wrapped.createIndex(key, options, voidToSuccessCallback(callback));
+            void execute(final SingleResultCallback<String> callback) {
+                wrapped.createIndex(key, options, callback);
             }
         };
     }
 
     @Override
-    public Publisher<Success> createIndexes(final List<IndexModel> indexes) {
-        return new SingleResultPublisher<Success>() {
+    public Publisher<String> createIndexes(final List<IndexModel> indexes) {
+        return new SingleResultListPublisher<String>() {
             @Override
-            void execute(final SingleResultCallback<Success> callback) {
-                wrapped.createIndexes(indexes, voidToSuccessCallback(callback));
+            void execute(final SingleResultCallback<List<String>> callback) {
+                wrapped.createIndexes(indexes, callback);
             }
         };
     }

driver/src/main/com/mongodb/reactivestreams/client/SingleResultListPublisher.java

@@ -0,0 +1,56 @@
+package com.mongodb.reactivestreams.client;
+
+import com.mongodb.async.SingleResultCallback;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+abstract class SingleResultListPublisher<TResult> implements Publisher<TResult> {
+
+    @Override
+    public void subscribe(final Subscriber<? super TResult> s) {
+        new SingleResultSubscription(s).start();
+    }
+
+    SingleResultCallback<List<TResult>> getCallback(final SubscriptionSupport<TResult> subscription) {
+        return new SingleResultCallback<List<TResult>>() {
+            @Override
+            public void onResult(final List<TResult> results, final Throwable t) {
+                subscription.log("result - " + results + " : " + t);
+                if (t != null) {
+                    subscription.onError(t);
+                } else {
+                    if (results != null) {
+                        for (TResult result : results) {
+                            subscription.onNext(result);
+                        }
+                    }
+                    subscription.onComplete();
+                }
+            }
+        };
+    }
+
+    private class SingleResultSubscription extends SubscriptionSupport<TResult> {
+        private final AtomicBoolean operationCompleted = new AtomicBoolean();
+
+        public SingleResultSubscription(final Subscriber<? super TResult> subscriber) {
+            super(subscriber);
+        }
+
+        @Override
+        protected void doRequest(final long n) {
+            log("doRequest : " + n);
+            if (operationCompleted.compareAndSet(false, true)) {
+                execute(getCallback(this));
+            }
+        }
+    }
+
+    abstract void execute(SingleResultCallback<List<TResult>> callback);
+
+    SingleResultListPublisher() {
+    }
+}

driver/src/test/functional/com/mongodb/reactivestreams/client/SmokeTestSpecification.groovy

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 MongoDB, Inc.
+ * Copyright 2014-2015 MongoDB, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 package com.mongodb.reactivestreams.client
 
 import com.mongodb.MongoNamespace
+import com.mongodb.client.model.IndexModel
 import com.mongodb.diagnostics.logging.Loggers
 import org.bson.Document
 
@@ -89,14 +90,29 @@ class SmokeTestSpecification extends FunctionalSpecification {
         run('The count is zero', collection.&count)[0] == 0
 
         then:
-        run('create an index', collection.&createIndex, new Document('test', 1))[0] == Success.SUCCESS
+        run('create an index', collection.&createIndex, new Document('test', 1))[0] == 'test_1'
 
         then:
         def indexNames = run('has the newly created index', collection.&listIndexes)*.name
 
         then:
         indexNames.containsAll('_id_', 'test_1')
 
+        then:
+        run('create multiple indexes', collection.&createIndexes, [new IndexModel(new Document('multi', 1))])[0] == 'multi_1'
+
+        then:
+        def indexNamesUpdated = run('has the newly created index', collection.&listIndexes)*.name
+
+        then:
+        indexNamesUpdated.containsAll('_id_', 'test_1', 'multi_1')
+
+        then:
+        run('drop the index', collection.&dropIndex, 'multi_1')[0] == Success.SUCCESS
+
+        then:
+        run('has a single index left "_id" ', collection.&listIndexes).size == 2
+
         then:
         run('drop the index', collection.&dropIndex, 'test_1')[0] == Success.SUCCESS
 

driver/src/test/unit/com/mongodb/reactivestreams/client/MongoCollectionSpecification.groovy

@@ -15,6 +15,7 @@
  */
 
 package com.mongodb.reactivestreams.client
+
 import com.mongodb.MongoNamespace
 import com.mongodb.ReadPreference
 import com.mongodb.WriteConcern