Skip to content

Commit 2eafdd5

Browse files
singsangssongsingsangssong
committed
FEAT: Add gat(get and touch) command to get item & update item expiration
1 parent 261bbdc commit 2eafdd5

13 files changed

Lines changed: 306 additions & 4 deletions

File tree

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

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -247,6 +247,26 @@ public GetFuture<CASValue<Object>> asyncGets(String key) {
247247
return this.getClient().asyncGets(key);
248248
}
249249

250+
@Override
251+
public <T> GetFuture<T> asyncGetAndTouch(String key, int exp, Transcoder<T> tc) {
252+
return this.getClient().asyncGetAndTouch(key, exp, tc);
253+
}
254+
255+
@Override
256+
public GetFuture<Object> asyncGetAndTouch(String key, int exp) {
257+
return this.getClient().asyncGetAndTouch(key, exp);
258+
}
259+
260+
@Override
261+
public <T> GetFuture<CASValue<T>> asyncGetsAndTouch(String key, int exp, Transcoder<T> tc) {
262+
return this.getClient().asyncGetsAndTouch(key, exp, tc);
263+
}
264+
265+
@Override
266+
public GetFuture<CASValue<Object>> asyncGetsAndTouch(String key, int exp) {
267+
return this.getClient().asyncGetsAndTouch(key, exp);
268+
}
269+
250270
@Override
251271
public <T> CASValue<T> gets(String key, Transcoder<T> tc)
252272
throws OperationTimeoutException {

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

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -980,6 +980,106 @@ public GetFuture<CASValue<Object>> asyncGets(final String key) {
980980
return asyncGets(key, transcoder);
981981
}
982982

983+
/**
984+
* Get the given key to reset its expiration time asynchronously.
985+
*
986+
* @param key the key to fetch
987+
* @param exp the new expiration to set for the given key
988+
* @param tc the transcoder to serialize and unserialize value
989+
* @return a future that will hold the return value of the fetch
990+
* @throws IllegalStateException in the rare circumstance where queue is too
991+
* full to accept any more requests
992+
*/
993+
public <T> GetFuture<T> asyncGetAndTouch(final String key, final int exp,
994+
final Transcoder<T> tc) {
995+
final CountDownLatch latch = new CountDownLatch(1);
996+
final GetFuture<T> future = new GetFuture<>(latch, operationTimeout);
997+
998+
Operation op = opFact.getAndTouch(key, exp,
999+
new GetOperation.Callback() {
1000+
private GetResult<T> result = null;
1001+
1002+
public void receivedStatus(OperationStatus status) {
1003+
future.set(result, status);
1004+
}
1005+
1006+
public void gotData(String k, int flags, byte[] data) {
1007+
assert k.equals(key) : "Wrong key returned";
1008+
result = new GetResultImpl<>(new CachedData(flags, data, tc.getMaxSize()), tc);
1009+
}
1010+
1011+
public void complete() {
1012+
latch.countDown();
1013+
}
1014+
});
1015+
future.setOperation(op);
1016+
addOp(key, op);
1017+
return future;
1018+
}
1019+
1020+
/**
1021+
* Get the given key to reset its expiration time asynchronously.
1022+
*
1023+
* @param key the key to fetch
1024+
* @param exp the new expiration to set for the given key
1025+
* @return a future that will hold the return value of the fetch
1026+
* @throws IllegalStateException in the rare circumstance where queue is too
1027+
* full to accept any more requests
1028+
*/
1029+
public GetFuture<Object> asyncGetAndTouch(final String key, final int exp) {
1030+
return asyncGetAndTouch(key, exp, transcoder);
1031+
}
1032+
1033+
/**
1034+
* Gets (with CAS support) the given key to reset its expiration time asynchronously.
1035+
*
1036+
* @param key the key to fetch
1037+
* @param exp the new expiration to set for the given key
1038+
* @param tc the transcoder to serialize and unserialize value
1039+
* @return a future that will hold the return value of the fetch
1040+
* @throws IllegalStateException in the rare circumstance where queue is too
1041+
* full to accept any more requests
1042+
*/
1043+
public <T> GetFuture<CASValue<T>> asyncGetsAndTouch(final String key, final int exp,
1044+
final Transcoder<T> tc) {
1045+
final CountDownLatch latch = new CountDownLatch(1);
1046+
final GetFuture<CASValue<T>> rv = new GetFuture<>(latch, operationTimeout);
1047+
1048+
Operation op = opFact.getsAndTouch(key, exp, new GetsOperation.Callback() {
1049+
private GetResult<CASValue<T>> val = null;
1050+
1051+
public void receivedStatus(OperationStatus status) {
1052+
rv.set(val, status);
1053+
}
1054+
1055+
public void gotData(String k, int flags, long cas, byte[] data) {
1056+
assert key.equals(k) : "Wrong key returned";
1057+
assert cas > 0 : "CAS was less than zero: " + cas;
1058+
val = new GetsResultImpl<>(cas, new CachedData(flags, data, tc.getMaxSize()), tc);
1059+
}
1060+
1061+
public void complete() {
1062+
latch.countDown();
1063+
}
1064+
});
1065+
rv.setOperation(op);
1066+
addOp(key, op);
1067+
return rv;
1068+
}
1069+
1070+
/**
1071+
* Gets (with CAS support) the given key to reset its expiration time asynchronously.
1072+
*
1073+
* @param key the key to fetch
1074+
* @param exp the new expiration to set for the given key
1075+
* @return a future that will hold the return value of the fetch
1076+
* @throws IllegalStateException in the rare circumstance where queue is too
1077+
* full to accept any more requests
1078+
*/
1079+
public GetFuture<CASValue<Object>> asyncGetsAndTouch(final String key, final int exp) {
1080+
return asyncGetsAndTouch(key, exp, transcoder);
1081+
}
1082+
9831083
/**
9841084
* Gets (with CAS support) with a single key.
9851085
*

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,15 @@ <T> Future<CASValue<T>> asyncGets(String key,
8282

8383
Future<CASValue<Object>> asyncGets(String key);
8484

85+
<T> Future<T> asyncGetAndTouch(final String key, final int exp, final Transcoder<T> tc);
86+
87+
Future<Object> asyncGetAndTouch(final String key, final int exp);
88+
89+
<T> Future<CASValue<T>> asyncGetsAndTouch(final String key, final int exp,
90+
final Transcoder<T> tc);
91+
92+
Future<CASValue<Object>> asyncGetsAndTouch(final String key, final int exp);
93+
8594
<T> CASValue<T> gets(String key, Transcoder<T> tc)
8695
throws OperationTimeoutException;
8796

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

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,26 @@ public interface OperationFactory {
152152
*/
153153
GetsOperation gets(Collection<String> keys, GetsOperation.Callback cb, boolean isMGet);
154154

155+
/**
156+
* Get the key and resets its timeout.
157+
*
158+
* @param key the key to get a value for and reset its timeout
159+
* @param expiration the new expiration for the key
160+
* @param cb the callback that will contain the result
161+
* @return a new GetAndTouchOperation
162+
*/
163+
GetOperation getAndTouch(String key, int expiration, GetOperation.Callback cb);
164+
165+
/**
166+
* Gets (with CAS support) the key and resets its timeout.
167+
*
168+
* @param key the key to get a value for and reset its timeout
169+
* @param expiration the new expiration for the key
170+
* @param cb the callback that will contain the result
171+
* @return a new GetsAndTouchOperation
172+
*/
173+
GetsOperation getsAndTouch(String key, int expiration, GetsOperation.Callback cb);
174+
155175
/**
156176
* Create a mutator operation.
157177
*

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ public enum APIType {
2525
INCR(OperationType.WRITE), DECR(OperationType.WRITE),
2626
DELETE(OperationType.WRITE),
2727
GET(OperationType.READ), GETS(OperationType.READ),
28+
GAT(OperationType.WRITE), GATS(OperationType.WRITE),
2829

2930
// List API Type
3031
LOP_CREATE(OperationType.WRITE),

‎src/main/java/net/spy/memcached/ops/OperationType.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ public enum OperationType {
2525
* BTreeInsertAndGetOperationImpl (asyncBopInsertAndGetTrimmed)
2626
* FlushOperationImpl / FlushByPrefixOperationImpl (flush)
2727
* SetAttrOperationImpl (asyncSetAttr)
28+
* GetAndTouchOperationImpl / GetsAndTouchOperationImpl (asyncGetAndTouch / asyncGetsAndTouch)
2829
*/
2930
WRITE,
3031

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,14 @@ public GetsOperation gets(Collection<String> keys, GetsOperation.Callback cb, bo
109109
return new GetsOperationImpl(keys, cb, isMGet);
110110
}
111111

112+
public GetOperation getAndTouch(String key, int expiration, GetOperation.Callback cb) {
113+
return new GetAndTouchOperationImpl(key, expiration, cb);
114+
}
115+
116+
public GetsOperation getsAndTouch(String key, int expiration, GetsOperation.Callback cb) {
117+
return new GetsAndTouchOperationImpl(key, expiration, cb);
118+
}
119+
112120
public MutatorOperation mutate(Mutator m, String key, int by,
113121
long def, int exp, OperationCallback cb) {
114122
return new MutatorOperationImpl(m, key, by, def, exp, cb);

‎src/main/java/net/spy/memcached/protocol/ascii/BaseGetOpImpl.java‎

Lines changed: 33 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ abstract class BaseGetOpImpl extends OperationImpl {
4141
private final String cmd;
4242
private final Collection<String> keys;
4343
private String currentKey = null;
44+
protected final int exp;
4445
private long casValue = 0;
4546
private int currentFlags = 0;
4647
private byte[] data = null;
@@ -55,9 +56,19 @@ public BaseGetOpImpl(String c,
5556
super(cb);
5657
cmd = c;
5758
keys = k;
59+
exp = 0;
5860
setOperationType(OperationType.READ);
5961
}
6062

63+
public BaseGetOpImpl(String c, int e,
64+
OperationCallback cb, Collection<String> k) {
65+
super(cb);
66+
cmd = c;
67+
keys = k;
68+
exp = e;
69+
setOperationType(OperationType.WRITE);
70+
}
71+
6172
/**
6273
* Get the keys this GetOperation is looking for.
6374
*/
@@ -73,6 +84,10 @@ public final void handleLine(String line) {
7384
...
7485
END\r\n
7586
*/
87+
if (hasSwitchedOver(line)) {
88+
prepareSwitchover(line);
89+
return;
90+
}
7691
if (line.equals("END")) {
7792
getLogger().debug("Get complete!");
7893
/* ENABLE_MIGRATION if */
@@ -131,12 +146,15 @@ public final void handleRead(ByteBuffer bb) {
131146
if (readOffset == data.length && lookingFor == '\0') {
132147
// The callback is most likely a get callback. If it's not, then
133148
// it's a gets callback.
134-
try {
135-
GetOperation.Callback gcb = (GetOperation.Callback) getCallback();
149+
OperationCallback cb = getCallback();
150+
if (cb instanceof GetOperation.Callback) {
151+
GetOperation.Callback gcb = (GetOperation.Callback) cb;
136152
gcb.gotData(currentKey, currentFlags, data);
137-
} catch (ClassCastException e) {
138-
GetsOperation.Callback gcb = (GetsOperation.Callback) getCallback();
153+
} else if (cb instanceof GetsOperation.Callback) {
154+
GetsOperation.Callback gcb = (GetsOperation.Callback) cb;
139155
gcb.gotData(currentKey, currentFlags, casValue, data);
156+
} else {
157+
throw new ClassCastException("Couldn't convert " + cb + "to a relevent op");
140158
}
141159
lookingFor = '\r';
142160
}
@@ -184,6 +202,14 @@ public final void initialize() {
184202
commandBuilder.append(' ');
185203
commandBuilder.append(keysString);
186204
commandBuilder.append(RN_STRING);
205+
} else if (cmd.equals("gat") || cmd.equals("gats")) {
206+
// syntax: gat || gats <exp> <key>\r\n
207+
commandBuilder.append(cmd);
208+
commandBuilder.append(' ');
209+
commandBuilder.append(exp);
210+
commandBuilder.append(' ');
211+
commandBuilder.append(keysString);
212+
commandBuilder.append(RN_STRING);
187213
} else {
188214
assert (cmd.equals("mget") || cmd.equals("mgets"))
189215
: "Unknown Command " + cmd;
@@ -231,4 +257,7 @@ public boolean isBulkOperation() {
231257
return keys.size() > 1;
232258
}
233259

260+
public int getExpiration() {
261+
return exp;
262+
}
234263
}

‎src/main/java/net/spy/memcached/protocol/ascii/GetAndTouchOperationImpl.java‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/*
2+
* arcus-java-client : Arcus Java client
3+
* Copyright 2010-2014 NAVER Corp.
4+
* Copyright 2014-present JaM2in Co., Ltd.
5+
*
6+
* Licensed under the Apache License, Version 2.0 (the "License");
7+
* you may not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package net.spy.memcached.protocol.ascii;
20+
21+
import java.util.Collections;
22+
23+
import net.spy.memcached.ops.APIType;
24+
import net.spy.memcached.ops.GetOperation;
25+
26+
/**
27+
* Implementation of the get and touch operation.
28+
*/
29+
class GetAndTouchOperationImpl extends BaseGetOpImpl implements GetOperation {
30+
private static final String CMD = "gat";
31+
32+
public GetAndTouchOperationImpl(String k, int e, GetOperation.Callback cb) {
33+
super(CMD, e, cb, Collections.singleton(k));
34+
setAPIType(APIType.GAT);
35+
}
36+
}

‎src/main/java/net/spy/memcached/protocol/ascii/GetsAndTouchOperationImpl.java‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/*
2+
* arcus-java-client : Arcus Java client
3+
* Copyright 2010-2014 NAVER Corp.
4+
* Copyright 2014-present JaM2in Co., Ltd.
5+
*
6+
* Licensed under the Apache License, Version 2.0 (the "License");
7+
* you may not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package net.spy.memcached.protocol.ascii;
20+
21+
import java.util.Collections;
22+
23+
import net.spy.memcached.ops.APIType;
24+
import net.spy.memcached.ops.GetsOperation;
25+
26+
/**
27+
* Implementation of the gets and touch operation.
28+
*/
29+
class GetsAndTouchOperationImpl extends BaseGetOpImpl implements GetsOperation {
30+
private static final String CMD = "gats";
31+
32+
public GetsAndTouchOperationImpl(String k, int e, GetsOperation.Callback cb) {
33+
super(CMD, e, cb, Collections.singleton(k));
34+
setAPIType(APIType.GATS);
35+
}
36+
}

0 commit comments

Comments
 (0)