FEAT: Add gat(get and touch) command to get item & update item expiration

Author: singsangssong

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

@@ -242,6 +242,16 @@ public GetFuture<CASValue<Object>> asyncGets(String key) {
     return this.getClient().asyncGets(key);
   }
 
+  @Override
+  public Future<CASValue<Object>> asyncGetAndTouch(String key, int exp) {
+    return this.getClient().asyncGetAndTouch(key, exp);
+  }
+
+  @Override
+  public <T> Future<CASValue<T>> asyncGetAndTouch(String key, int exp, Transcoder<T> tc) {
+    return this.getClient().asyncGetAndTouch(key, exp, tc);
+  }
+
   @Override
   public <T> CASValue<T> gets(String key, Transcoder<T> tc)
           throws OperationTimeoutException {
@@ -264,6 +274,16 @@ public Object get(String key) throws OperationTimeoutException {
     return this.getClient().get(key);
   }
 
+  @Override
+  public CASValue<Object> getAndTouch(String key, int exp) {
+    return this.getClient().getAndTouch(key, exp);
+  }
+
+  @Override
+  public <T> CASValue<T> getAndTouch(String key, int exp, Transcoder<T> tc) {
+    return this.getClient().getAndTouch(key, exp, tc);
+  }
+
   @Override
   public <T> BulkFuture<Map<String, T>> asyncGetBulk(Collection<String> keys,
                                                      Iterator<Transcoder<T>> tcs) {

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

@@ -32,6 +32,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
@@ -58,6 +59,7 @@
 import net.spy.memcached.ops.CancelledOperationStatus;
 import net.spy.memcached.ops.ConcatenationType;
 import net.spy.memcached.ops.DeleteOperation;
+import net.spy.memcached.ops.GetAndTouchOperation;
 import net.spy.memcached.ops.GetOperation;
 import net.spy.memcached.ops.GetsOperation;
 import net.spy.memcached.ops.Mutator;
@@ -980,6 +982,58 @@ public GetFuture<CASValue<Object>> asyncGets(final String key) {
     return asyncGets(key, transcoder);
   }
 
+  /**
+   * Get the given key to reset its expiration time.
+   *
+   * @param key the key to fetch
+   * @param exp the new expiration to set for the given key
+   * @param tc the transcoder to serialize and unserialize value
+   * @return a future that will hold the return value of the fetch
+   * @throws IllegalStateException in the rare circumstance where queue is too
+   *           full to accept any more requests
+   */
+  public <T> OperationFuture<CASValue<T>> asyncGetAndTouch(final String key,
+                                                           final int exp, final Transcoder<T> tc) {
+    final CountDownLatch latch = new CountDownLatch(1);
+    final OperationFuture<CASValue<T>> rv =
+            new OperationFuture<CASValue<T>>(latch, operationTimeout);
+
+    Operation op = opFact.getAndTouch(key, exp,
+        new GetAndTouchOperation.Callback() {
+          private CASValue<T> val;
+
+          public void receivedStatus(OperationStatus status) {
+            rv.set(val, status);
+          }
+
+          public void complete() {
+            latch.countDown();
+          }
+
+          public void gotData(String k, int flags, long cas, byte[] data) {
+            assert k.equals(key) : "Wrong key returned";
+            val = new CASValue<T>(cas, tc.decode(new CachedData(flags, data, tc.getMaxSize())));
+          }
+        });
+    rv.setOperation(op);
+    addOp(key, op);
+    return rv;
+  }
+
+  /**
+   * Get the given key to reset its expiration time.
+   *
+   * @param key the key to fetch
+   * @param exp the new expiration to set for the given key
+   * @return a future that will hold the return value of the fetch
+   * @throws IllegalStateException in the rare circumstance where queue is too
+   *           full to accept any more requests
+   */
+  public OperationFuture<CASValue<Object>> asyncGetAndTouch(final String key,
+                                                            final int exp) {
+    return asyncGetAndTouch(key, exp, transcoder);
+  }
+
   /**
    * Gets (with CAS support) with a single key.
    *
@@ -1042,6 +1096,52 @@ public Object get(String key) {
     return get(key, transcoder);
   }
 
+  /**
+   * Get with a single key and reset its expiration.
+   *
+   * @param <T>
+   * @param key the key to get
+   * @param exp the new expiration for the key
+   * @param tc the transcoder to serialize and unserialize value
+   * @return the result from the cache (null if there is none)
+   * @throws OperationTimeoutException if the global operation timeout is
+   *           exceeded
+   * @throws java.util.concurrent.CancellationException if operation was canceled
+   * @throws IllegalStateException in the rare circumstance where queue is too
+   *           full to accept any more requests
+   */
+  public <T> CASValue<T> getAndTouch(String key, int exp, Transcoder<T> tc) {
+    try {
+      return asyncGetAndTouch(key, exp, tc).get(operationTimeout,
+              TimeUnit.MILLISECONDS);
+    } catch (InterruptedException e) {
+      throw new RuntimeException("Interrupted waiting for value", e);
+    } catch (ExecutionException e) {
+      if (e.getCause() instanceof CancellationException) {
+        throw (CancellationException) e.getCause();
+      } else {
+        throw new RuntimeException("Exception waiting for value", e);
+      }
+    } catch (TimeoutException e) {
+      throw new OperationTimeoutException("Timeout waiting for value", e);
+    }
+  }
+
+  /**
+   * Get a single key and reset its expiration using the default transcoder.
+   *
+   * @param key the key to get
+   * @param exp the new expiration for the key
+   * @return the result from the cache and CAS id (null if there is none)
+   * @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 CASValue<Object> getAndTouch(String key, int exp) {
+    return getAndTouch(key, exp, transcoder);
+  }
+
   /**
    * Asynchronously get a bunch of objects from the cache.
    *

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

@@ -82,6 +82,10 @@ <T> Future<CASValue<T>> asyncGets(String key,
 
   Future<CASValue<Object>> asyncGets(String key);
 
+  Future<CASValue<Object>> asyncGetAndTouch(final String key, final int exp);
+
+  <T> Future<CASValue<T>> asyncGetAndTouch(final String key, final int exp, final Transcoder<T> tc);
+
   <T> CASValue<T> gets(String key, Transcoder<T> tc)
           throws OperationTimeoutException;
 
@@ -92,6 +96,10 @@ <T> T get(String key, Transcoder<T> tc)
 
   Object get(String key) throws OperationTimeoutException;
 
+  CASValue<Object> getAndTouch(String key, int exp);
+
+  <T> CASValue<T> getAndTouch(String key, int exp, Transcoder<T> tc);
+
   <T> BulkFuture<Map<String, T>> asyncGetBulk(Collection<String> keys,
                                               Iterator<Transcoder<T>> tcs);
 

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

@@ -64,6 +64,7 @@
 import net.spy.memcached.ops.ConcatenationType;
 import net.spy.memcached.ops.DeleteOperation;
 import net.spy.memcached.ops.FlushOperation;
+import net.spy.memcached.ops.GetAndTouchOperation;
 import net.spy.memcached.ops.GetAttrOperation;
 import net.spy.memcached.ops.GetOperation;
 import net.spy.memcached.ops.GetsOperation;
@@ -142,6 +143,17 @@ public interface OperationFactory {
    */
   GetOperation get(Collection<String> keys, GetOperation.Callback cb, boolean isMGet);
 
+  /**
+   * Gets the value of a key and resets its timeout.
+   *
+   * @param key the key to get a value for and reset its timeout
+   * @param expiration the new expiration for the key
+   * @param cb the callback that will contain the result
+   * @return a new GATOperation
+   */
+  GetAndTouchOperation getAndTouch(String key, int expiration,
+                                   GetAndTouchOperation.Callback cb);
+
   /**
    * Create a gets operation.
    *

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

@@ -88,6 +88,10 @@ public Collection<Operation> clone(KeyedOperation op) {
       GetAttrOperation c = (GetAttrOperation) op;
       rv.add(getAttr(first(c.getKeys()),
               (GetAttrOperation.Callback) c.getCallback()));
+    } else if (op instanceof GetAndTouchOperation) {
+      GetAndTouchOperation gt = (GetAndTouchOperation) op;
+      rv.add(getAndTouch(first(gt.getKeys()), gt.getExpiration(),
+              (GetAndTouchOperation.Callback) gt.getCallback()));
     } else if (op instanceof CollectionInsertOperation) {
       CollectionInsertOperation c = (CollectionInsertOperation) op;
       rv.add(collectionInsert(first(c.getKeys()), c.getSubKey(),

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

@@ -0,0 +1,44 @@
+/*
+ * 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.ops;
+
+/**
+ * Gat operation.
+ */
+public interface GetAndTouchOperation extends KeyedOperation {
+
+  /**
+   * Operation callback for the gat request.
+   */
+  interface Callback extends OperationCallback {
+    /**
+     * Callback for each result from a gat.
+     *
+     * @param key the key that was retrieved
+     * @param flags the flags for this value
+     * @param data the data stored under this key
+     */
+    void gotData(String key, int flags, long cas, byte[] data);
+  }
+
+  /**
+   * Get the expiration to set in case of a new entry.
+   */
+  int getExpiration();
+}

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

@@ -64,6 +64,7 @@
 import net.spy.memcached.ops.ConcatenationType;
 import net.spy.memcached.ops.DeleteOperation;
 import net.spy.memcached.ops.FlushOperation;
+import net.spy.memcached.ops.GetAndTouchOperation;
 import net.spy.memcached.ops.GetAttrOperation;
 import net.spy.memcached.ops.GetOperation;
 import net.spy.memcached.ops.GetsOperation;
@@ -109,6 +110,11 @@ public GetsOperation gets(Collection<String> keys, GetsOperation.Callback cb, bo
     return new GetsOperationImpl(keys, cb, isMGet);
   }
 
+  public GetAndTouchOperation getAndTouch(String key, int expiration,
+                                          GetAndTouchOperation.Callback cb) {
+    return new GetAndTouchOperationImpl(key, expiration, cb, key);
+  }
+
   public MutatorOperation mutate(Mutator m, String key, int by,
                                  long def, int exp, OperationCallback cb) {
     return new MutatorOperationImpl(m, key, by, def, exp, cb);