Streaming results for List Keys/Buckets, 2i, and MR

src/main/java/com/basho/riak/client/api/commands/indexes/SecondaryIndexQuery.java

@@ -625,6 +625,16 @@ protected Response(Namespace queryLocation,
             this.coreResponse = coreResponse;
         }
 
+        /**
+         * Get an iterator over the result data.
+         *
+         * If using the streaming API, this method will block
+         * and wait for more data if none is immediately available.
+         * It is also advisable to check {@link Thread#isInterrupted()}
+         * in environments where thread interrupts must be obeyed.
+         *
+         * @return an iterator over the result data.
+         */
         public Iterator<E> iterator()
         {
             if (isStreamable()) {
@@ -644,6 +654,10 @@ protected E convert(SecondaryIndexQueryOperation.Response.Entry e)
         /**
          * Check if this response has a continuation.
          *
+         * If using the streaming API, this property's value
+         * may change while data is being received, therefore
+         * it is best to call it after the operation is complete.
+         *
          * @return true if the response contains a continuation.
          */
         public boolean hasContinuation()
@@ -659,6 +673,10 @@ public boolean hasContinuation()
         /**
          * Get the continuation from this response.
          *
+         * If using the streaming API, this property's value
+         * may change while data is being received, therefore
+         * it is best to call it after the operation is complete.
+         *
          * @return the continuation, or null if none is present.
          */
         public BinaryValue getContinuation()
@@ -674,6 +692,11 @@ public BinaryValue getContinuation()
         /**
          * Check is this response contains any entries.
          *
+         * If using the streaming API, this method will block
+         * and wait for more data if none is immediately available.
+         * It is also advisable to check {@link Thread#isInterrupted()}
+         * in environments where thread interrupts must be obeyed.
+         *
          * @return true if entries are present, false otherwise.
          */
         public boolean hasEntries()
@@ -686,8 +709,19 @@ public boolean hasEntries()
             return !coreResponse.getEntryList().isEmpty();
         }
 
+        /**
+         * Get a list of the result entries for this response.
+         * If using the streaming API this method will return an empty list.
+         *
+         * @return A list of result entries.
+         */
         public final List<E> getEntries()
         {
+            if(isStreamable())
+            {
+                return new ArrayList<>(0);
+            }
+
             final List<SecondaryIndexQueryOperation.Response.Entry> coreEntries = coreResponse.getEntryList();
             final List<E> convertedList = new ArrayList<>(coreEntries.size());
 
@@ -699,13 +733,6 @@ public final List<E> getEntries()
             return convertedList;
         }
 
-        /**
-         * Factory method.
-         * @param location
-         * @param coreEntry
-         * @param converter
-         * @return
-         */
         @SuppressWarnings("unchecked")
         protected E createEntry(Location location, SecondaryIndexQueryOperation.Response.Entry coreEntry, IndexConverter<T> converter)
         {

src/test/java/com/basho/riak/client/api/commands/indexes/itest/ITestBigIntIndexQuery.java

@@ -107,6 +107,7 @@ public void testStreamingRangeQuery() throws ExecutionException, InterruptedExce
         final BigIntIndexQuery.Response streamingResponse = streamingFuture.get();
 
         assertTrue(streamingResponse.hasEntries());
+        assertTrue(streamingResponse.getEntries().isEmpty());
 
         final String expectedObjectKey = objectKey(1);
         boolean found = false;

src/test/java/com/basho/riak/client/api/commands/indexes/itest/ITestBinIndexQuery.java

@@ -21,13 +21,15 @@
 import com.basho.riak.client.api.annotations.RiakIndex;
 import com.basho.riak.client.api.annotations.RiakKey;
 import com.basho.riak.client.api.commands.indexes.BinIndexQuery;
+import com.basho.riak.client.api.commands.indexes.SecondaryIndexQuery;
 import com.basho.riak.client.api.commands.kv.StoreValue;
 import com.basho.riak.client.core.RiakFuture;
 import com.basho.riak.client.core.query.Location;
 import org.junit.Assume;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
+import java.util.List;
 import java.util.concurrent.ExecutionException;
 
 import static org.junit.Assert.assertEquals;
@@ -105,6 +107,7 @@ public void testStreamingRangeQuery() throws ExecutionException, InterruptedExce
         final BinIndexQuery.Response streamingResponse = streamingFuture.get();
 
         assertTrue(streamingResponse.hasEntries());
+        assertTrue(streamingResponse.getEntries().isEmpty());
 
         final String expectedObjectKey = objectKey(1);
         final String expectedIndexKey = indexKey(1);

src/test/java/com/basho/riak/client/api/commands/indexes/itest/ITestIntIndexQuery.java

@@ -113,6 +113,7 @@ public void testStreamingRangeQuery() throws ExecutionException, InterruptedExce
         final IntIndexQuery.Response streamingResponse = streamingFuture.get();
 
         assertTrue(streamingResponse.hasEntries());
+        assertTrue(streamingResponse.getEntries().isEmpty());
 
         final String expectedObjectKey = objectKey(1);
         boolean found = false;

src/test/java/com/basho/riak/client/api/commands/indexes/itest/ITestRawIndexQuery.java

@@ -182,6 +182,7 @@ public void testBucketIndexQueryStreaming() throws InterruptedException, Executi
         final BinIndexQuery.Response streamingResponse = indexResult.get();
 
         assertTrue(streamingResponse.hasEntries());
+        assertTrue(streamingResponse.getEntries().isEmpty());
         assertEquals(100, StreamSupport.stream(streamingResponse.spliterator(), false).count());
 
         // Assert everything was consumed