INTERNAL: Creating piped collection operation future.

Author: brido4125

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

@@ -126,6 +126,7 @@
 import net.spy.memcached.internal.CollectionGetBulkFuture;
 import net.spy.memcached.internal.OperationFuture;
 import net.spy.memcached.internal.SMGetFuture;
+import net.spy.memcached.internal.PipedCollectionFuture;
 import net.spy.memcached.ops.BTreeFindPositionOperation;
 import net.spy.memcached.ops.BTreeFindPositionWithGetOperation;
 import net.spy.memcached.ops.BTreeGetBulkOperation;
@@ -948,15 +949,10 @@ public void gotStatus(Integer index, OperationStatus status) {
   <T> CollectionFuture<Map<Integer, CollectionOperationStatus>> asyncCollectionPipedUpdate(
           final String key, final List<CollectionPipedUpdate<T>> updateList) {
 
-    final ConcurrentLinkedQueue<Operation> ops = new ConcurrentLinkedQueue<Operation>();
-
     final CountDownLatch latch = new CountDownLatch(updateList.size());
-
-    final List<CollectionOperationStatus> mergedOperationStatus = Collections
-            .synchronizedList(new ArrayList<CollectionOperationStatus>(updateList.size()));
-
-    final Map<Integer, CollectionOperationStatus> mergedResult =
-        new ConcurrentHashMap<Integer, CollectionOperationStatus>();
+    final PipedCollectionFuture<Integer, CollectionOperationStatus> rv =
+            new PipedCollectionFuture<Integer, CollectionOperationStatus>(latch, operationTimeout, updateList.size());
+    Collection<Operation> ops = new ConcurrentLinkedQueue<Operation>();
 
     for (int i = 0; i < updateList.size(); i++) {
       final CollectionPipedUpdate<T> update = updateList.get(i);
@@ -974,7 +970,7 @@ public void receivedStatus(OperationStatus status) {
                 getLogger().warn("Unhandled state: " + status);
                 cstatus = new CollectionOperationStatus(status);
               }
-              mergedOperationStatus.add(cstatus);
+              rv.addOperationStatus(cstatus);
             }
 
             // complete
@@ -985,97 +981,19 @@ public void complete() {
             // got status
             public void gotStatus(Integer index, OperationStatus status) {
               if (status instanceof CollectionOperationStatus) {
-                mergedResult.put(index + (idx * CollectionPipedUpdate.MAX_PIPED_ITEM_COUNT),
+                rv.addEachResult(index + (idx * CollectionPipedUpdate.MAX_PIPED_ITEM_COUNT),
                                 (CollectionOperationStatus) status);
               } else {
-                mergedResult.put(index + (idx * CollectionPipedUpdate.MAX_PIPED_ITEM_COUNT),
+                rv.addEachResult(index + (idx * CollectionPipedUpdate.MAX_PIPED_ITEM_COUNT),
                                 new CollectionOperationStatus(status));
               }
             }
           });
-      addOp(key, op);
       ops.add(op);
     }
-
-    return new CollectionFuture<Map<Integer, CollectionOperationStatus>>(
-            latch, operationTimeout) {
-
-      @Override
-      public boolean cancel(boolean ign) {
-        boolean rv = false;
-        for (Operation op : ops) {
-          op.cancel("by application.");
-          rv |= op.getState() == OperationState.WRITE_QUEUED;
-        }
-        return rv;
-      }
-
-      @Override
-      public boolean isCancelled() {
-        for (Operation op : ops) {
-          if (op.isCancelled()) {
-            return true;
-          }
-        }
-        return false;
-      }
-
-      @Override
-      public Map<Integer, CollectionOperationStatus> get(long duration,
-                                                         TimeUnit units)
-          throws InterruptedException, TimeoutException, ExecutionException {
-
-        if (!latch.await(duration, units)) {
-          Collection<Operation> timedoutOps = new HashSet<Operation>();
-          for (Operation op : ops) {
-            if (op.getState() != OperationState.COMPLETE) {
-              timedoutOps.add(op);
-            } else {
-              MemcachedConnection.opSucceeded(op);
-            }
-          }
-          if (timedoutOps.size() > 0) {
-            MemcachedConnection.opTimedOut(timedoutOps.iterator().next());
-            throw new CheckedOperationTimeoutException(duration, units, timedoutOps);
-          }
-        } else {
-          // continuous timeout counter will be reset only once in pipe
-          MemcachedConnection.opSucceeded(ops.iterator().next());
-        }
-
-        for (Operation op : ops) {
-          if (op != null && op.hasErrored()) {
-            throw new ExecutionException(op.getException());
-          }
-
-          if (op != null && op.isCancelled()) {
-            throw new ExecutionException(new RuntimeException(op.getCancelCause()));
-          }
-        }
-
-        return mergedResult;
-      }
-
-      @Override
-      public CollectionOperationStatus getOperationStatus() {
-        for (CollectionOperationStatus status : mergedOperationStatus) {
-          if (!status.isSuccess()) {
-            return status;
-          }
-        }
-        return new CollectionOperationStatus(true, "END", CollectionResponse.END);
-      }
-
-      @Override
-      public boolean isDone() {
-        for (Operation op : ops) {
-          if (!(op.getState() == OperationState.COMPLETE || op.isCancelled())) {
-            return false;
-          }
-        }
-        return true;
-      }
-    };
+    rv.setOps(ops);
+    addOps(key, ops);
+    return rv;
   }
 
   /**
@@ -3903,15 +3821,10 @@ public SMGetFuture<List<SMGetElement<Object>>> asyncBopSortMergeGet(
   <T> CollectionFuture<Map<Integer, CollectionOperationStatus>> asyncCollectionPipedInsert(
           final String key, final List<CollectionPipedInsert<T>> insertList) {
 
-    final ConcurrentLinkedQueue<Operation> ops = new ConcurrentLinkedQueue<Operation>();
-
     final CountDownLatch latch = new CountDownLatch(insertList.size());
-
-    final List<CollectionOperationStatus> mergedOperationStatus = Collections
-            .synchronizedList(new ArrayList<CollectionOperationStatus>(insertList.size()));
-
-    final Map<Integer, CollectionOperationStatus> mergedResult =
-        new ConcurrentHashMap<Integer, CollectionOperationStatus>();
+    final PipedCollectionFuture<Integer, CollectionOperationStatus> rv =
+            new PipedCollectionFuture<Integer, CollectionOperationStatus>(latch, operationTimeout, insertList.size());
+    Collection<Operation> ops = new ConcurrentLinkedQueue<Operation>();
 
     for (int i = 0; i < insertList.size(); i++) {
       final CollectionPipedInsert<T> insert = insertList.get(i);
@@ -3929,7 +3842,7 @@ public void receivedStatus(OperationStatus status) {
                 getLogger().warn("Unhandled state: " + status);
                 cstatus = new CollectionOperationStatus(status);
               }
-              mergedOperationStatus.add(cstatus);
+              rv.addOperationStatus(cstatus);
             }
 
             // complete
@@ -3940,96 +3853,19 @@ public void complete() {
             // got status
             public void gotStatus(Integer index, OperationStatus status) {
               if (status instanceof CollectionOperationStatus) {
-                mergedResult.put(index + (idx * CollectionPipedInsert.MAX_PIPED_ITEM_COUNT),
+                rv.addEachResult(index + (idx * CollectionPipedInsert.MAX_PIPED_ITEM_COUNT),
                                 (CollectionOperationStatus) status);
               } else {
-                mergedResult.put(index + (idx * CollectionPipedInsert.MAX_PIPED_ITEM_COUNT),
+                rv.addEachResult(index + (idx * CollectionPipedInsert.MAX_PIPED_ITEM_COUNT),
                                 new CollectionOperationStatus(status));
               }
             }
           });
-      addOp(key, op);
       ops.add(op);
     }
-
-    return new CollectionFuture<Map<Integer, CollectionOperationStatus>>(
-            latch, operationTimeout) {
-
-      @Override
-      public boolean cancel(boolean ign) {
-        boolean rv = false;
-        for (Operation op : ops) {
-          op.cancel("by application.");
-          rv |= op.getState() == OperationState.WRITE_QUEUED;
-        }
-        return rv;
-      }
-
-      @Override
-      public boolean isCancelled() {
-        for (Operation op : ops) {
-          if (op.isCancelled()) {
-            return true;
-          }
-        }
-        return false;
-      }
-
-      @Override
-      public Map<Integer, CollectionOperationStatus> get(long duration,
-                                                         TimeUnit units)
-          throws InterruptedException, TimeoutException, ExecutionException {
-
-        if (!latch.await(duration, units)) {
-          Collection<Operation> timedoutOps = new HashSet<Operation>();
-          for (Operation op : ops) {
-            if (op.getState() != OperationState.COMPLETE) {
-              timedoutOps.add(op);
-            } else {
-              MemcachedConnection.opSucceeded(op);
-            }
-          }
-          if (timedoutOps.size() > 0) {
-            MemcachedConnection.opTimedOut(timedoutOps.iterator().next());
-            throw new CheckedOperationTimeoutException(duration, units, timedoutOps);
-          }
-        } else {
-          // continuous timeout counter will be reset only once in pipe
-          MemcachedConnection.opSucceeded(ops.iterator().next());
-        }
-        for (Operation op : ops) {
-          if (op != null && op.hasErrored()) {
-            throw new ExecutionException(op.getException());
-          }
-
-          if (op != null && op.isCancelled()) {
-            throw new ExecutionException(new RuntimeException(op.getCancelCause()));
-          }
-        }
-
-        return mergedResult;
-      }
-
-      @Override
-      public CollectionOperationStatus getOperationStatus() {
-        for (CollectionOperationStatus status : mergedOperationStatus) {
-          if (!status.isSuccess()) {
-            return status;
-          }
-        }
-        return new CollectionOperationStatus(true, "END", CollectionResponse.END);
-      }
-
-      @Override
-      public boolean isDone() {
-        for (Operation op : ops) {
-          if (!(op.getState() == OperationState.COMPLETE || op.isCancelled())) {
-            return false;
-          }
-        }
-        return true;
-      }
-    };
+    rv.setOps(ops);
+    addOps(key, ops);
+    return rv;
   }
 
   @Override

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

@@ -366,6 +366,12 @@ protected Operation addOp(final MemcachedNode node, final Operation op) {
     return op;
   }
 
+  protected void addOps(final String key, final Collection<Operation> ops) {
+    validateKey(key);
+    checkState();
+    conn.addOperations(key, ops);
+  }
+
   protected void addOpMap(final Map<String, Operation> opMap) {
     checkState();
     for (Map.Entry<String, Operation> me : opMap.entrySet()) {

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

@@ -1464,14 +1464,7 @@ public void insertOperation(final MemcachedNode node, final Operation o) {
   }
 
   public void addOperation(final MemcachedNode node, final Operation o) {
-    if (node == null) {
-      o.cancel("no node");
-      return;
-    }
-    if ((!node.isActive() && !node.isFirstConnecting()) &&
-        failureMode == FailureMode.Cancel) {
-      o.setHandlingNode(node);
-      o.cancel("inactive node");
+    if (checkNodeState(node, o)) {
       return;
     }
     node.addOpToInputQ(o);
@@ -1481,17 +1474,29 @@ public void addOperation(final MemcachedNode node, final Operation o) {
     getLogger().debug("Added %s to %s", o, node);
   }
 
-  public void addOperations(final Map<MemcachedNode, Operation> ops) {
-    for (Map.Entry<MemcachedNode, Operation> me : ops.entrySet()) {
-      final MemcachedNode node = me.getKey();
-      Operation o = me.getValue();
-      node.addOpToInputQ(o);
-      addedQueue.offer(node);
+  public void addOperations(String key, Collection<Operation> ops) {
+    MemcachedNode findNode = findNodeByKey(key);
+    for (Operation op : ops) {
+      addOperation(findNode, op);
     }
     Selector s = selector.wakeup();
     assert s == selector : "Wakeup returned the wrong selector.";
   }
 
+  private boolean checkNodeState(MemcachedNode findNode, Operation op) {
+    if (findNode == null) {
+      op.cancel("no node");
+      return true;
+    }
+    if ((!findNode.isActive() && !findNode.isFirstConnecting()) &&
+            failureMode == FailureMode.Cancel) {
+      op.setHandlingNode(findNode);
+      op.cancel("inactive node");
+      return true;
+    }
+    return false;
+  }
+
   /**
    * Broadcast an operation to all nodes.
    */