feat: support deadletter apis

Yesphet · Yesphet · commit 145c821765df · 2019-12-13T11:26:44.000+08:00
diff --git a/src/main/java/com/meitu/platform/lmstfy/client/LmstfyClient.java b/src/main/java/com/meitu/platform/lmstfy/client/LmstfyClient.java
@@ -2,12 +2,10 @@
 
 import com.meitu.platform.lmstfy.exception.LmstfyException;
 import com.meitu.platform.lmstfy.exception.LmstfyIllegalRequestException;
+import com.meitu.platform.lmstfy.exception.LmstfyNotJobException;
 import com.meitu.platform.lmstfy.exception.LmstfyUnexpectedException;
-import com.meitu.platform.lmstfy.response.ErrorResponse;
+import com.meitu.platform.lmstfy.response.*;
 import com.meitu.platform.lmstfy.Job;
-import com.meitu.platform.lmstfy.response.LmstfyResponse;
-import com.meitu.platform.lmstfy.response.PublishResponse;
-import com.meitu.platform.lmstfy.response.QueueSizeResponse;
 import okhttp3.*;
 
 import java.io.IOException;
@@ -31,6 +29,7 @@ public class LmstfyClient {
     private static final String QUERY_TRIES = "tries";
     private static final String QUERY_TIMEOUT = "timeout";
     private static final String QUERY_TTR = "ttr";
+    private static final String QUERY_LIMIT = "limit";
 
     private String namespace;
     private String token;
@@ -39,7 +38,9 @@ public class LmstfyClient {
     private HttpUrl serviceAddress;
 
 
-    public LmstfyClient(String host, int port, String namespace, String token) {
+    public LmstfyClient(String host, int port, String namespace, String token,
+                        int readTimeoutSecond, int writeTimeoutSecond, int connectTimeoutSecond,
+                        int retryTimes, int retryIntervalMilliseconds) {
         this.namespace = namespace;
         this.token = token;
         this.serviceAddress = new HttpUrl.Builder()
@@ -50,14 +51,18 @@ public LmstfyClient(String host, int port, String namespace, String token) {
 
         this.http = new OkHttpClient.Builder()
                 .retryOnConnectionFailure(true)
-                .addInterceptor(new OkHttpRetryInterceptor(3, 100))
-                .readTimeout(600, TimeUnit.SECONDS)
-                .writeTimeout(600, TimeUnit.SECONDS)
-                .connectTimeout(5, TimeUnit.SECONDS)
+                .addInterceptor(new OkHttpRetryInterceptor(retryTimes, retryIntervalMilliseconds))
+                .readTimeout(readTimeoutSecond, TimeUnit.SECONDS)
+                .writeTimeout(writeTimeoutSecond, TimeUnit.SECONDS)
+                .connectTimeout(connectTimeoutSecond, TimeUnit.SECONDS)
                 .connectionPool(new ConnectionPool(100, 5, TimeUnit.MINUTES))
                 .build();
     }
 
+    public LmstfyClient(String host, int port, String namespace, String token) {
+        this(host, port, namespace, token, 600, 600, 5, 3, 100);
+    }
+
     private LmstfyResponse doRequest(String method, HttpUrl url, RequestBody body) throws LmstfyException {
         Request request = new Request.Builder()
                 .url(url)
@@ -114,7 +119,7 @@ public String publish(String queue, byte[] data, int ttlSecond, short tries, int
      *                      a job; if it's positive, this method would polling for new job until timeout.
      * @param queues        You can consume multiple queues of the same namespace at once. The order of the queues in
      *                      the params implies the priority.
-     * @return Return the job, if there is no job available, it will return null.
+     * @return Return the job.
      * @throws LmstfyException
      */
     public Job consume(int ttrSecond, int timeoutSecond, String... queues) throws LmstfyException {
@@ -126,14 +131,12 @@ public Job consume(int ttrSecond, int timeoutSecond, String... queues) throws Lm
         LmstfyResponse response = this.doRequest("GET", url, null);
         switch (response.getCode()) {
             case HTTP_OK:
-                Job job = response.unmarshalBody(Job.class);
-                job.setData(new String(Base64.getDecoder().decode(job.getBase64Data())));
-                return job;
+                return response.unmarshalToJob();
             case HTTP_BAD_REQUEST:
                 ErrorResponse errorResponse = response.unmarshalBody(ErrorResponse.class);
                 throw new LmstfyIllegalRequestException(response.getCode(), errorResponse.getError());
             case HTTP_NOT_FOUND:
-                return null;
+                throw new LmstfyNotJobException();
             default:
                 throw new LmstfyUnexpectedException(response.getCode());
         }
@@ -223,20 +226,83 @@ public Job peekJob(String queue, String jobID) throws LmstfyException {
         return this.peek(url);
     }
 
+    /**
+     * Peek the deadletter of the queue
+     *
+     * @param queue
+     * @return
+     * @throws LmstfyException
+     */
+    public DeadLetterResponse peekDeadLetter(String queue) throws LmstfyException {
+        HttpUrl url = genServiceUrlBuilder(PATH_API, this.namespace, queue, "deadletter")
+                .build();
+
+        LmstfyResponse response = doRequest("GET", url, null);
+        switch (response.getCode()) {
+            case HTTP_OK:
+                DeadLetterResponse deadLetterResponse = response.unmarshalBody(DeadLetterResponse.class);
+                return deadLetterResponse;
+            default:
+                throw new LmstfyUnexpectedException(response.getCode());
+        }
+    }
+
     private Job peek(HttpUrl url) throws LmstfyException {
         LmstfyResponse response = doRequest("GET", url, null);
         switch (response.getCode()) {
             case HTTP_OK:
-                Job job = response.unmarshalBody(Job.class);
-                job.setData(new String(Base64.getDecoder().decode(job.getBase64Data())));
-                return job;
+                return response.unmarshalToJob();
             case HTTP_NOT_FOUND:
                 return null;
             default:
                 throw new LmstfyUnexpectedException(response.getCode());
         }
     }
 
+    /**
+     * Respawn job(s) in the dead letter
+     *
+     * @param queue
+     * @param limit     The number (upper limit) of the jobs to be respawned.
+     * @param ttlSecond Time-to-live of this job in seconds, 0 means forever.
+     * @return The number of jobs that're respawned.
+     * @throws LmstfyException
+     */
+    public int respawnDeadLetter(String queue, int limit, int ttlSecond) throws LmstfyException {
+        HttpUrl url = genServiceUrlBuilder(PATH_API, this.namespace, queue, "deadletter")
+                .addQueryParameter(QUERY_LIMIT, String.valueOf(limit))
+                .addQueryParameter(QUERY_TTL, String.valueOf(ttlSecond))
+                .build();
+        LmstfyResponse response = doRequest("PUT", url, null);
+        switch (response.getCode()) {
+            case HTTP_OK:
+                RespawnResponse respawnResponse = response.unmarshalBody(RespawnResponse.class);
+                return respawnResponse.getCount();
+            default:
+                throw new LmstfyUnexpectedException(response.getCode());
+        }
+    }
+
+    /**
+     * Delete job(s) in the dead letter
+     *
+     * @param queue
+     * @param limit The number (upper limit) of the jobs to be deleted
+     * @throws LmstfyException
+     */
+    public void deleteDeadLetter(String queue, int limit) throws LmstfyException {
+        HttpUrl url = genServiceUrlBuilder(PATH_API, this.namespace, queue, "deadletter")
+                .addQueryParameter(QUERY_LIMIT, String.valueOf(limit))
+                .build();
+        LmstfyResponse response = doRequest("DELETE", url, null);
+        switch (response.getCode()) {
+            case HTTP_NO_CONTENT:
+                return;
+            default:
+                throw new LmstfyUnexpectedException(response.getCode());
+        }
+    }
+
     private HttpUrl.Builder genServiceUrlBuilder(String... pathSegments) {
         HttpUrl.Builder builder = serviceAddress.newBuilder();
         for (String pathSegment : pathSegments) {
diff --git a/src/main/java/com/meitu/platform/lmstfy/example.java b/src/main/java/com/meitu/platform/lmstfy/example.java
@@ -16,26 +16,30 @@ public class example {
     private static String host = "localhost";
     private static int port = 7777;
     private static String namespace = "sdk-test";
+    private static String queue = "sdk-test-queue";
+    private static LmstfyClient client = new LmstfyClient(host, port, namespace, token);
 
     public static void main(String[] args) {
         publish();
         consume();
+        deleteAndAck();
+        queueSize();
+        peekQueue();
+        peekJob();
     }
 
     private static void publish() {
-        LmstfyClient client = new LmstfyClient(host, port, namespace, token);
         try {
-            String jobID = client.publish("sdk-test-queue", "test data".getBytes(), 5, (short) 2, 0);
+            String jobID = client.publish(queue, "test data".getBytes(), 5, (short) 2, 0);
             System.out.println(jobID);
         } catch (LmstfyException e) {
             e.printStackTrace();
         }
     }
 
     private static void consume() {
-        LmstfyClient client = new LmstfyClient(host, port, namespace, token);
         try {
-            Job job = client.consume(3, 2, "sdk-test-queue");
+            Job job = client.consume(3, 2, queue);
             System.out.println(job.getData());
         } catch (LmstfyNotJobException e) {
             System.out.println("There has no available job");
@@ -44,4 +48,59 @@ private static void consume() {
         }
     }
 
+    private static void deleteAndAck() {
+        try {
+            String jobID1 = client.publish(queue, "test data".getBytes(), 5, (short) 2, 0);
+            client.delete(queue, jobID1);
+            String jobID2 = client.publish(queue, "test data".getBytes(), 5, (short) 2, 0);
+            client.ack(queue, jobID2);
+        } catch (LmstfyException e) {
+            System.out.println("Request Lmstfy error, " + e.getMessage());
+        }
+    }
+
+    private static void queueSize() {
+        try {
+            int size1 = client.queueSize(queue);
+            client.publish(queue, "test data".getBytes(), 5, (short) 2, 0);
+            int size2 = client.queueSize(queue);
+            System.out.printf("%d %d\n", size1, size2);
+        } catch (LmstfyException e) {
+            System.out.println("Request Lmstfy error, " + e.getMessage());
+        }
+    }
+
+    private static void peekQueue() {
+        try {
+            Job job = client.peekQueue(queue);
+            if (job == null) {
+                System.out.println("There has no available job to peek");
+                return;
+            }
+            if (job.getData() == null) {
+                System.out.printf("The job %s is expired\n", job.getJobID());
+            } else {
+                System.out.println(job.getData());
+            }
+
+        } catch (LmstfyException e) {
+            System.out.println("Request Lmstfy error, " + e.getMessage());
+        }
+    }
+
+    private static void peekJob() {
+        try {
+            String jobID = client.publish(queue, "test data".getBytes(), 5, (short) 2, 0);
+            Job job = client.peekJob(queue, jobID);
+            if (job == null) {
+                System.out.println("There has no available job");
+            } else {
+                System.out.printf("%s\n", job.getData());
+            }
+        } catch (LmstfyException e) {
+            System.out.println("Request Lmstfy error, " + e.getMessage());
+        }
+    }
+
+
 }
diff --git a/src/main/java/com/meitu/platform/lmstfy/response/DeadLetterResponse.java b/src/main/java/com/meitu/platform/lmstfy/response/DeadLetterResponse.java
@@ -0,0 +1,51 @@
+package com.meitu.platform.lmstfy.response;
+
+import com.google.gson.annotations.SerializedName;
+
+/**
+ * Description:
+ *
+ * @author Yesphet
+ * @date 2019-12-10
+ */
+public class DeadLetterResponse {
+
+    private String namespace;
+    private String queue;
+    @SerializedName("deadletter_size")
+    private int deadLetterSize;
+    @SerializedName("deadletter_head")
+    private String deadLetterHead;
+
+    public String getNamespace() {
+        return namespace;
+    }
+
+    public void setNamespace(String namespace) {
+        this.namespace = namespace;
+    }
+
+    public String getQueue() {
+        return queue;
+    }
+
+    public void setQueue(String queue) {
+        this.queue = queue;
+    }
+
+    public int getDeadLetterSize() {
+        return deadLetterSize;
+    }
+
+    public void setDeadLetterSize(int deadLetterSize) {
+        this.deadLetterSize = deadLetterSize;
+    }
+
+    public String getDeadLetterHead() {
+        return deadLetterHead;
+    }
+
+    public void setDeadLetterHead(String deadLetterHead) {
+        this.deadLetterHead = deadLetterHead;
+    }
+}
diff --git a/src/main/java/com/meitu/platform/lmstfy/response/LmstfyResponse.java b/src/main/java/com/meitu/platform/lmstfy/response/LmstfyResponse.java
@@ -1,9 +1,11 @@
 package com.meitu.platform.lmstfy.response;
 
 import com.google.gson.Gson;
+import com.meitu.platform.lmstfy.Job;
 import okhttp3.Response;
 
 import java.io.IOException;
+import java.util.Base64;
 
 /**
  * Description:
@@ -46,4 +48,12 @@ public String getRequestID() {
     public <T> T unmarshalBody(Class<T> tClass) {
         return gson.fromJson(this.body, tClass);
     }
+
+    public Job unmarshalToJob() {
+        Job job = gson.fromJson(this.body, Job.class);
+        if (job.getBase64Data() != null) {
+            job.setData(new String(Base64.getDecoder().decode(job.getBase64Data())));
+        }
+        return job;
+    }
 }
diff --git a/src/main/java/com/meitu/platform/lmstfy/response/RespawnResponse.java b/src/main/java/com/meitu/platform/lmstfy/response/RespawnResponse.java
@@ -0,0 +1,29 @@
+package com.meitu.platform.lmstfy.response;
+
+/**
+ * Description:
+ *
+ * @author Yesphet
+ * @date 2019-12-10
+ */
+public class RespawnResponse {
+
+    private String msg;
+    private int count;
+
+    public String getMsg() {
+        return msg;
+    }
+
+    public void setMsg(String msg) {
+        this.msg = msg;
+    }
+
+    public int getCount() {
+        return count;
+    }
+
+    public void setCount(int count) {
+        this.count = count;
+    }
+}
