more apis to enqueue messages.

Author: sonus21

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 **NOTE**: The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.1.1] - 24-Sep-2020
+### Added
+* More apis to enqueue unique message
+
 ## [2.1.0] - 16-Sep-2020
 ### Added
 * Allow application to provide message id while enqueuing messages
@@ -102,5 +106,6 @@ Fixes:
 [2.0.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue/2.0.0-RELEASE
 [2.0.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue/2.0.1-RELEASE
 [2.0.2]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue/2.0.2-RELEASE
-[2.0.4]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue/2.0.4-RELEASE
-[2.1.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue/2.1.0-RELEASE
+[2.0.4]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.0.4-RELEASE
+[2.1.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.1.0-RELEASE
+[2.1.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.1.1-RELEASE

build.gradle

@@ -67,7 +67,7 @@ ext {
 
 subprojects {
     group = 'com.github.sonus21'
-    version = '2.1.0-RELEASE'
+    version = '2.1.1-RELEASE'
 
     dependencies {
         // https://mvnrepository.com/artifact/org.springframework/spring-messaging

rqueue-common-test/src/main/java/com/github/sonus21/rqueue/test/application/ApplicationBasicConfiguration.java

@@ -41,15 +41,20 @@ public abstract class ApplicationBasicConfiguration {
   private static final Logger monitorLogger = LoggerFactory.getLogger("monitor");
   protected RedisServer redisServer;
   protected ExecutorService executorService;
-  protected List<RProcess> processes;
+  protected List<MonitorProcess> processes;
+
   @Value("${mysql.db.name}")
   protected String dbName;
+
   @Value("${spring.redis.port}")
   protected int redisPort;
+
   @Value("${spring.redis.host}")
   protected String redisHost;
+
   @Value("${use.system.redis:false}")
   protected boolean useSystemRedis;
+
   @Value("${monitor.thread.count:0}")
   protected int monitorThreads;
 
@@ -73,10 +78,10 @@ protected void destroy() {
     }
 
     if (processes != null) {
-      for (RProcess rProcess : processes) {
-        rProcess.process.destroy();
-        monitorLogger.info("RedisNode {} ", rProcess.redisNode);
-        for (String line : rProcess.out) {
+      for (MonitorProcess monitorProcess : processes) {
+        monitorProcess.process.destroy();
+        monitorLogger.info("RedisNode {} ", monitorProcess.redisNode);
+        for (String line : monitorProcess.out) {
           monitorLogger.info("{}", line);
         }
       }
@@ -94,8 +99,9 @@ protected void monitor(String host, int port) {
                 Runtime.getRuntime()
                     .exec("redis-cli " + " -h " + host + " -p " + port + " monitor");
             List<String> lines = new LinkedList<>();
-            RProcess rProcess = new RProcess(process, new RedisNode(host, port), lines);
-            processes.add(rProcess);
+            MonitorProcess monitorProcess =
+                new MonitorProcess(process, new RedisNode(host, port), lines);
+            processes.add(monitorProcess);
             BufferedReader br = new BufferedReader(new InputStreamReader(process.getInputStream()));
             String s;
             while ((s = br.readLine()) != null) {
@@ -131,7 +137,7 @@ public ObjectMapper objectMapper() {
   }
 
   @AllArgsConstructor
-  public static class RProcess {
+  public static class MonitorProcess {
     Process process;
     RedisNode redisNode;
     List<String> out;

rqueue-common-test/src/main/java/com/github/sonus21/rqueue/test/application/MultiRedisSprigBaseApplication.java

@@ -54,8 +54,8 @@ public void preDestroy() {
     if (redisServer2 != null) {
       redisServer2.stop();
     }
-    for (RProcess rProcess : processes) {
-      for (String line : rProcess.out) {
+    for (MonitorProcess monitorProcess : processes) {
+      for (String line : monitorProcess.out) {
         assert line.equals("OK");
       }
     }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueMessageEnqueuer.java

@@ -29,15 +29,19 @@
  * com.github.sonus21.rqueue.exception.QueueDoesNotExist}. In such case register your queue using
  * {@link RqueueEndpointManager#registerQueue(String, String...)} method.
  *
- * <p>There are three types of interfaces in this 1. enqueueXYZ 2. enqueueInXYZ 3. enqueueAtXYZ
+ * <p>There are four types of interfaces in this
  *
- * <p>Messages enqueue using enqueueXYZ shall be consumed as soon as possible
+ * <p>enqueueXYZ : Messages enqueue using this method shall be consumed as soon as possible
  *
- * <p>Messages enqueue using enqueueInXYZ shall be consumed once the given time is elapsed, like in
- * 30 seconds.
+ * <p>enqueueInXYZ: Messages enqueue using enqueueInXYZ shall be consumed once the given time is
+ * elapsed, like in 30 seconds.
  *
- * <p>Messages send using enqueueAtXYZ shall be consumed as soon as the given time is reached for
- * example 3PM tomorrow.
+ * <p>enqueueAtXYZ: Messages send using enqueueAtXYZ shall be consumed as soon as the given time is
+ * reached for example 3PM tomorrow.
+ *
+ * <p>enqueueUniqueXYZ: This method enqueue unique messages on a queue. New messages overwrite the
+ * existing message, the overwriting only works till the point message is not consumed, once message
+ * is consumed it's of no use.
  *
  * @author Sonu Kumar
  */
@@ -61,6 +65,18 @@ public interface RqueueMessageEnqueuer {
    */
   boolean enqueue(String queueName, String messageId, Object message);
 
+  /**
+   * Enqueue unique message on a given queue without any delay, consume as soon as possible.
+   *
+   * @param queueName on which queue message has to be send
+   * @param messageId the message id for uniqueness
+   * @param message message object it could be any arbitrary object.
+   * @return message id on successful enqueue otherwise null.
+   */
+  default boolean enqueueUnique(String queueName, String messageId, Object message) {
+    return enqueue(queueName, messageId, message);
+  }
+
   /**
    * Enqueue a message on the given queue with the given retry count. This message would not be
    * consumed more than the specified time due to failure in underlying systems.
@@ -107,18 +123,6 @@ public interface RqueueMessageEnqueuer {
    */
   boolean enqueueWithPriority(String queueName, String priority, String messageId, Object message);
 
-  /**
-   * Enqueue unique message on a given queue without any delay, consume as soon as possible.
-   *
-   * @param queueName on which queue message has to be send
-   * @param messageId the message id for uniqueness
-   * @param message message object it could be any arbitrary object.
-   * @return message id on successful enqueue otherwise null.
-   */
-  default boolean enqueueUnique(String queueName, String messageId, Object message) {
-    return enqueue(queueName, messageId, message);
-  }
-
   /**
    * Enqueue unique message on given queue, that will be consumed as soon as possible.
    *
@@ -213,6 +217,20 @@ default boolean enqueueIn(
     return enqueueIn(queueName, messageId, message, unit.toMillis(delay));
   }
 
+  /**
+   * Enqueue a message on given queue with delay, consume as soon as the delayed is expired.
+   *
+   * @param queueName on which queue message has to be send
+   * @param messageId the message id for uniqueness
+   * @param message message object it could be any arbitrary object.
+   * @param delayInMillisecond total execution delay
+   * @return message id on successful enqueue otherwise {@literal null}.
+   */
+  default boolean enqueueUniqueIn(
+      String queueName, String messageId, Object message, long delayInMillisecond) {
+    return enqueueIn(queueName, messageId, message, delayInMillisecond);
+  }
+
   /**
    * Enqueue a task that would be scheduled to run in the specified milli seconds.
    *
@@ -339,17 +357,25 @@ default boolean enqueueInWithPriority(
   }
 
   /**
-   * Enqueue a message on given queue with delay, consume as soon as the delayed is expired.
+   * Schedule unique message on the given queue at the provided time. It will be executed as soon as
+   * the given delay is elapse.
    *
    * @param queueName on which queue message has to be send
-   * @param messageId the message id for uniqueness
+   * @param priority the name of the priority level
+   * @param messageId the message id
    * @param message message object it could be any arbitrary object.
-   * @param delayInMillisecond total execution delay
-   * @return message id on successful enqueue otherwise {@literal null}.
+   * @param delay time to wait before it can be consumed.
+   * @param unit unit of the delay
+   * @return message was enqueue successfully or failed.
    */
-  default boolean enqueueUniqueIn(
-      String queueName, String messageId, Object message, long delayInMillisecond) {
-    return enqueueIn(queueName, messageId, message, delayInMillisecond);
+  default boolean enqueueUniqueInWithPriority(
+      String queueName,
+      String priority,
+      String messageId,
+      Object message,
+      long delay,
+      TimeUnit unit) {
+    return enqueueInWithPriority(queueName, priority, messageId, message, unit.toMillis(delay));
   }
 
   /**
@@ -435,6 +461,22 @@ default boolean enqueueAt(String queueName, String messageId, Object message, Da
     return enqueueAt(queueName, messageId, message, starTime.toInstant());
   }
 
+  /**
+   * Schedule unique messages on the given queue at the provided time. It will be available to
+   * consume as soon as the given time is reached.
+   *
+   * @param queueName on which queue message has to be send
+   * @param message message object it could be any arbitrary object.
+   * @param messageId a unique identifier message id for this message
+   * @param timeInMilliSeconds time at which this message has to be consumed.
+   * @return message was enqueue successfully or failed.
+   */
+  default boolean enqueueUniqueAt(
+      String queueName, String messageId, Object message, long timeInMilliSeconds) {
+    return enqueueUniqueIn(
+        queueName, messageId, message, timeInMilliSeconds - System.currentTimeMillis());
+  }
+
   /**
    * Schedule a message on the given queue at the provided time. It will be executed as soon as the
    * given time is reached, time must be in the future.
@@ -543,14 +585,19 @@ default boolean enqueueAtWithPriority(
    * consume as soon as the given time is reached.
    *
    * @param queueName on which queue message has to be send
+   * @param priority priority of the given message
    * @param message message object it could be any arbitrary object.
    * @param messageId a unique identifier message id for this message
    * @param timeInMilliSeconds time at which this message has to be consumed.
    * @return message was enqueue successfully or failed.
    */
-  default boolean enqueueUniqueAt(
-      String queueName, String messageId, Object message, long timeInMilliSeconds) {
-    return enqueueUniqueIn(
-        queueName, messageId, message, timeInMilliSeconds - System.currentTimeMillis());
+  default boolean enqueueUniqueAtWithPriority(
+      String queueName,
+      String priority,
+      String messageId,
+      Object message,
+      long timeInMilliSeconds) {
+    return enqueueAtWithPriority(
+        queueName, priority, messageId, message, timeInMilliSeconds - System.currentTimeMillis());
   }
 }