FEATURE: Add CompletableFuture Pipeline API

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

@@ -196,12 +196,12 @@ public class ArcusClient extends FrontCacheMemcachedClient implements ArcusClien
 
   private final Transcoder<Object> collectionTranscoder;
 
-  private static final int BOPGET_BULK_CHUNK_SIZE = 200;
-  private static final int SMGET_CHUNK_SIZE = 500;
+  public static final int BOPGET_BULK_CHUNK_SIZE = 200;
+  public static final int SMGET_CHUNK_SIZE = 500;
   private static final int NON_PIPED_BULK_INSERT_CHUNK_SIZE = 500;
 
-  private static final int MAX_GETBULK_ELEMENT_COUNT = 50;
-  private static final int MAX_SMGET_COUNT = 1000; // server configuration is 2000.
+  public static final int MAX_GETBULK_ELEMENT_COUNT = 50;
+  public static final int MAX_SMGET_COUNT = 1000; // server configuration is 2000.
 
   private static final int SHUTDOWN_TIMEOUT_MILLISECONDS = 2000;
   private static final AtomicInteger CLIENT_ID = new AtomicInteger(1);

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

@@ -531,4 +531,15 @@ BTreeInsertAndGetOperation bopInsertAndGet(String key,
                                              BTreeInsertAndGet<?> get, byte[] dataToInsert,
                                              OperationCallback cb);
 
+  /**
+   * Create a pipeline operation for executing multiple commands in batch.
+   *
+   * @param ops             the list of operations to execute
+   * @param keys            the unique keys related to the pipe operation
+   * @param cb              the status callback
+   * @return a new pipeline operation
+   */
+  Operation pipeline(List<KeyedOperation> ops, List<String> keys,
+                     OperationCallback cb);
+
 }

src/main/java/net/spy/memcached/ops/APIType.java

@@ -66,6 +66,7 @@ public enum APIType {
   SETATTR(OperationType.WRITE), GETATTR(OperationType.READ),
 
   // Other API
+  PIPE(OperationType.WRITE),
   FLUSH(OperationType.WRITE),
   STATS(OperationType.ETC),
   VERSION(OperationType.ETC),

src/main/java/net/spy/memcached/ops/BaseOperationFactory.java

@@ -25,6 +25,7 @@
 import net.spy.memcached.collection.BTreeGetBulk;
 import net.spy.memcached.collection.BTreeSMGet;
 import net.spy.memcached.collection.CollectionBulkInsert;
+import net.spy.memcached.protocol.ascii.PipelineOperationImpl;
 
 /**
  * Base class for operation factories.
@@ -151,6 +152,16 @@ public Operation cloneMultiOperation(KeyedOperation op, MemcachedNode node,
     } else if (op instanceof BTreeSortMergeGetOperation) {
       final BTreeSMGet<?> smGet = ((BTreeSortMergeGetOperation) op).getSMGet();
       return bopsmget(smGet.clone(node, redirectKeys), (BTreeSortMergeGetOperation.Callback) mcb);
+    } else if (op instanceof PipelineOperation) {
+      PipelineOperationImpl pipeOp = (PipelineOperationImpl) op;
+      List<KeyedOperation> originalOps = pipeOp.getOps();
+      List<KeyedOperation> newOps = new ArrayList<>();
+      for (KeyedOperation originalOp : originalOps) {
+        if (redirectKeys.contains(originalOp.getKeys().iterator().next())) {
+          newOps.add(originalOp);
+        }
+      }
+      return pipeline(newOps, redirectKeys, mcb);
     } else {
       assert false : "Unhandled operation type: " + op.getClass();
     }

src/main/java/net/spy/memcached/ops/PipelineOperation.java

@@ -0,0 +1,8 @@
+package net.spy.memcached.ops;
+
+public interface PipelineOperation extends KeyedOperation {
+
+  interface Callback extends OperationCallback {
+    void gotStatus(Operation op, OperationStatus status);
+  }
+}

src/main/java/net/spy/memcached/protocol/ascii/AsciiOperationFactory.java

@@ -17,6 +17,7 @@
 package net.spy.memcached.protocol.ascii;
 
 import java.util.Collection;
+import java.util.List;
 
 import javax.security.sasl.SaslClient;
 
@@ -66,9 +67,11 @@
 import net.spy.memcached.ops.GetAttrOperation;
 import net.spy.memcached.ops.GetOperation;
 import net.spy.memcached.ops.GetsOperation;
+import net.spy.memcached.ops.KeyedOperation;
 import net.spy.memcached.ops.Mutator;
 import net.spy.memcached.ops.MutatorOperation;
 import net.spy.memcached.ops.NoopOperation;
+import net.spy.memcached.ops.Operation;
 import net.spy.memcached.ops.OperationCallback;
 import net.spy.memcached.ops.SASLAuthOperation;
 import net.spy.memcached.ops.SASLMechsOperation;
@@ -299,4 +302,10 @@ public BTreeInsertAndGetOperation bopInsertAndGet(String key,
     return new BTreeInsertAndGetOperationImpl(key, get, dataToInsert, cb);
   }
 
+  @Override
+  public Operation pipeline(List<KeyedOperation> ops, List<String> keys,
+                            OperationCallback cb) {
+    return new PipelineOperationImpl(ops, keys, cb);
+  }
+
 }

src/main/java/net/spy/memcached/protocol/ascii/PipelineOperationImpl.java

@@ -0,0 +1,260 @@
+/*
+ * arcus-java-client : Arcus Java client
+ * Copyright 2010-2014 NAVER Corp.
+ * Copyright 2014-present JaM2in Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package net.spy.memcached.protocol.ascii;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import net.spy.memcached.ops.APIType;
+import net.spy.memcached.ops.KeyedOperation;
+import net.spy.memcached.ops.Operation;
+import net.spy.memcached.ops.OperationCallback;
+import net.spy.memcached.ops.OperationState;
+import net.spy.memcached.ops.OperationStatus;
+import net.spy.memcached.ops.OperationType;
+import net.spy.memcached.ops.PipelineOperation;
+import net.spy.memcached.ops.StatusCode;
+
+/**
+ * Operation for executing multiple commands as pipeline.
+ */
+public final class PipelineOperationImpl extends OperationImpl implements PipelineOperation {
+
+  // PIPE RESPONSES
+  private static final OperationStatus END =
+      new OperationStatus(true, "END", StatusCode.SUCCESS);
+  private static final OperationStatus FAILED_END =
+      new OperationStatus(false, "FAILED_END", StatusCode.ERR_FAILED_END);
+
+  // EACH COMMANDS' SUCCEED RESPONSES
+  private static final OperationStatus CREATED_STORED =
+      new OperationStatus(true, "CREATED_STORED", StatusCode.SUCCESS);
+  private static final OperationStatus STORED =
+      new OperationStatus(true, "STORED", StatusCode.SUCCESS);
+  private static final OperationStatus REPLACED =
+      new OperationStatus(true, "REPLACED", StatusCode.SUCCESS);
+  private static final OperationStatus UPDATED =
+      new OperationStatus(true, "UPDATED", StatusCode.SUCCESS);
+  private static final OperationStatus EXIST =
+      new OperationStatus(true, "EXIST", StatusCode.EXIST);
+  private static final OperationStatus NOT_EXIST =
+      new OperationStatus(true, "NOT_EXIST", StatusCode.NOT_EXIST);
+  private static final OperationStatus DELETED =
+      new OperationStatus(true, "DELETED", StatusCode.SUCCESS);
+  private static final OperationStatus DELETED_DROPPED =
+      new OperationStatus(true, "DELETED_DROPPED", StatusCode.SUCCESS);
+
+  // EACH COMMANDS' FAILED RESPONSES
+  private static final OperationStatus NOT_FOUND =
+      new OperationStatus(false, "NOT_FOUND", StatusCode.ERR_NOT_FOUND);
+  private static final OperationStatus NOT_FOUND_ELEMENT =
+      new OperationStatus(false, "NOT_FOUND_ELEMENT", StatusCode.ERR_NOT_FOUND_ELEMENT);
+  private static final OperationStatus NOTHING_TO_UPDATE =
+      new OperationStatus(false, "NOTHING_TO_UPDATE", StatusCode.ERR_NOTHING_TO_UPDATE);
+  private static final OperationStatus ELEMENT_EXISTS =
+      new OperationStatus(false, "ELEMENT_EXISTS", StatusCode.ERR_ELEMENT_EXISTS);
+  private static final OperationStatus OVERFLOWED =
+      new OperationStatus(false, "OVERFLOWED", StatusCode.ERR_OVERFLOWED);
+  private static final OperationStatus OUT_OF_RANGE =
+      new OperationStatus(false, "OUT_OF_RANGE", StatusCode.ERR_OUT_OF_RANGE);
+  private static final OperationStatus TYPE_MISMATCH =
+      new OperationStatus(false, "TYPE_MISMATCH", StatusCode.ERR_TYPE_MISMATCH);
+  private static final OperationStatus BKEY_MISMATCH =
+      new OperationStatus(false, "BKEY_MISMATCH", StatusCode.ERR_BKEY_MISMATCH);
+  private static final OperationStatus EFLAG_MISMATCH =
+      new OperationStatus(false, "EFLAG_MISMATCH", StatusCode.ERR_EFLAG_MISMATCH);
+  private static final OperationStatus UNREADABLE =
+      new OperationStatus(false, "UNREADABLE", StatusCode.ERR_UNREADABLE);
+
+  private final List<KeyedOperation> ops;
+  private final List<String> keys;
+  private final PipelineOperation.Callback cb;
+
+  private int responseIndex = 0;
+  private boolean expectingResponse = false;
+  private boolean successAll = true;
+
+  public PipelineOperationImpl(List<KeyedOperation> ops, List<String> keys,
+                               OperationCallback cb) {
+    super(cb);
+    if (ops == null || ops.isEmpty()) {
+      throw new IllegalArgumentException("Ops cannot be null or empty");
+    }
+    this.ops = ops;
+    this.keys = keys;
+    this.cb = (PipelineOperation.Callback) cb;
+    setAPIType(APIType.PIPE);
+    setOperationType(OperationType.WRITE);
+  }
+
+  @Override
+  public void initialize() {
+    // 1) Initialize operations and collect each buffers
+    // to handle switchover/redirect single key situations,
+    // make buffer from responseIndex Operation
+    int opCount = ops.size() - responseIndex;
+    ByteBuffer[] buffers = new ByteBuffer[opCount];
+    int bufferCount = 0;
+    for (int i = responseIndex; i < ops.size(); i++) {
+      Operation op = ops.get(i);
+      op.initialize();
+      ByteBuffer buffer = op.getBuffer();
+      if (buffer != null && buffer.hasRemaining()) {
+        buffers[bufferCount++] = buffer;
+      }
+    }
+
+    // 2) Remove "pipe" from the last command buffer
+    if (bufferCount > 0) {
+      buffers[bufferCount - 1] = removePipeFromLastBuffer(buffers[bufferCount - 1]);
+    }
+
+    // 3) Create a concatenated pipedBuffer
+    int totalSize = 0;
+    for (int i = 0; i < bufferCount; i++) {
+      totalSize += buffers[i].remaining();
+    }
+
+    ByteBuffer pipedBuffer = ByteBuffer.allocate(totalSize);
+    for (int i = 0; i < bufferCount; i++) {
+      pipedBuffer.put(buffers[i]);
+    }
+
+    pipedBuffer.flip();
+    setBuffer(pipedBuffer);
+  }
+
+  private static ByteBuffer removePipeFromLastBuffer(ByteBuffer buffer) {
+    byte[] bufferBytes = new byte[buffer.remaining()];
+    buffer.mark();
+    buffer.get(bufferBytes);
+    buffer.reset();
+
+    String command = new String(bufferBytes);
+    String modifiedCommand = command.replaceAll("\\s+pipe\\r\\n", "\r\n");
+    byte[] modifiedBytes = modifiedCommand.getBytes();
+    ByteBuffer newBuffer = ByteBuffer.allocate(modifiedBytes.length);
+    newBuffer.put(modifiedBytes);
+    newBuffer.flip();
+    return newBuffer;
+  }
+
+  @Override
+  public void handleLine(String line) {
+
+    /* ENABLE_REPLICATION if */
+    if (hasSwitchedOver(line)) {
+      prepareSwitchover(line);
+      return;
+    }
+    /* ENABLE_REPLICATION end */
+
+    /* ENABLE_MIGRATION if */
+    if (hasNotMyKey(line)) {
+      String key = ops.get(responseIndex).getKeys().iterator().next();
+      if (isBulkOperation()) {
+        addRedirectMultiKeyOperation(line, key);
+        responseIndex++;
+      } else {
+        // Only one NOT_MY_KEY is provided in response of
+        // single key piped operation when redirection.
+        addRedirectSingleKeyOperation(line, key);
+        transitionState(OperationState.REDIRECT);
+      }
+      return;
+    }
+    /* ENABLE_MIGRATION end */
+
+    /*
+      RESPONSE <count>\r\n
+      <status of the 1st pipelined command>\r\n
+      [ ... ]
+      <status of the last pipelined command>\r\n
+      END|PIPE_ERROR <error_string>\r\n
+    */
+    if (line.startsWith("END")) {
+      /* ENABLE_MIGRATION if */
+      if (needRedirect()) {
+        transitionState(OperationState.REDIRECT);
+        return;
+      }
+      /* ENABLE_MIGRATION end */
+
+      OperationStatus status = successAll ? END : FAILED_END;
+      complete(status);
+    } else if (line.startsWith("PIPE_ERROR")) {
+      String errorMessage = line.substring(11);
+      OperationStatus status =
+          new OperationStatus(false, errorMessage, StatusCode.ERR_INTERNAL);
+      complete(status);
+    } else if (line.startsWith("RESPONSE ")) {
+      expectingResponse = true;
+      responseIndex = 0;
+    } else if (expectingResponse) {
+      // Handle status line for each command
+      OperationStatus status = parseStatusLine(line);
+      if (!status.isSuccess()) {
+        successAll = false;
+      }
+
+      // Notify callback with current response index
+      cb.gotStatus(ops.get(responseIndex), status);
+      responseIndex++;
+    } else {
+      // Handle single command response (non-pipe case)
+      // When only one command or last command without "pipe", server sends direct status
+      OperationStatus status = parseStatusLine(line);
+      if (!status.isSuccess()) {
+        successAll = false;
+      }
+
+      // Notify callback for single command
+      cb.gotStatus(ops.get(0), status);
+
+      // Complete the operation immediately for single command
+      complete(successAll ? END : FAILED_END);
+    }
+  }
+
+  private OperationStatus parseStatusLine(String line) {
+    return matchStatus(line,
+        END, FAILED_END, CREATED_STORED, STORED, REPLACED, UPDATED,
+        EXIST, NOT_EXIST, DELETED, DELETED_DROPPED, NOT_FOUND, NOT_FOUND_ELEMENT,
+        NOTHING_TO_UPDATE, ELEMENT_EXISTS, OVERFLOWED, OUT_OF_RANGE,
+        TYPE_MISMATCH, BKEY_MISMATCH, EFLAG_MISMATCH, UNREADABLE);
+  }
+
+  @Override
+  public boolean isPipeOperation() {
+    return ops.size() > 1;
+  }
+
+  @Override
+  public boolean isBulkOperation() {
+    return keys.size() > 1;
+  }
+
+  @Override
+  public List<String> getKeys() {
+    return keys;
+  }
+
+  public List<KeyedOperation> getOps() {
+    return ops;
+  }
+}

src/main/java/net/spy/memcached/protocol/binary/BinaryOperationFactory.java

@@ -17,6 +17,7 @@
 package net.spy.memcached.protocol.binary;
 
 import java.util.Collection;
+import java.util.List;
 
 import javax.security.sasl.SaslClient;
 
@@ -67,9 +68,11 @@
 import net.spy.memcached.ops.GetOperation;
 import net.spy.memcached.ops.GetOperation.Callback;
 import net.spy.memcached.ops.GetsOperation;
+import net.spy.memcached.ops.KeyedOperation;
 import net.spy.memcached.ops.Mutator;
 import net.spy.memcached.ops.MutatorOperation;
 import net.spy.memcached.ops.NoopOperation;
+import net.spy.memcached.ops.Operation;
 import net.spy.memcached.ops.OperationCallback;
 import net.spy.memcached.ops.SASLAuthOperation;
 import net.spy.memcached.ops.SASLMechsOperation;
@@ -329,4 +332,11 @@ public BTreeInsertAndGetOperation bopInsertAndGet(String key,
             "BTree insert and get operation is not supported in binary protocol yet.");
   }
 
+  @Override
+  public Operation pipeline(List<KeyedOperation> ops, List<String> keys,
+                            OperationCallback cb) {
+    throw new RuntimeException(
+        "Pipeline operation is not supported in binary protocol yet.");
+  }
+
 }