Add Iterator versions of getAsyncBulk and getBulk

Add the following methods to Memcached/MemcachedIF:
- <T> BulkFuture<Map<String, T>> asyncGetBulk(Iterator<String> keys,
      Iterator<Transcoder<T>> tcs);
- <T> BulkFuture<Map<String, T>> asyncGetBulk(Iterator<String> keys,
      Transcoder<T> tc);
- BulkFuture<Map<String, Object>> asyncGetBulk(Iterator<String> keys);
- <T> Map<String, T> getBulk(Iterator<String> keys, Transcoder<T> tc);
- Map<String, Object> getBulk(Iterator<String> keys);

Some times it is more convenient/memory efficent to pass in an
Iterator of keys than it is to have a Collection of them.

Example:

public PrefixAdderIterator implements Iterator<String> {

  private final String prefix;
  private final Iterator<String> iterator;
  public PrefixAdderIterator(String prefix,
         Iterator<String> iterator) {
    this.prefix = prefix;
    this.iterator = iterator;
  }
  ...
  public String next() {
    return prefix+iterator.next();
  }
 ...
}

rather than

 List<String> prefixedKeys = new ArrayList<String>();
 for (String key: keys) {
  prefixedKeys.add(prefix+key);
 }

Change-Id: I15821983ea2ebd07fd98feeb968ce9fa578ded0b
Reviewed-on: http://review.couchbase.org/12283
Reviewed-by: Ted Crossman <[email protected]>
Reviewed-by: Matt Ingenthron <[email protected]>
Tested-by: Matt Ingenthron <[email protected]>

Author: Ted Crossman

src/main/java/net/spy/memcached/MemcachedClient.java

@@ -1022,7 +1022,7 @@ public Object get(String key) {
    * Asynchronously get a bunch of objects from the cache.
    *
    * @param <T>
-   * @param keys the keys to request
+   * @param keyIter Iterator that produces keys.
    * @param tcIter an iterator of transcoders to serialize and unserialize
    *          values; the transcoders are matched with the keys in the same
    *          order. The minimum of the key collection length and number of
@@ -1032,7 +1032,7 @@ public Object get(String key) {
    * @throws IllegalStateException in the rare circumstance where queue is too
    *           full to accept any more requests
    */
-  public <T> BulkFuture<Map<String, T>> asyncGetBulk(Collection<String> keys,
+  public <T> BulkFuture<Map<String, T>> asyncGetBulk(Iterator<String> keyIter,
       Iterator<Transcoder<T>> tcIter) {
     final Map<String, Future<T>> m = new ConcurrentHashMap<String, Future<T>>();
 
@@ -1046,7 +1046,7 @@ public <T> BulkFuture<Map<String, T>> asyncGetBulk(Collection<String> keys,
     final Map<MemcachedNode, Collection<String>> chunks =
         new HashMap<MemcachedNode, Collection<String>>();
     final NodeLocator locator = mconn.getLocator();
-    Iterator<String> keyIter = keys.iterator();
+
     while (keyIter.hasNext() && tcIter.hasNext()) {
       String key = keyIter.next();
       tcMap.put(key, tcIter.next());
@@ -1113,6 +1113,41 @@ public void complete() {
     return rv;
   }
 
+  /**
+   * Asynchronously get a bunch of objects from the cache.
+   *
+   * @param <T>
+   * @param keys the keys to request
+   * @param tcIter an iterator of transcoders to serialize and unserialize
+   *          values; the transcoders are matched with the keys in the same
+   *          order. The minimum of the key collection length and number of
+   *          transcoders is used and no exception is thrown if they do not
+   *          match
+   * @return a Future result of that fetch
+   * @throws IllegalStateException in the rare circumstance where queue is too
+   *           full to accept any more requests
+   */
+  public <T> BulkFuture<Map<String, T>> asyncGetBulk(Collection<String> keys,
+          Iterator<Transcoder<T>> tcIter) {
+    return asyncGetBulk(keys.iterator(), tcIter);
+  }
+
+  /**
+   * Asynchronously get a bunch of objects from the cache.
+   *
+   * @param <T>
+   * @param keyIter Iterator for the keys to request
+   * @param tc the transcoder to serialize and unserialize values
+   * @return a Future result of that fetch
+   * @throws IllegalStateException in the rare circumstance where queue is too
+   *           full to accept any more requests
+   */
+  public <T> BulkFuture<Map<String, T>> asyncGetBulk(Iterator<String> keyIter,
+      Transcoder<T> tc) {
+    return asyncGetBulk(keyIter,
+            new SingleElementInfiniteIterator<Transcoder<T>>(tc));
+  }
+
   /**
    * Asynchronously get a bunch of objects from the cache.
    *
@@ -1129,6 +1164,20 @@ public <T> BulkFuture<Map<String, T>> asyncGetBulk(Collection<String> keys,
         tc));
   }
 
+  /**
+   * Asynchronously get a bunch of objects from the cache and decode them with
+   * the given transcoder.
+   *
+   * @param keyIter Iterator that produces the keys to request
+   * @return a Future result of that fetch
+   * @throws IllegalStateException in the rare circumstance where queue is too
+   *           full to accept any more requests
+   */
+  public BulkFuture<Map<String, Object>> asyncGetBulk(
+         Iterator<String> keyIter) {
+    return asyncGetBulk(keyIter, transcoder);
+  }
+
   /**
    * Asynchronously get a bunch of objects from the cache and decode them with
    * the given transcoder.
@@ -1228,18 +1277,18 @@ public void gotData(String k, int flags, long cas, byte[] data) {
    * Get the values for multiple keys from the cache.
    *
    * @param <T>
-   * @param keys the keys
+   * @param keyIter Iterator that produces the keys
    * @param tc the transcoder to serialize and unserialize value
    * @return a map of the values (for each value that exists)
    * @throws OperationTimeoutException if the global operation timeout is
    *           exceeded
    * @throws IllegalStateException in the rare circumstance where queue is too
    *           full to accept any more requests
    */
-  public <T> Map<String, T> getBulk(Collection<String> keys,
+  public <T> Map<String, T> getBulk(Iterator<String> keyIter,
       Transcoder<T> tc) {
     try {
-      return asyncGetBulk(keys, tc).get(operationTimeout,
+      return asyncGetBulk(keyIter, tc).get(operationTimeout,
           TimeUnit.MILLISECONDS);
     } catch (InterruptedException e) {
       throw new RuntimeException("Interrupted getting bulk values", e);
@@ -1250,6 +1299,37 @@ public <T> Map<String, T> getBulk(Collection<String> keys,
     }
   }
 
+  /**
+   * Get the values for multiple keys from the cache.
+   *
+   * @param keyIter Iterator that produces the keys
+   * @return a map of the values (for each value that exists)
+   * @throws OperationTimeoutException if the global operation timeout is
+   *           exceeded
+   * @throws IllegalStateException in the rare circumstance where queue is too
+   *           full to accept any more requests
+   */
+  public Map<String, Object> getBulk(Iterator<String> keyIter) {
+    return getBulk(keyIter, transcoder);
+  }
+
+  /**
+   * Get the values for multiple keys from the cache.
+   *
+   * @param <T>
+   * @param keys the keys
+   * @param tc the transcoder to serialize and unserialize value
+   * @return a map of the values (for each value that exists)
+   * @throws OperationTimeoutException if the global operation timeout is
+   *           exceeded
+   * @throws IllegalStateException in the rare circumstance where queue is too
+   *           full to accept any more requests
+   */
+  public <T> Map<String, T> getBulk(Collection<String> keys,
+      Transcoder<T> tc) {
+    return getBulk(keys.iterator(), tc);
+  }
+
   /**
    * Get the values for multiple keys from the cache.
    *

src/main/java/net/spy/memcached/MemcachedClientIF.java

@@ -107,20 +107,27 @@ <T> Future<CASValue<T>> asyncGetAndTouch(final String key, final int exp,
 
   Object get(String key);
 
+  <T> BulkFuture<Map<String, T>> asyncGetBulk(Iterator<String> keys,
+      Iterator<Transcoder<T>> tcs);
   <T> BulkFuture<Map<String, T>> asyncGetBulk(Collection<String> keys,
       Iterator<Transcoder<T>> tcs);
 
+  <T> BulkFuture<Map<String, T>> asyncGetBulk(Iterator<String> keys,
+      Transcoder<T> tc);
   <T> BulkFuture<Map<String, T>> asyncGetBulk(Collection<String> keys,
       Transcoder<T> tc);
 
+  BulkFuture<Map<String, Object>> asyncGetBulk(Iterator<String> keys);
   BulkFuture<Map<String, Object>> asyncGetBulk(Collection<String> keys);
 
   <T> BulkFuture<Map<String, T>> asyncGetBulk(Transcoder<T> tc, String... keys);
 
   BulkFuture<Map<String, Object>> asyncGetBulk(String... keys);
 
+  <T> Map<String, T> getBulk(Iterator<String> keys, Transcoder<T> tc);
   <T> Map<String, T> getBulk(Collection<String> keys, Transcoder<T> tc);
 
+  Map<String, Object> getBulk(Iterator<String> keys);
   Map<String, Object> getBulk(Collection<String> keys);
 
   <T> Map<String, T> getBulk(Transcoder<T> tc, String... keys);