feature: support set nx ex protocol

src/main/java/org/redis2asp/protocol/RedisCommandDecoder.java

@@ -44,22 +44,22 @@ public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
         List<String> params = new ArrayList<>(length);
         for (int i = 0; i < length; i++) {
             String param = readParam(in);
-            params.add(param);
+            params.add(param.toLowerCase());
         }
         // convert to RedisRequest
         out.add(convert2RedisRequest(params));
     }
 
     private RedisRequest<?> convert2RedisRequest(List<String> params) {
         String cmd = params.get(0);
-        LOGGER.info("cmd: {}", cmd);
-        switch (cmd.toLowerCase()) {
+        LOGGER.info("cmd: {}", params);
+        switch (cmd) {
             case "get":
                 return new GetRequest(params.get(1));
             case "command":
                 return new CommandRequest();
             case "set":
-                return new SetRequest(params.get(1), params.get(2));
+                return new SetRequest(params.get(1), params.get(2), params);
             default:
                 return null;
         }

src/main/java/org/redis2asp/protocol/RedisCommandHandler.java

@@ -25,6 +25,8 @@
 import com.aerospike.client.Record;
 import com.aerospike.client.listener.RecordListener;
 import com.aerospike.client.listener.WriteListener;
+import com.aerospike.client.policy.RecordExistsAction;
+import com.aerospike.client.policy.WritePolicy;
 import com.alipay.remoting.CommandCode;
 import com.alipay.remoting.CommandHandler;
 import com.alipay.remoting.RemotingContext;
@@ -47,6 +49,7 @@ public class RedisCommandHandler implements CommandHandler {
     public void handleCommand(RemotingContext ctx, Object msg) {
         if (msg instanceof RedisRequest) {
             RedisRequest<?> redisRequest = (RedisRequest) msg;
+            logger.info("redisRequest:{}", redisRequest);
             if (redisRequest instanceof GetRequest) {
                 GetRequest getRequest = (GetRequest) redisRequest;
                 Key key = new Key(AeroSpikeClientFactory.namespace, AeroSpikeClientFactory.set, getRequest.getKey());
@@ -67,7 +70,6 @@ public void onSuccess(Key key, Record record) {
                     @Override
                     public void onFailure(AerospikeException ae) {
                         logger.error(ae.getMessage(), ae);
-                        getRequest.setResponse(ae.getMessage().getBytes(StandardCharsets.UTF_8));
                         ctx.writeAndFlush(redisRequest.getResponse());
                     }
                 }, client.getReadPolicyDefault(), key);
@@ -76,6 +78,57 @@ public void onFailure(AerospikeException ae) {
                 SetRequest setRequest = (SetRequest) redisRequest;
                 Bin bin = new Bin(setRequest.getKey(), setRequest.getValue());
                 Key key = new Key(AeroSpikeClientFactory.namespace, AeroSpikeClientFactory.set, setRequest.getKey());
+                WritePolicy writePolicy = null;
+                if (setRequest.getTtl() != null) {
+                    writePolicy = new WritePolicy(client.getWritePolicyDefault());
+                    if (setRequest.getTtlType() == SetRequest.TtlType.EX) {
+                        writePolicy.expiration = setRequest.getTtl().intValue();
+                    } else {
+                        writePolicy.expiration = Integer.max((int) (setRequest.getTtl() / 1000), 1);
+                    }
+                }
+                if (setRequest.getOperate() != null) {
+                    if (writePolicy == null) {
+                        writePolicy = new WritePolicy(client.getWritePolicyDefault());
+                    }
+                    if (setRequest.getOperate() == SetRequest.Operate.NX) {
+                        writePolicy.recordExistsAction = RecordExistsAction.CREATE_ONLY;
+                    }
+                    if (setRequest.getOperate() == SetRequest.Operate.XX) {
+                        client.get(null, new RecordListener() {
+                            @Override
+                            public void onSuccess(Key key, Record record) {
+                                if (record == null) {
+                                    ctx.writeAndFlush(redisRequest.getResponse());
+                                } else {
+                                    client.put(null, new WriteListener() {
+                                        @Override
+                                        public void onSuccess(Key key) {
+                                            setRequest.setResponse("OK".getBytes(StandardCharsets.UTF_8));
+                                            ctx.writeAndFlush(redisRequest.getResponse());
+                                        }
+
+                                        @Override
+                                        public void onFailure(AerospikeException ae) {
+                                            logger.error(ae.getMessage(), ae);
+                                            ctx.writeAndFlush(redisRequest.getResponse());
+                                        }
+                                    }, client.getWritePolicyDefault(), key, bin);
+                                }
+                            }
+
+                            @Override
+                            public void onFailure(AerospikeException ae) {
+                                logger.error(ae.getMessage(), ae);
+                                ctx.writeAndFlush(redisRequest.getResponse());
+                            }
+                        }, client.getReadPolicyDefault(), key);
+                        return;
+                    }
+                }
+                if (writePolicy == null) {
+                    writePolicy = client.getWritePolicyDefault();
+                }
                 client.put(null, new WriteListener() {
                     @Override
                     public void onSuccess(Key key) {
@@ -85,10 +138,10 @@ public void onSuccess(Key key) {
 
                     @Override
                     public void onFailure(AerospikeException ae) {
-                        setRequest.setResponse(ae.getMessage().getBytes(StandardCharsets.UTF_8));
+                        logger.error(ae.getMessage(), ae);
                         ctx.writeAndFlush(redisRequest.getResponse());
                     }
-                }, client.getWritePolicyDefault(), key, bin);
+                }, writePolicy, key, bin);
             }
             if (redisRequest instanceof CommandRequest) {
                 CommandRequest commandRequest = (CommandRequest) redisRequest;
@@ -100,12 +153,10 @@ public void onFailure(AerospikeException ae) {
 
     @Override
     public void registerProcessor(CommandCode cmd, RemotingProcessor<?> processor) {
-
     }
 
     @Override
     public void registerDefaultExecutor(ExecutorService executor) {
-
     }
 
     @Override

src/main/java/org/redis2asp/protocol/request/CommandRequest.java

@@ -1,3 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.redis2asp.protocol.request;
 
 import org.redis2asp.protocol.RedisRequest;
@@ -17,4 +33,9 @@ public RedisResponse<byte[]> getResponse() {
     public void setResponse(byte[] data) {
         response.setData(data);
     }
+
+    @Override
+    public String toString() {
+        return "CommandRequest{" + "response=" + response + '}';
+    }
 }

src/main/java/org/redis2asp/protocol/request/GetRequest.java

@@ -44,4 +44,8 @@ public RedisResponse<byte[]> getResponse() {
         return response;
     }
 
+    @Override
+    public String toString() {
+        return "GetRequest{" + "key='" + key + '\'' + ", response=" + response + '}';
+    }
 }

src/main/java/org/redis2asp/protocol/request/SetRequest.java

@@ -17,69 +17,62 @@
 package org.redis2asp.protocol.request;
 
 import java.nio.charset.StandardCharsets;
+import java.util.List;
 import org.redis2asp.protocol.RedisRequest;
 import org.redis2asp.protocol.RedisResponse;
 import org.redis2asp.protocol.response.BulkResponse;
 
 public class SetRequest implements RedisRequest<byte[]> {
 
-    String       key;
+    final String key;
 
-    String       value;
+    final String value;
 
     TtlType      ttlType;
 
-    long         ttl;
+    Long         ttl;
 
     Operate      operate;
 
     BulkResponse response = new BulkResponse();
 
-    public SetRequest(String key, String value) {
+    public SetRequest(String key, String value, List<String> params) {
         this.key = key;
         this.value = value;
+        if (params.contains("nx")) {
+            this.operate = Operate.NX;
+        } else if (params.contains("xx")) {
+            this.operate = Operate.XX;
+        }
+        if (params.contains("ex")) {
+            this.ttlType = TtlType.EX;
+            this.ttl = Long.parseLong(params.get(params.indexOf("ex") + 1));
+        } else if (params.contains("px")) {
+            this.ttlType = TtlType.PX;
+            this.ttl = Long.parseLong(params.get(params.indexOf("px") + 1));
+        }
     }
 
     public String getKey() {
         return key;
     }
 
-    public void setKey(String key) {
-        this.key = key;
-    }
-
     public String getValue() {
         return value;
     }
 
-    public void setValue(String value) {
-        this.value = value;
-    }
-
     public TtlType getTtlType() {
         return ttlType;
     }
 
-    public void setTtlType(String ttlType) {
-        this.ttlType = TtlType.valueOf(ttlType.toUpperCase());
-    }
-
-    public long getTtl() {
+    public Long getTtl() {
         return ttl;
     }
 
-    public void setTtl(long ttl) {
-        this.ttl = ttl;
-    }
-
     public Operate getOperate() {
         return operate;
     }
 
-    public void setOperate(String operate) {
-        this.operate = Operate.valueOf(operate.toUpperCase());
-    }
-
     @Override
     public void setResponse(byte[] data) {
         this.response.setData(data);
@@ -90,12 +83,17 @@ public RedisResponse<byte[]> getResponse() {
         return response;
     }
 
-    static enum TtlType {
+    public enum TtlType {
         EX, PX
     }
 
-    static enum Operate {
+    public enum Operate {
         NX, XX
     }
 
+    @Override
+    public String toString() {
+        return "SetRequest{" + "key='" + key + '\'' + ", value='" + value + '\'' + ", ttlType=" + ttlType + ", ttl="
+               + ttl + ", operate=" + operate + ", response=" + response + '}';
+    }
 }

src/main/java/org/redis2asp/protocol/response/BulkResponse.java

@@ -1,3 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.redis2asp.protocol.response;
 
 import java.io.IOException;

src/main/java/org/redis2asp/protocol/response/IntegerResponse.java

@@ -1,3 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.redis2asp.protocol.response;
 
 import java.io.IOException;

src/test/java/org/redis2asp/ServerTest.java

@@ -32,6 +32,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import redis.clients.jedis.Jedis;
+import redis.clients.jedis.params.SetParams;
 
 public class ServerTest {
     static Server           server;
@@ -81,6 +82,45 @@ public void testGetNilAsp() {
         }
     }
 
+    @Test
+    public void testSetExAsp() {
+        String key = String.valueOf(ThreadLocalRandom.current().nextInt(50000));
+        try (Jedis jedis = new Jedis("127.0.0.1", 6789)) {
+            String result = jedis.set(key, "b", SetParams.setParams().ex(1L));
+            Assertions.assertEquals(result, "OK");
+            Thread.sleep(3000);
+            result = jedis.get(key);
+            Assertions.assertNull(result);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Test
+    public void testSetNxNilAsp() {
+        String key = String.valueOf(ThreadLocalRandom.current().nextInt(50000));
+        try (Jedis jedis = new Jedis("127.0.0.1", 6789)) {
+            String result = jedis.set(key, "b", SetParams.setParams().nx());
+            Assertions.assertEquals(result, "OK");
+            result = jedis.set(key, "b", SetParams.setParams().nx());
+            Assertions.assertNull(result);
+        }
+    }
+
+    @Test
+    public void testSetExNxAsp() {
+        String key = String.valueOf(ThreadLocalRandom.current().nextInt(50000));
+        try (Jedis jedis = new Jedis("127.0.0.1", 6789)) {
+            String result = jedis.set(key, "b", SetParams.setParams().nx().ex(1L));
+            Assertions.assertEquals(result, "OK");
+            Thread.sleep(3000);
+            result = jedis.get(key);
+            Assertions.assertNull(result);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     @AfterAll
     public static void shutdown() {
         Optional.ofNullable(server).ifPresent(Server::shutdown);