check duplicate message before enqueue (#260)

* check duplicate message before enqueue

* add duplicate check in unique in as well

* add copyright

* update changelog file

* version update

* doc update

* reuse the method

* fix: duplicate message test

* wip

* throw error when duplicate message is saved

* test compilation issue

* fix: misuse of mockito argument checker

* increase timeout

* test: revert test properties

* fix: the enque in should not use push periodic message

* delete test data

* fix: do not use system redis

* do not retry test

* fix: reactive message enqueue

* chore: change BiFunction to Function

* chore: update changelog

* chore: update changelog

Author: sonus21

README.md

@@ -62,9 +62,7 @@ well, where all services code is in Spring.
 
 ### Dependency
 
-Snapshot
-Version: [https://s01.oss.sonatype.org/content/repositories/snapshots/com/github/sonus21/](https://s01.oss.sonatype.org/content/repositories/snapshots/com/github/sonus21/)
-<br/>Release Version: [Maven central](https://search.maven.org/search?q=g:com.github.sonus21)
+Release Version: [Maven central](https://search.maven.org/search?q=g:com.github.sonus21)
 
 #### Spring Boot
 

build.gradle

@@ -69,7 +69,7 @@ ext {
 
 subprojects {
     group = "com.github.sonus21"
-    version = "3.3.0-RELEASE"
+    version = "3.4.0-RELEASE"
 
     dependencies {
         // https://mvnrepository.com/artifact/org.springframework/spring-messaging

docs/CHANGELOG.md

@@ -8,6 +8,16 @@ layout: default
 
 All notable user-facing changes to this project are documented in this file.
 
+## Release [3.4.0] 22-July-2025
+### Fixes
+* Fixed unique enqueue message to reject the message upfront instead of identifying it later #259 
+
+
+## Release [3.3.0] 29-June-2025
+### Fixes
+# Custom Message Converter was ignored #256
+# LockKey prefix was not used #239
+
 ## Release [3.2.0] 10-July-2024
 ### Fixes
 * Fixed typo #218

publish.sh

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 #
-# Copyright (c) 2022-2023 Sonu Kumar
+# Copyright (c) 2022-2025 Sonu Kumar
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # You may not use this file except in compliance with the License.
@@ -15,6 +15,6 @@
 #
 #
 
-./gradlew rqueue-core:publish
-./gradlew rqueue-spring:publish
-./gradlew rqueue-spring-boot-starter:publish
+./gradlew rqueue-core:publishToMavenCentral
+./gradlew rqueue-spring:publishToMavenCentral
+./gradlew rqueue-spring-boot-starter:publishToMavenCentral

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2023 Sonu Kumar
+ * Copyright (c) 2020-2025 Sonu Kumar
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * You may not use this file except in compliance with the License.
@@ -29,6 +29,7 @@
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
 import com.github.sonus21.rqueue.core.impl.MessageSweeper.MessageDeleteRequest;
 import com.github.sonus21.rqueue.dao.RqueueStringDao;
+import com.github.sonus21.rqueue.exception.DuplicateMessageException;
 import com.github.sonus21.rqueue.listener.QueueDetail;
 import com.github.sonus21.rqueue.models.db.MessageMetadata;
 import com.github.sonus21.rqueue.models.enums.MessageStatus;
@@ -50,12 +51,9 @@ abstract class BaseMessageSender {
   protected final MessageHeaders messageHeaders;
   protected final MessageConverter messageConverter;
   protected final RqueueMessageTemplate messageTemplate;
-  @Autowired
-  protected RqueueStringDao rqueueStringDao;
-  @Autowired
-  protected RqueueConfig rqueueConfig;
-  @Autowired
-  protected RqueueMessageMetadataService rqueueMessageMetadataService;
+  @Autowired protected RqueueStringDao rqueueStringDao;
+  @Autowired protected RqueueConfig rqueueConfig;
+  @Autowired protected RqueueMessageMetadataService rqueueMessageMetadataService;
 
   BaseMessageSender(
       RqueueMessageTemplate messageTemplate,
@@ -69,13 +67,13 @@ abstract class BaseMessageSender {
   }
 
   protected Object storeMessageMetadata(
-      RqueueMessage rqueueMessage, Long delayInMillis, boolean reactive) {
+      RqueueMessage rqueueMessage, Long delayInMillis, boolean reactive, boolean isUnique) {
     MessageMetadata messageMetadata = new MessageMetadata(rqueueMessage, MessageStatus.ENQUEUED);
     Duration duration = rqueueConfig.getMessageDurability(delayInMillis);
     if (reactive) {
-      return rqueueMessageMetadataService.saveReactive(messageMetadata, duration);
+      return rqueueMessageMetadataService.saveReactive(messageMetadata, duration, isUnique);
     } else {
-      rqueueMessageMetadataService.save(messageMetadata, duration);
+      rqueueMessageMetadataService.save(messageMetadata, duration, isUnique);
     }
     return null;
   }
@@ -112,7 +110,8 @@ protected String pushMessage(
       String messageId,
       Object message,
       Integer retryCount,
-      Long delayInMilliSecs) {
+      Long delayInMilliSecs,
+      boolean isUnique) {
     QueueDetail queueDetail = EndpointRegistry.get(queueName);
     RqueueMessage rqueueMessage =
         buildMessage(
@@ -124,17 +123,26 @@ protected String pushMessage(
             delayInMilliSecs,
             messageHeaders);
     try {
+      storeMessageMetadata(rqueueMessage, delayInMilliSecs, false, isUnique);
       enqueue(queueDetail, rqueueMessage, delayInMilliSecs, false);
-      storeMessageMetadata(rqueueMessage, delayInMilliSecs, false);
+    } catch (DuplicateMessageException e) {
+      log.warn(
+          "Duplicate message enqueue attempted queue: {}, messageId: {}",
+          queueName,
+          rqueueMessage.getId());
+      return null;
     } catch (Exception e) {
-      log.error("Queue: {} Message {} could not be pushed {}", queueName, rqueueMessage, e);
+      log.error("Queue: {} Message {} could not be pushed", queueName, rqueueMessage.getId(), e);
       return null;
     }
     return rqueueMessage.getId();
   }
 
   protected String pushPeriodicMessage(
-      String queueName, String messageId, Object message, long periodInMilliSeconds) {
+      String queueName,
+      String messageId,
+      Object message,
+      long periodInMilliSeconds) {
     QueueDetail queueDetail = EndpointRegistry.get(queueName);
     RqueueMessage rqueueMessage =
         buildPeriodicMessage(
@@ -146,13 +154,13 @@ protected String pushPeriodicMessage(
             periodInMilliSeconds,
             messageHeaders);
     try {
+      storeMessageMetadata(rqueueMessage, periodInMilliSeconds, false, false);
       enqueue(queueDetail, rqueueMessage, periodInMilliSeconds, false);
-      storeMessageMetadata(rqueueMessage, periodInMilliSeconds, false);
+      return rqueueMessage.getId();
     } catch (Exception e) {
-      log.error("Queue: {} Message {} could not be pushed {}", queueName, rqueueMessage, e);
+      log.error("Queue: {} Message {} could not be pushed", queueName, rqueueMessage, e);
       return null;
     }
-    return rqueueMessage.getId();
   }
 
   protected Object deleteAllMessages(QueueDetail queueDetail) {