diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/Job.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/Job.java index 07899d55..840d10c0 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/Job.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/Job.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2021-2023 Sonu Kumar * - * Licensed 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 + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -27,10 +27,10 @@ /** * On each execution Rqueue creates a job to track it's status and execution progress. * - *

A job belongs to a single message poll, each message listener call creates an execution {@link - * com.github.sonus21.rqueue.models.db.Execution} and that has a detail for specific execution. - * Overall job status can be found using this job interface. This object is available via {@link - * org.springframework.messaging.handler.annotation.Header} in listener method. + *

A job belongs to a single message poll, each message listener call creates an execution + * {@link com.github.sonus21.rqueue.models.db.Execution} and that has a detail for specific + * execution. Overall job status can be found using this job interface. This object is available via + * {@link org.springframework.messaging.handler.annotation.Header} in listener method. */ public interface Job { @@ -81,7 +81,7 @@ public interface Job { * return zero value * * @return remaining duration that this job can take, otherwise other listener will consume this - * message + * message */ Duration getVisibilityTimeout(); @@ -181,8 +181,8 @@ public interface Job { * Release this job back to the queue, the released job would be available for re-execution after * the duration time. * - * @param status job status - * @param why why do want to release this job + * @param status job status + * @param why why do want to release this job * @param duration any positive duration */ void release(JobStatus status, Serializable why, Duration duration); @@ -191,7 +191,7 @@ public interface Job { * Release this job back to queue, this job available for execution after one second. * * @param status what should be the job status - * @param why why do you want to delete this job + * @param why why do you want to delete this job */ void release(JobStatus status, Serializable why); @@ -199,7 +199,7 @@ public interface Job { * Delete this job * * @param status what should be the job status - * @param why why do you want to delete this job + * @param why why do you want to delete this job */ void delete(JobStatus status, Serializable why); diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueMessageEnqueuer.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueMessageEnqueuer.java index d412fafb..a116048a 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueMessageEnqueuer.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueMessageEnqueuer.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2020-2023 Sonu Kumar * - * Licensed 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 + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -22,13 +22,14 @@ import java.time.Instant; import java.util.Date; import java.util.concurrent.TimeUnit; +import org.springframework.messaging.converter.MessageConverter; /** * RqueueMessageEnqueuer enqueue message to Redis queue using different mechanism. Use any of the * methods from this interface to enqueue message to any queue. Queue must exist, if a queue does - * not exist then it will throw an error of the {@link - * com.github.sonus21.rqueue.exception.QueueDoesNotExist}. In such case register your queue using - * {@link RqueueEndpointManager#registerQueue(String, String...)} method. + * not exist then it will throw an error of the + * {@link com.github.sonus21.rqueue.exception.QueueDoesNotExist}. In such case register your queue + * using {@link RqueueEndpointManager#registerQueue(String, String...)} method. * *

There are four types of interfaces in this * @@ -52,7 +53,7 @@ public interface RqueueMessageEnqueuer { * Enqueue a message on given queue without any delay, consume as soon as possible. * * @param queueName on which queue message has to be send - * @param message message object it could be any arbitrary object. + * @param message message object it could be any arbitrary object. * @return message id on successful enqueue otherwise null. */ String enqueue(String queueName, Object message); @@ -62,7 +63,7 @@ public interface RqueueMessageEnqueuer { * * @param queueName on which queue message has to be send * @param messageId message id - * @param message message object it could be any arbitrary object. + * @param message message object it could be any arbitrary object. * @return message was enqueue successfully or failed. */ boolean enqueue(String queueName, String messageId, Object message); @@ -72,7 +73,7 @@ public interface RqueueMessageEnqueuer { * * @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 message message object it could be any arbitrary object. * @return message id on successful enqueue otherwise null. */ boolean enqueueUnique(String queueName, String messageId, Object message); @@ -81,10 +82,10 @@ public interface RqueueMessageEnqueuer { * 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. * - * @param queueName on which queue message has to be send - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param message message object it could be any arbitrary object. * @param retryCount how many times a message would be retried, before it can be discarded or send - * to dead letter queue configured using {@link RqueueListener#numRetries()} + * to dead letter queue configured using {@link RqueueListener#numRetries()} * @return message id on successful enqueue otherwise null. */ String enqueueWithRetry(String queueName, Object message, int retryCount); @@ -93,11 +94,11 @@ public interface RqueueMessageEnqueuer { * 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. * - * @param queueName on which queue message has to be send - * @param messageId message id for this message. - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param messageId message id for this message. + * @param message message object it could be any arbitrary object. * @param retryCount how many times a message would be retried, before it can be discarded or send - * to dead letter queue configured using {@link RqueueListener#numRetries()} + * to dead letter queue configured using {@link RqueueListener#numRetries()} * @return message was enqueue successfully or failed. */ boolean enqueueWithRetry(String queueName, String messageId, Object message, int retryCount); @@ -106,8 +107,8 @@ public interface RqueueMessageEnqueuer { * Enqueue a message on given queue, that will be consumed as soon as possible. * * @param queueName on which queue message has to be send - * @param priority the priority for this message, like high, low, medium etc - * @param message message object it could be any arbitrary object. + * @param priority the priority for this message, like high, low, medium etc + * @param message message object it could be any arbitrary object. * @return message id on successful enqueue otherwise null. */ String enqueueWithPriority(String queueName, String priority, Object message); @@ -116,9 +117,9 @@ public interface RqueueMessageEnqueuer { * Enqueue a message on given queue, that will be consumed as soon as possible. * * @param queueName on which queue message has to be send - * @param priority the priority for this message, like high, low, medium etc + * @param priority the priority for this message, like high, low, medium etc * @param messageId the message id for this message - * @param message message object it could be any arbitrary object. + * @param message message object it could be any arbitrary object. * @return message was enqueued successfully or not. */ boolean enqueueWithPriority(String queueName, String priority, String messageId, Object message); @@ -127,9 +128,9 @@ public interface RqueueMessageEnqueuer { * Enqueue unique message on given queue, that will be consumed as soon as possible. * * @param queueName on which queue message has to be send - * @param priority the priority for this message, like high, low, medium etc + * @param priority the priority for this message, like high, low, medium etc * @param messageId the message id for this message - * @param message message object it could be any arbitrary object. + * @param message message object it could be any arbitrary object. * @return message was enqueue successfully or failed. */ default boolean enqueueUniqueWithPriority( @@ -142,8 +143,8 @@ default boolean enqueueUniqueWithPriority( * Schedule a message on the given queue with the provided delay. It will be available to consume * as soon as the delay elapse, for example process in 10 seconds * - * @param queueName on which queue message has to be send - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param message message object it could be any arbitrary object. * @param delayInMilliSecs delay in milliseconds * @return message id on successful enqueue otherwise null. */ @@ -153,9 +154,9 @@ default boolean enqueueUniqueWithPriority( * Schedule a message on the given queue with the provided delay. It will be available to consume * as soon as the delay elapse. * - * @param queueName on which queue message has to be send - * @param messageId the message id, using which this message will be identified - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param messageId the message id, using which this message will be identified + * @param message message object it could be any arbitrary object. * @param delayInMilliSecs delay in milliseconds * @return message was enqueue successfully or failed. */ @@ -166,8 +167,8 @@ default boolean enqueueUniqueWithPriority( * as soon as the delay elapse. * * @param queueName on which queue message has to be send - * @param message message object it could be any arbitrary object. - * @param delay time to wait before it can be executed. + * @param message message object it could be any arbitrary object. + * @param delay time to wait before it can be executed. * @return message id on successful enqueue otherwise null. */ default String enqueueIn(String queueName, Object message, Duration delay) { @@ -180,8 +181,8 @@ default String enqueueIn(String queueName, Object message, Duration delay) { * * @param queueName on which queue message has to be send * @param messageId the message id, using which this message will be identified - * @param message message object it could be any arbitrary object. - * @param delay time to wait before it can be executed. + * @param message message object it could be any arbitrary object. + * @param delay time to wait before it can be executed. * @return success or failure. */ default boolean enqueueIn(String queueName, String messageId, Object message, Duration delay) { @@ -193,9 +194,9 @@ default boolean enqueueIn(String queueName, String messageId, Object message, Du * as soon as the specified delay elapse. * * @param queueName on which queue message has to be send - * @param message message object it could be any arbitrary object. - * @param delay time to wait before it can be executed. - * @param unit unit of the delay + * @param message message object it could be any arbitrary object. + * @param delay time to wait before it can be executed. + * @param unit unit of the delay * @return message id on successful enqueue otherwise null. */ default String enqueueIn(String queueName, Object message, long delay, TimeUnit unit) { @@ -208,9 +209,9 @@ default String enqueueIn(String queueName, Object message, long delay, TimeUnit * * @param queueName on which queue message has to be send * @param messageId message id using which this message can be identified - * @param message message object it could be any arbitrary object. - * @param delay time to wait before it can be executed. - * @param unit unit of the delay + * @param message message object it could be any arbitrary object. + * @param delay time to wait before it can be executed. + * @param unit unit of the delay * @return success or failure. */ default boolean enqueueIn( @@ -221,9 +222,9 @@ default boolean enqueueIn( /** * Enqueue a message on given queue with delay, consume as soon as the scheduled 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 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}. */ @@ -233,12 +234,13 @@ boolean enqueueUniqueIn( /** * Enqueue a task that would be scheduled to run in the specified milliseconds. * - * @param queueName on which queue message has to be sent - * @param message message object it could be any arbitrary object. - * @param retryCount how many times a message would be retried, before it can be discarded or sent - * to dead letter queue configured using {@link RqueueListener#numRetries()} ()} + * @param queueName on which queue message has to be sent + * @param message message object it could be any arbitrary object. + * @param retryCount how many times a message would be retried, before it can be discarded + * or sent to dead letter queue configured using + * {@link RqueueListener#numRetries()} ()} * @param delayInMilliSecs delay in milliseconds, this message would be only visible to the - * listener when number of millisecond has elapsed. + * listener when number of millisecond has elapsed. * @return message id on successful enqueue otherwise {@literal null} */ String enqueueInWithRetry( @@ -247,13 +249,14 @@ String enqueueInWithRetry( /** * Enqueue a task that would be scheduled to run in the specified milliseconds. * - * @param queueName on which queue message has to be sent - * @param messageId the message identifier - * @param message message object it could be any arbitrary object. - * @param retryCount how many times a message would be retried, before it can be discarded or sent - * to dead letter queue configured using {@link RqueueListener#numRetries()} ()} + * @param queueName on which queue message has to be sent + * @param messageId the message identifier + * @param message message object it could be any arbitrary object. + * @param retryCount how many times a message would be retried, before it can be discarded + * or sent to dead letter queue configured using + * {@link RqueueListener#numRetries()} ()} * @param delayInMilliSecs delay in milliseconds, this message would be only visible to the - * listener when number of millisecond has elapsed. + * listener when number of millisecond has elapsed. * @return message was enqueue successfully or failed. */ boolean enqueueInWithRetry( @@ -263,9 +266,9 @@ boolean enqueueInWithRetry( * Schedule a 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 priority the name of the priority level - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param priority the name of the priority level + * @param message message object it could be any arbitrary object. * @param delayInMilliSecs delay in milliseconds * @return message id on successful enqueue otherwise {@literal null}. */ @@ -279,10 +282,10 @@ default String enqueueInWithPriority( * Schedule a 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 priority the name of the priority level - * @param messageId the message id - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param priority the name of the priority level + * @param messageId the message id + * @param message message object it could be any arbitrary object. * @param delayInMilliSecs delay in milliseconds * @return message was enqueue successfully or failed. */ @@ -300,9 +303,9 @@ default boolean enqueueInWithPriority( * given delay is elapse. * * @param queueName on which queue message has to be send - * @param priority the name of the priority level - * @param message message object it could be any arbitrary object. - * @param delay time to wait before it can be consumed. + * @param priority the name of the priority level + * @param message message object it could be any arbitrary object. + * @param delay time to wait before it can be consumed. * @return message id on successful enqueue otherwise {@literal null}. */ default String enqueueInWithPriority( @@ -315,10 +318,10 @@ default String enqueueInWithPriority( * given delay is elapse. * * @param queueName on which queue message has to be send - * @param priority the name of the priority level + * @param priority the name of the priority level * @param messageId the message id - * @param message message object it could be any arbitrary object. - * @param delay time to wait before it can be consumed. + * @param message message object it could be any arbitrary object. + * @param delay time to wait before it can be consumed. * @return message was enqueue successfully or failed. */ default boolean enqueueInWithPriority( @@ -331,10 +334,10 @@ default boolean enqueueInWithPriority( * given delay is elapse. * * @param queueName on which queue message has to be send - * @param priority the name of the priority level - * @param message message object it could be any arbitrary object. - * @param delay time to wait before it can be consumed. - * @param unit unit of the delay + * @param priority the name of the priority level + * @param message message object it could be any arbitrary object. + * @param delay time to wait before it can be consumed. + * @param unit unit of the delay * @return message id on a successful enqueue otherwise {@literal null}. */ default String enqueueInWithPriority( @@ -347,11 +350,11 @@ default String enqueueInWithPriority( * given delay is elapse. * * @param queueName on which queue message has to be send - * @param priority the name of the priority level + * @param priority the name of the priority level * @param messageId the message id - * @param message message object it could be any arbitrary object. - * @param delay time to wait before it can be consumed. - * @param unit unit of the delay + * @param message message object it could be any arbitrary object. + * @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 enqueueInWithPriority( @@ -369,11 +372,11 @@ default boolean enqueueInWithPriority( * the given delay is elapse. * * @param queueName on which queue message has to be send - * @param priority the name of the priority level + * @param priority the name of the priority level * @param messageId the message id - * @param message message object it could be any arbitrary object. - * @param delay time to wait before it can be consumed. - * @param unit unit of the delay + * @param message message object it could be any arbitrary object. + * @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 enqueueUniqueInWithPriority( @@ -394,8 +397,8 @@ default boolean enqueueUniqueInWithPriority( * Schedule a message 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 queueName on which queue message has to be send + * @param message message object it could be any arbitrary object. * @param startTimeInMilliSeconds time at which this message has to be consumed. * @return message id on successful enqueue otherwise {@literal null}. */ @@ -407,9 +410,9 @@ default String enqueueAt(String queueName, Object message, long startTimeInMilli * Schedule a message 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 messageId message id - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param messageId message id + * @param message message object it could be any arbitrary object. * @param startTimeInMilliSeconds time at which this message has to be consumed. * @return message was enqueued successfully or failed. */ @@ -424,8 +427,8 @@ default boolean enqueueAt( * 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 starTime time at which this message has to be consumed. + * @param message message object it could be any arbitrary object. + * @param starTime time at which this message has to be consumed. * @return message id on successful enqueue otherwise {@literal null}. */ default String enqueueAt(String queueName, Object message, Instant starTime) { @@ -438,8 +441,8 @@ default String enqueueAt(String queueName, Object message, Instant starTime) { * * @param queueName on which queue message has to be send * @param messageId the message id - * @param message message object it could be any arbitrary object. - * @param starTime time at which this message has to be consumed. + * @param message message object it could be any arbitrary object. + * @param starTime time at which this message has to be consumed. * @return message was enqueued successfully or failed. */ default boolean enqueueAt(String queueName, String messageId, Object message, Instant starTime) { @@ -451,8 +454,8 @@ default boolean enqueueAt(String queueName, String messageId, Object message, In * 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 starTime time at which this message has to be consumed. + * @param message message object it could be any arbitrary object. + * @param starTime time at which this message has to be consumed. * @return message id on successful enqueue otherwise {@literal null}. */ default String enqueueAt(String queueName, Object message, Date starTime) { @@ -465,8 +468,8 @@ default String enqueueAt(String queueName, Object message, Date starTime) { * * @param queueName on which queue message has to be send * @param messageId the message id - * @param message message object it could be any arbitrary object. - * @param starTime time at which this message has to be consumed. + * @param message message object it could be any arbitrary object. + * @param starTime time at which this message has to be consumed. * @return message was enqueued successfully or failed. */ default boolean enqueueAt(String queueName, String messageId, Object message, Date starTime) { @@ -477,9 +480,9 @@ default boolean enqueueAt(String queueName, String messageId, Object message, Da * 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 for this message + * @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 for this message * @param timeInMilliSeconds time at which this message has to be consumed. * @return message was enqueue successfully or failed. */ @@ -493,9 +496,9 @@ default boolean enqueueUniqueAt( * 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. * - * @param queueName on which queue message has to be send - * @param priority the name of the priority level - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param priority the name of the priority level + * @param message message object it could be any arbitrary object. * @param startTimeInMilliSecond time at which the message would be consumed. * @return message id on successful enqueue otherwise {@literal null}. */ @@ -509,10 +512,10 @@ default String enqueueAtWithPriority( * 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. * - * @param queueName on which queue message has to be send - * @param priority the name of the priority level - * @param messageId the message id - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param priority the name of the priority level + * @param messageId the message id + * @param message message object it could be any arbitrary object. * @param startTimeInMilliSecond time at which the message would be consumed. * @return message was enqueue successfully or failed. */ @@ -535,8 +538,8 @@ default boolean enqueueAtWithPriority( * given time is reached, time must be in the future. * * @param queueName on which queue message has to be send - * @param priority the name of the priority level - * @param message message object it could be any arbitrary object. + * @param priority the name of the priority level + * @param message message object it could be any arbitrary object. * @param startTime time at which message is supposed to consume * @return message id on successful enqueue otherwise {@literal null} */ @@ -550,10 +553,10 @@ default String enqueueAtWithPriority( * given time is reached, time must be in the future. * * @param queueName on which queue message has to be send - * @param priority the name of the priority level - * @param message message object it could be any arbitrary object. + * @param priority the name of the priority level + * @param message message object it could be any arbitrary object. * @param messageId the message id - * @param instant time at which message is supposed to consume + * @param instant time at which message is supposed to consume * @return message was enqueue successfully or failed. */ default boolean enqueueAtWithPriority( @@ -566,9 +569,9 @@ default boolean enqueueAtWithPriority( * given time is reached, time must be in the future. * * @param queueName on which queue message has to be send - * @param priority the name of the priority level - * @param message message object it could be any arbitrary object. - * @param time time at which message would be consumed. + * @param priority the name of the priority level + * @param message message object it could be any arbitrary object. + * @param time time at which message would be consumed. * @return a message id on successful enqueue otherwise {@literal null} */ default String enqueueAtWithPriority( @@ -581,10 +584,10 @@ default String enqueueAtWithPriority( * given time is reached, time must be in the future. * * @param queueName on which queue message has to be send - * @param priority the name of the priority level + * @param priority the name of the priority level * @param messageId the message id - * @param message message object it could be any arbitrary object. - * @param time time at which message would be consumed. + * @param message message object it could be any arbitrary object. + * @param time time at which message would be consumed. * @return message was enqueue successfully or failed. */ default boolean enqueueAtWithPriority( @@ -596,10 +599,10 @@ default boolean enqueueAtWithPriority( * 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 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 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. */ @@ -622,8 +625,8 @@ default boolean enqueueUniqueAtWithPriority( * Enqueue a message on given queue that will be running after a given period. It works like * periodic cron that's scheduled at certain interval, for example every 30 seconds. * - * @param queueName on which queue message has to be send - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param message message object it could be any arbitrary object. * @param periodInMilliSeconds period of this job in milliseconds. * @return message id on successful enqueue otherwise null. */ @@ -634,9 +637,9 @@ default boolean enqueueUniqueAtWithPriority( * periodic cron that's scheduled at certain interval, for example every 30 seconds. * * @param queueName on which queue message has to be send - * @param message message object it could be any arbitrary object. - * @param period period of this job - * @param unit period unit + * @param message message object it could be any arbitrary object. + * @param period period of this job + * @param unit period unit * @return message id on successful enqueue otherwise null. */ default String enqueuePeriodic(String queueName, Object message, long period, TimeUnit unit) { @@ -648,8 +651,8 @@ default String enqueuePeriodic(String queueName, Object message, long period, Ti * periodic cron that's scheduled at certain interval, for example every 30 seconds. * * @param queueName on which queue message has to be send - * @param message message object it could be any arbitrary object. - * @param period job period + * @param message message object it could be any arbitrary object. + * @param period job period * @return message id on successful enqueue otherwise null. */ default String enqueuePeriodic(String queueName, Object message, Duration period) { @@ -660,9 +663,9 @@ default String enqueuePeriodic(String queueName, Object message, Duration period * Enqueue a message on given queue that will be running after a given period. It works like * periodic cron that's scheduled at certain interval, for example every 30 seconds. * - * @param queueName on which queue message has to be send - * @param messageId message id corresponding to this message - * @param message message object it could be any arbitrary object. + * @param queueName on which queue message has to be send + * @param messageId message id corresponding to this message + * @param message message object it could be any arbitrary object. * @param periodInMilliSeconds period of this job in milliseconds. * @return success or failure */ @@ -675,9 +678,9 @@ boolean enqueuePeriodic( * * @param queueName on which queue message has to be send * @param messageId message id corresponding to this message - * @param message message object it could be any arbitrary object. - * @param period period of this job . - * @param unit unit of this period + * @param message message object it could be any arbitrary object. + * @param period period of this job . + * @param unit unit of this period * @return success or failure */ default boolean enqueuePeriodic( @@ -691,12 +694,14 @@ default boolean enqueuePeriodic( * * @param queueName on which queue message has to be send * @param messageId message id corresponding to this message - * @param message message object it could be any arbitrary object. - * @param period period of this job . + * @param message message object it could be any arbitrary object. + * @param period period of this job . * @return success or failure */ default boolean enqueuePeriodic( String queueName, String messageId, Object message, Duration period) { return enqueuePeriodic(queueName, messageId, message, period.toMillis()); } + + MessageConverter getMessageConverter(); } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueMessageManager.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueMessageManager.java index 62b411a8..009642ac 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueMessageManager.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueMessageManager.java @@ -1,22 +1,23 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2020-2023 Sonu Kumar * - * Licensed 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 + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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 com.github.sonus21.rqueue.core; import com.github.sonus21.rqueue.config.RqueueConfig; +import com.github.sonus21.rqueue.utils.Constants; import com.github.sonus21.rqueue.utils.PriorityUtils; import java.util.List; import org.springframework.messaging.converter.MessageConverter; @@ -43,7 +44,7 @@ public interface RqueueMessageManager { * Delete all message for the given that has some priority like high,medium and low * * @param queueName queue name - * @param priority the priority for the queue + * @param priority the priority for the queue * @return fail/success */ default boolean deleteAllMessages(String queueName, String priority) { @@ -65,7 +66,7 @@ default boolean deleteAllMessages(String queueName, String priority) { * method {@link #getAllMessages(String)} * * @param queueName queue name to be query for - * @param priority the priority of the queue + * @param priority the priority of the queue * @return list of enqueued messages. */ default List getAllMessages(String queueName, String priority) { @@ -77,7 +78,7 @@ default List getAllMessages(String queueName, String priority) { * consumption message has a fixed lifetime. * * @param queueName queue name on which message was enqueued - * @param id message id + * @param id message id * @return the enqueued message, it could be null if message is not found or it's deleted. * @see RqueueConfig */ @@ -88,8 +89,8 @@ default List getAllMessages(String queueName, String priority) { * priority queue. * * @param queueName queue name on which message was enqueued - * @param priority the priority of the queue - * @param id message id + * @param priority the priority of the queue + * @param id message id * @return the enqueued message, it could be null if message is not found or it's deleted. */ default Object getMessage(String queueName, String priority, String id) { @@ -101,17 +102,18 @@ default Object getMessage(String queueName, String priority, String id) { * returns true/false. * * @param queueName queue name on which message was enqueued - * @param id message id + * @param id message id * @return whether the message exist or not */ boolean exist(String queueName, String id); /** - * Extension to the method {@link #exist(String, String)}, that checks message for priority queue. + * Extension to the method {@link #exist(String, String)}, that checks message for priority + * queue. * * @param queueName queue name on which message was enqueued - * @param priority priority of the given queue - * @param id message id + * @param priority priority of the given queue + * @param id message id * @return whether the message exist or not */ default boolean exist(String queueName, String priority, String id) { @@ -122,7 +124,7 @@ default boolean exist(String queueName, String priority, String id) { * Extension to the method {@link #getMessage(String, String)}, this returns internal message. * * @param queueName queue name on which message was enqueued - * @param id message id + * @param id message id * @return the enqueued message */ RqueueMessage getRqueueMessage(String queueName, String id); @@ -131,8 +133,8 @@ default boolean exist(String queueName, String priority, String id) { * Extension to the method {@link #getRqueueMessage(String, String)} * * @param queueName queue name on which message was enqueued - * @param priority the priority of the queue - * @param id message id + * @param priority the priority of the queue + * @param id message id * @return the enqueued message */ default RqueueMessage getRqueueMessage(String queueName, String priority, String id) { @@ -151,7 +153,7 @@ default RqueueMessage getRqueueMessage(String queueName, String priority, String * Extension to the method {@link #getAllRqueueMessage(String)} * * @param queueName queue name on which message was enqueued - * @param priority the priority of the queue + * @param priority the priority of the queue * @return the enqueued message */ default List getAllRqueueMessage(String queueName, String priority) { @@ -171,7 +173,7 @@ default List getAllRqueueMessage(String queueName, String priorit * Delete a message that's enqueued to a queue with some priority * * @param queueName queue on which message was enqueued - * @param priority priority of the message like high/low/medium + * @param priority priority of the message like high/low/medium * @param messageId messageId corresponding to this message * @return success/failure */ @@ -185,4 +187,27 @@ default boolean deleteMessage(String queueName, String priority, String messageI * @return message converter that's used for message (de)serialization */ MessageConverter getMessageConverter(); + + /** + * Move messages from Dead Letter queue to the destination queue. This push the messages at the + * FRONT of destination queue, so that it can be reprocessed as soon as possible. + * + * @param deadLetterQueueName dead letter queue name + * @param queueName queue name + * @param maxMessages number of messages to be moved by default move + * {@link Constants#MAX_MESSAGES} messages + * @return success or failure. + */ + boolean moveMessageFromDeadLetterToQueue( + String deadLetterQueueName, String queueName, Integer maxMessages); + + /** + * A shortcut to the method {@link #moveMessageFromDeadLetterToQueue(String, String, Integer)} + * + * @param deadLetterQueueName dead letter queue name + * @param queueName queue name + * @return success or failure + */ + boolean moveMessageFromDeadLetterToQueue(String deadLetterQueueName, String queueName); + } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java index 860da4b5..3002a204 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java @@ -1,16 +1,16 @@ /* - * Copyright 2022 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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -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; @@ -66,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; } @@ -109,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( @@ -121,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( @@ -143,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) { diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/ReactiveRqueueMessageEnqueuerImpl.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/ReactiveRqueueMessageEnqueuerImpl.java index b1f2bb20..bf752c0b 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/ReactiveRqueueMessageEnqueuerImpl.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/ReactiveRqueueMessageEnqueuerImpl.java @@ -1,34 +1,29 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2021-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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 com.github.sonus21.rqueue.core.impl; -import static com.github.sonus21.rqueue.utils.Validator.validateDelay; -import static com.github.sonus21.rqueue.utils.Validator.validateMessage; -import static com.github.sonus21.rqueue.utils.Validator.validateMessageId; -import static com.github.sonus21.rqueue.utils.Validator.validatePeriod; -import static com.github.sonus21.rqueue.utils.Validator.validatePriority; -import static com.github.sonus21.rqueue.utils.Validator.validateQueue; -import static com.github.sonus21.rqueue.utils.Validator.validateRetryCount; +import static com.github.sonus21.rqueue.utils.Validator.*; import com.github.sonus21.rqueue.core.EndpointRegistry; import com.github.sonus21.rqueue.core.ReactiveRqueueMessageEnqueuer; import com.github.sonus21.rqueue.core.RqueueMessage; import com.github.sonus21.rqueue.core.RqueueMessageTemplate; import com.github.sonus21.rqueue.core.support.RqueueMessageUtils; +import com.github.sonus21.rqueue.exception.DuplicateMessageException; import com.github.sonus21.rqueue.listener.QueueDetail; import com.github.sonus21.rqueue.utils.PriorityUtils; import lombok.extern.slf4j.Slf4j; @@ -37,6 +32,9 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import java.util.function.BiFunction; +import java.util.function.Function; + @Slf4j public class ReactiveRqueueMessageEnqueuerImpl extends BaseMessageSender implements ReactiveRqueueMessageEnqueuer { @@ -56,7 +54,8 @@ private Mono pushReactiveMessage( Object message, Integer retryCount, Long delayInMilliSecs, - MonoConverterGenerator monoConverterGenerator) { + boolean isUnique, + Function> monoConverter) { QueueDetail queueDetail = EndpointRegistry.get(queueName); RqueueMessage rqueueMessage = builder.build( @@ -67,24 +66,46 @@ private Mono pushReactiveMessage( retryCount, delayInMilliSecs, messageHeaders); - MonoConverter monoConverter = monoConverterGenerator.create(rqueueMessage); try { - Object o1 = enqueue(queueDetail, rqueueMessage, delayInMilliSecs, true); - Mono storeMessageResult = - (Mono) storeMessageMetadata(rqueueMessage, delayInMilliSecs, true); - Mono longMono; - if (o1 instanceof Flux) { - longMono = ((Flux) o1).elementAt(0); - } else { - longMono = (Mono) o1; - } - return longMono.zipWith(storeMessageResult, monoConverter::call); + Mono storeResult = + (Mono) storeMessageMetadata(rqueueMessage, delayInMilliSecs, true, isUnique); + return storeResult.flatMap( + success -> { + if (Boolean.TRUE.equals(success)) { + Object result = enqueue(queueDetail, rqueueMessage, delayInMilliSecs, true); + Mono enqueueMono; + if (result instanceof Flux) { + enqueueMono = ((Flux) result).next(); + } else if (result instanceof Mono) { + enqueueMono = (Mono) result; + } else { + return Mono.error( + new IllegalStateException( + "Unexpected enqueue result type: " + result.getClass())); + } + return enqueueMono.flatMap(ignore -> monoConverter.apply(rqueueMessage)); + } else { + return Mono.error(new DuplicateMessageException(rqueueMessage.getId())); + } + }); } catch (Exception e) { - log.error("Queue: {} Message {} could not be pushed {}", queueName, rqueueMessage, e); + log.error( + "Failed to enqueue message [{}] to queue [{}]", rqueueMessage.getId(), queueName, e); return Mono.error(e); } } + private void validateBasic(String queue, Object msg) { + validateQueue(queue); + validateMessage(msg); + } + + private void validateWithId(String queue, String id, Object msg) { + validateQueue(queue); + validateMessageId(id); + validateMessage(msg); + } + private Mono pushReactiveMessage( String queueName, Object message, Integer retryCount, Long delayInMilliSecs) { return pushReactiveMessage( @@ -94,7 +115,8 @@ private Mono pushReactiveMessage( message, retryCount, delayInMilliSecs, - new StrMonoConverterGenerator()); + false, + (rqueueMessage) -> Mono.just(rqueueMessage.getId())); } private Mono pushReactiveWithMessageId( @@ -102,7 +124,8 @@ private Mono pushReactiveWithMessageId( String messageId, Object message, Integer retryCount, - Long delayInMilliSecs) { + Long delayInMilliSecs, + boolean isUnique) { return pushReactiveMessage( RqueueMessageUtils::buildMessage, queueName, @@ -110,7 +133,8 @@ private Mono pushReactiveWithMessageId( message, retryCount, delayInMilliSecs, - new BooleanMonoConverterGenerator()); + isUnique, + (rqueueMessage) -> Mono.just(Boolean.TRUE)); } private Mono pushReactivePeriodicMessage( @@ -122,7 +146,8 @@ private Mono pushReactivePeriodicMessage( message, null, periodInMilliSeconds, - new StrMonoConverterGenerator()); + false, + (rqueueMessage) -> Mono.just(rqueueMessage.getId())); } private Mono pushReactivePeriodicMessageWithMessageId( @@ -134,37 +159,31 @@ private Mono pushReactivePeriodicMessageWithMessageId( message, null, periodInMilliSeconds, - new BooleanMonoConverterGenerator()); + false, + (rqueueMessage) -> Mono.just(Boolean.TRUE)); } @Override public Mono enqueue(String queueName, Object message) { - validateQueue(queueName); - validateMessage(message); + validateBasic(queueName, message); return pushReactiveMessage(queueName, message, null, null); } @Override public Mono enqueue(String queueName, String messageId, Object message) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); - return pushReactiveWithMessageId(queueName, messageId, message, null, null); + validateWithId(queueName, messageId, message); + return pushReactiveWithMessageId(queueName, messageId, message, null, null, false); } @Override public Mono enqueueUnique(String queueName, String messageId, Object message) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); - // TODO uniqueness - return pushReactiveWithMessageId(queueName, messageId, message, null, null); + validateWithId(queueName, messageId, message); + return pushReactiveWithMessageId(queueName, messageId, message, null, null, true); } @Override public Mono enqueueWithRetry(String queueName, Object message, int retryCount) { - validateQueue(queueName); - validateMessage(message); + validateBasic(queueName, message); validateRetryCount(retryCount); return pushReactiveMessage(queueName, message, retryCount, null); } @@ -172,11 +191,9 @@ public Mono enqueueWithRetry(String queueName, Object message, int retry @Override public Mono enqueueWithRetry( String queueName, String messageId, Object message, int retryCount) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); + validateWithId(queueName, messageId, message); validateRetryCount(retryCount); - return pushReactiveWithMessageId(queueName, messageId, message, retryCount, null); + return pushReactiveWithMessageId(queueName, messageId, message, retryCount, null, false); } @Override @@ -191,18 +208,20 @@ public Mono enqueueWithPriority(String queueName, String priority, Objec @Override public Mono enqueueWithPriority( String queueName, String priority, String messageId, Object message) { - validateQueue(queueName); + validateWithId(queueName, messageId, message); validatePriority(priority); - validateMessageId(messageId); - validateMessage(message); return pushReactiveWithMessageId( - PriorityUtils.getQueueNameForPriority(queueName, priority), messageId, message, null, null); + PriorityUtils.getQueueNameForPriority(queueName, priority), + messageId, + message, + null, + null, + false); } @Override public Mono enqueueIn(String queueName, Object message, long delayInMilliSecs) { - validateQueue(queueName); - validateMessage(message); + validateBasic(queueName, message); validateDelay(delayInMilliSecs); return pushReactiveMessage(queueName, message, null, delayInMilliSecs); } @@ -210,29 +229,23 @@ public Mono enqueueIn(String queueName, Object message, long delayInMill @Override public Mono enqueueIn( String queueName, String messageId, Object message, long delayInMilliSecs) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); + validateWithId(queueName, messageId, message); validateDelay(delayInMilliSecs); - return pushReactiveWithMessageId(queueName, messageId, message, null, delayInMilliSecs); + return pushReactiveWithMessageId(queueName, messageId, message, null, delayInMilliSecs, false); } @Override public Mono enqueueUniqueIn( String queueName, String messageId, Object message, long delayInMillisecond) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); + validateWithId(queueName, messageId, message); validateDelay(delayInMillisecond); - // TODO unique?? - return pushReactiveWithMessageId(queueName, messageId, message, null, delayInMillisecond); + return pushReactiveWithMessageId(queueName, messageId, message, null, delayInMillisecond, true); } @Override public Mono enqueueInWithRetry( String queueName, Object message, int retryCount, long delayInMilliSecs) { - validateQueue(queueName); - validateMessage(message); + validateBasic(queueName, message); validateRetryCount(retryCount); validateDelay(delayInMilliSecs); return pushReactiveMessage(queueName, message, retryCount, delayInMilliSecs); @@ -241,18 +254,16 @@ public Mono enqueueInWithRetry( @Override public Mono enqueueInWithRetry( String queueName, String messageId, Object message, int retryCount, long delayInMilliSecs) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); - validateDelay(retryCount); + validateWithId(queueName, messageId, message); + validateRetryCount(retryCount); validateDelay(delayInMilliSecs); - return pushReactiveWithMessageId(queueName, messageId, message, retryCount, delayInMilliSecs); + return pushReactiveWithMessageId( + queueName, messageId, message, retryCount, delayInMilliSecs, false); } @Override public Mono enqueuePeriodic(String queueName, Object message, long periodInMilliSeconds) { - validateQueue(queueName); - validateMessage(message); + validateBasic(queueName, message); validatePeriod(periodInMilliSeconds); return pushReactivePeriodicMessage(queueName, message, periodInMilliSeconds); } @@ -260,21 +271,19 @@ public Mono enqueuePeriodic(String queueName, Object message, long perio @Override public Mono enqueuePeriodic( String queueName, String messageId, Object message, long periodInMilliSeconds) { - validateQueue(queueName); - validateMessage(message); - validateMessageId(messageId); + validateWithId(queueName, messageId, message); validatePeriod(periodInMilliSeconds); return pushReactivePeriodicMessageWithMessageId( queueName, messageId, message, periodInMilliSeconds); } + @FunctionalInterface private interface MonoConverter { - - T call(Long a, Boolean b); + T convert(Long id, Boolean success); } + @FunctionalInterface private interface MessageBuilder { - RqueueMessage build( MessageConverter converter, String queueName, @@ -284,49 +293,4 @@ RqueueMessage build( Long delay, MessageHeaders messageHeaders); } - - private interface MonoConverterGenerator { - - MonoConverter create(RqueueMessage rqueueMessage); - } - - private static class StrMonoConverter implements MonoConverter { - - private final RqueueMessage message; - - private StrMonoConverter(RqueueMessage message) { - this.message = message; - } - - @Override - public String call(Long a, Boolean b) { - return message.getId(); - } - } - - private static class BoolMonoConverter implements MonoConverter { - - private BoolMonoConverter(RqueueMessage message) {} - - @Override - public Boolean call(Long a, Boolean b) { - return Boolean.TRUE; - } - } - - private static class StrMonoConverterGenerator implements MonoConverterGenerator { - - @Override - public MonoConverter create(RqueueMessage rqueueMessage) { - return new StrMonoConverter(rqueueMessage); - } - } - - private static class BooleanMonoConverterGenerator implements MonoConverterGenerator { - - @Override - public MonoConverter create(RqueueMessage rqueueMessage) { - return new BoolMonoConverter(rqueueMessage); - } - } } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageEnqueuerImpl.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageEnqueuerImpl.java index c333f7e0..e4f4fc04 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageEnqueuerImpl.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageEnqueuerImpl.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -31,6 +31,8 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; +import java.util.Objects; + @Slf4j public class RqueueMessageEnqueuerImpl extends BaseMessageSender implements RqueueMessageEnqueuer { @@ -41,129 +43,136 @@ public RqueueMessageEnqueuerImpl( super(messageTemplate, messageConverter, messageHeaders); } + private void validateBasic(String queue, Object message) { + validateQueue(queue); + validateMessage(message); + } + + private void validateWithId(String queue, String messageId, Object message) { + validateQueue(queue); + validateMessageId(messageId); + validateMessage(message); + } + @Override public String enqueue(String queueName, Object message) { - validateQueue(queueName); - validateMessage(message); - return pushMessage(queueName, null, message, null, null); + validateBasic(queueName, message); + return pushMessage(queueName, null, message, null, null, false); } @Override public boolean enqueue(String queueName, String messageId, Object message) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); - return pushMessage(queueName, messageId, message, null, null) != null; + validateWithId(queueName, messageId, message); + return pushMessage(queueName, messageId, message, null, null, false) != null; } @Override public boolean enqueueUnique(String queueName, String messageId, Object message) { - // TODO? is using monotonic time sufficient for handling uniqueness - return enqueue(queueName, messageId, message); + validateWithId(queueName, messageId, message); + return Objects.nonNull(pushMessage(queueName, messageId, message, null, null, true)); } @Override public String enqueueWithRetry(String queueName, Object message, int retryCount) { - validateQueue(queueName); - validateMessage(message); + validateBasic(queueName, message); validateRetryCount(retryCount); - return pushMessage(queueName, null, message, retryCount, null); + return pushMessage(queueName, null, message, retryCount, null, false); } @Override public boolean enqueueWithRetry( String queueName, String messageId, Object message, int retryCount) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); + validateWithId(queueName, messageId, message); validateRetryCount(retryCount); - return pushMessage(queueName, messageId, message, retryCount, null) != null; + return pushMessage(queueName, messageId, message, retryCount, null, false) != null; } @Override public String enqueueWithPriority(String queueName, String priority, Object message) { - validateQueue(queueName); + validateBasic(queueName, message); validatePriority(priority); - validateMessage(message); return pushMessage( - PriorityUtils.getQueueNameForPriority(queueName, priority), null, message, null, null); + PriorityUtils.getQueueNameForPriority(queueName, priority), + null, + message, + null, + null, + false); } @Override public boolean enqueueWithPriority( String queueName, String priority, String messageId, Object message) { - validateQueue(queueName); + validateWithId(queueName, messageId, message); validatePriority(priority); - validateMessageId(messageId); - validateMessage(message); return pushMessage( PriorityUtils.getQueueNameForPriority(queueName, priority), messageId, message, null, - null) + null, + false) != null; } @Override public String enqueueIn(String queueName, Object message, long delayInMilliSecs) { - validateQueue(queueName); - validateMessage(message); + validateBasic(queueName, message); validateDelay(delayInMilliSecs); - return pushMessage(queueName, null, message, null, delayInMilliSecs); + return pushMessage(queueName, null, message, null, delayInMilliSecs, false); } @Override public boolean enqueueIn( String queueName, String messageId, Object message, long delayInMilliSecs) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); + validateWithId(queueName, messageId, message); validateDelay(delayInMilliSecs); - return pushMessage(queueName, messageId, message, null, delayInMilliSecs) != null; + return pushMessage(queueName, messageId, message, null, delayInMilliSecs, false) != null; } @Override public boolean enqueueUniqueIn( - String queueName, String messageId, Object message, long delayInMillisecond) { - return enqueueIn(queueName, messageId, message, delayInMillisecond); + String queueName, String messageId, Object message, long delayInMilliSecs) { + validateWithId(queueName, messageId, message); + validateDelay(delayInMilliSecs); + return Objects.nonNull( + pushMessage(queueName, messageId, message, null, delayInMilliSecs, true)); } @Override public String enqueueInWithRetry( String queueName, Object message, int retryCount, long delayInMilliSecs) { - validateQueue(queueName); - validateMessage(message); + validateBasic(queueName, message); validateRetryCount(retryCount); validateDelay(delayInMilliSecs); - return pushMessage(queueName, null, message, retryCount, delayInMilliSecs); + return pushMessage(queueName, null, message, retryCount, delayInMilliSecs, false); } @Override public boolean enqueueInWithRetry( String queueName, String messageId, Object message, int retryCount, long delayInMilliSecs) { - validateQueue(queueName); - validateMessageId(messageId); - validateMessage(message); + validateWithId(queueName, messageId, message); validateRetryCount(retryCount); validateDelay(delayInMilliSecs); - return pushMessage(queueName, messageId, message, retryCount, delayInMilliSecs) != null; + return pushMessage(queueName, messageId, message, retryCount, delayInMilliSecs, false) != null; } @Override public String enqueuePeriodic(String queueName, Object message, long period) { - validateQueue(queueName); - validateMessage(message); + validateBasic(queueName, message); validatePeriod(period); return pushPeriodicMessage(queueName, null, message, period); } @Override public boolean enqueuePeriodic(String queueName, String messageId, Object message, long period) { - validateMessageId(messageId); - validateQueue(queueName); - validateMessage(message); + validateWithId(queueName, messageId, message); validatePeriod(period); return pushPeriodicMessage(queueName, messageId, message, period) != null; } + + @Override + public MessageConverter getMessageConverter() { + return messageConverter; + } } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageManagerImpl.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageManagerImpl.java index 7d799547..9e63256b 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageManagerImpl.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageManagerImpl.java @@ -1,21 +1,24 @@ /* - * Copyright 2022 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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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 com.github.sonus21.rqueue.core.impl; +import static org.springframework.util.Assert.isTrue; +import static org.springframework.util.Assert.notNull; + import com.github.sonus21.rqueue.common.RqueueLockManager; import com.github.sonus21.rqueue.core.EndpointRegistry; import com.github.sonus21.rqueue.core.RqueueMessage; @@ -25,7 +28,13 @@ import com.github.sonus21.rqueue.exception.LockCanNotBeAcquired; import com.github.sonus21.rqueue.listener.QueueDetail; import com.github.sonus21.rqueue.listener.RqueueMessageHeaders; +import com.github.sonus21.rqueue.models.MessageMoveResult; import com.github.sonus21.rqueue.models.db.MessageMetadata; +import com.github.sonus21.rqueue.utils.Constants; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.Message; @@ -33,11 +42,6 @@ import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.MessageBuilder; -import java.time.Duration; -import java.util.ArrayList; -import java.util.List; -import java.util.UUID; - @Slf4j public class RqueueMessageManagerImpl extends BaseMessageSender implements RqueueMessageManager { @@ -104,19 +108,14 @@ public List getAllRqueueMessage(String queueName) { @Override public boolean exist(String queueName, String id) { - String lockValue = UUID.randomUUID().toString(); - if (rqueueLockManager.acquireLock(queueName, lockValue, Duration.ofSeconds(1))) { - boolean exist = getMessage(queueName, id) != null; - rqueueLockManager.releaseLock(queueName, lockValue); - return exist; - } - throw new LockCanNotBeAcquired(queueName); + return getMessage(queueName, id) != null; } @Override public boolean deleteMessage(String queueName, String id) { RqueueMessage rqueueMessage = getRqueueMessage(queueName, id); if (rqueueMessage == null) { + log.error("Delete message failed, no such message: {}", id); return false; } Duration duration = rqueueConfig.getMessageDurability(rqueueMessage.getPeriod()); @@ -127,4 +126,30 @@ public boolean deleteMessage(String queueName, String id) { public MessageConverter getMessageConverter() { return messageConverter; } + + @Override + public boolean moveMessageFromDeadLetterToQueue( + String deadLetterQueueName, String queueName, Integer maxMessages) { + return moveMessageListToList(deadLetterQueueName, queueName, maxMessages).isSuccess(); + } + + @Override + public boolean moveMessageFromDeadLetterToQueue(String deadLetterQueueName, String queueName) { + return moveMessageListToList(deadLetterQueueName, queueName, null).isSuccess(); + } + + private MessageMoveResult moveMessageListToList( + String sourceQueue, String destinationQueue, Integer maxMessage) { + notNull(sourceQueue, "sourceQueue must not be null"); + notNull(destinationQueue, "destinationQueue must not be null"); + isTrue( + !sourceQueue.equals(destinationQueue), + "sourceQueue and destinationQueue must be different"); + Integer messageCount = maxMessage; + if (messageCount == null) { + messageCount = Constants.MAX_MESSAGES; + } + isTrue(messageCount > 0, "maxMessage must be greater than zero"); + return messageTemplate.moveMessageListToList(sourceQueue, destinationQueue, messageCount); + } } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageSenderImpl.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageSenderImpl.java index 9a30d5f4..f0afb4a8 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageSenderImpl.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageSenderImpl.java @@ -54,7 +54,7 @@ public RqueueMessageSenderImpl( public boolean enqueue(String queueName, Object message) { validateQueue(queueName); validateMessage(message); - return pushMessage(queueName, null, message, null, null) != null; + return pushMessage(queueName, null, message, null, null, false) != null; } @Override @@ -62,7 +62,7 @@ public boolean enqueueWithRetry(String queueName, Object message, int retryCount validateQueue(queueName); validateMessage(message); validateRetryCount(retryCount); - return pushMessage(queueName, null, message, retryCount, null) != null; + return pushMessage(queueName, null, message, retryCount, null, false) != null; } @Override @@ -71,7 +71,7 @@ public boolean enqueueWithPriority(String queueName, String priority, Object mes validatePriority(priority); validateMessage(message); return pushMessage( - PriorityUtils.getQueueNameForPriority(queueName, priority), null, message, null, null) + PriorityUtils.getQueueNameForPriority(queueName, priority), null, message, null, null, false) != null; } @@ -80,7 +80,7 @@ public boolean enqueueIn(String queueName, Object message, long delayInMilliSecs validateQueue(queueName); validateMessage(message); validateDelay(delayInMilliSecs); - return pushMessage(queueName, null, message, null, delayInMilliSecs) != null; + return pushMessage(queueName, null, message, null, delayInMilliSecs, false) != null; } @Override @@ -90,7 +90,7 @@ public boolean enqueueInWithRetry( validateMessage(message); validateRetryCount(retryCount); validateDelay(delayInMilliSecs); - return pushMessage(queueName, null, message, retryCount, delayInMilliSecs) != null; + return pushMessage(queueName, null, message, retryCount, delayInMilliSecs, false) != null; } @Override @@ -105,7 +105,8 @@ public boolean enqueueInWithPriority( null, message, null, - delayInMilliSecs) + delayInMilliSecs, + false) != null; } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/dao/RqueueMessageMetadataDao.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/dao/RqueueMessageMetadataDao.java index c59bb8d7..05be7034 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/dao/RqueueMessageMetadataDao.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/dao/RqueueMessageMetadataDao.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2019-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -28,11 +28,11 @@ public interface RqueueMessageMetadataDao { List findAll(Collection ids); - void save(MessageMetadata messageMetadata, Duration ttl); + void save(MessageMetadata messageMetadata, Duration ttl, boolean checkUniqueNess); void delete(String id); void deleteAll(Collection ids); - Mono saveReactive(MessageMetadata messageMetadata, Duration duration); + Mono saveReactive(MessageMetadata messageMetadata, Duration duration, boolean isUnique); } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/dao/impl/RqueueMessageMetadataDaoImpl.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/dao/impl/RqueueMessageMetadataDaoImpl.java index 7e99408b..c045b9f1 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/dao/impl/RqueueMessageMetadataDaoImpl.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/dao/impl/RqueueMessageMetadataDaoImpl.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2021-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -20,6 +20,7 @@ import com.github.sonus21.rqueue.common.RqueueRedisTemplate; import com.github.sonus21.rqueue.config.RqueueConfig; import com.github.sonus21.rqueue.dao.RqueueMessageMetadataDao; +import com.github.sonus21.rqueue.exception.DuplicateMessageException; import com.github.sonus21.rqueue.models.db.MessageMetadata; import java.time.Duration; import java.util.Collection; @@ -57,8 +58,15 @@ public List findAll(Collection ids) { } @Override - public void save(MessageMetadata messageMetadata, Duration duration) { + public void save(MessageMetadata messageMetadata, Duration duration,boolean checkUniqueNess) { Assert.notNull(messageMetadata.getId(), "messageMetadata id cannot be null"); + if(checkUniqueNess){ + Boolean value = template.setIfAbsent(messageMetadata.getId(), messageMetadata, duration); + if(Boolean.FALSE.equals(value)){ + throw new DuplicateMessageException(messageMetadata.getId()); + } + return; + } template.set(messageMetadata.getId(), messageMetadata, duration); } @@ -73,8 +81,19 @@ public void deleteAll(Collection ids) { } @Override - public Mono saveReactive(MessageMetadata messageMetadata, Duration ttl) { + public Mono saveReactive(MessageMetadata messageMetadata, Duration ttl, boolean isUnique) { Assert.notNull(messageMetadata.getId(), "messageMetadata id cannot be null"); + if(isUnique){ + return reactiveRedisTemplate.template().opsForValue() + .setIfAbsent(messageMetadata.getId(), messageMetadata) + .flatMap(result -> { + if (Boolean.TRUE.equals(result)) { + return reactiveRedisTemplate.template().expire(messageMetadata.getId(), ttl) + .thenReturn(true); + } + return Mono.just(false); + }); + } return reactiveRedisTemplate .template() .opsForValue() diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/exception/DuplicateMessageException.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/exception/DuplicateMessageException.java new file mode 100644 index 00000000..9aa4acf5 --- /dev/null +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/exception/DuplicateMessageException.java @@ -0,0 +1,24 @@ +/* + * Copyright (c) 2015-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. + * You may obtain a copy of the License at + * + * https://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 com.github.sonus21.rqueue.exception; + +public class DuplicateMessageException extends RuntimeException { + + public DuplicateMessageException(String message) { + super(message); + } +} \ No newline at end of file diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/JobImpl.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/JobImpl.java index 369eb2a4..220346c8 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/JobImpl.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/JobImpl.java @@ -1,16 +1,16 @@ /* - * Copyright 2022 Sonu Kumar + * Copyright (c) 2021-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -34,14 +34,14 @@ import com.github.sonus21.rqueue.utils.Constants; import com.github.sonus21.rqueue.utils.TimeoutUtils; import com.github.sonus21.rqueue.web.service.RqueueMessageMetadataService; -import lombok.extern.slf4j.Slf4j; -import org.springframework.data.redis.RedisSystemException; -import org.springframework.util.CollectionUtils; import java.io.Serializable; import java.time.Duration; import java.util.List; import java.util.UUID; import java.util.concurrent.Callable; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.RedisSystemException; +import org.springframework.util.CollectionUtils; @Slf4j @SuppressWarnings("java:S107") @@ -336,14 +336,14 @@ private void saveMessageMetadata(Callable callable) { void updateMessageStatus(MessageStatus messageStatus) { setMessageStatus(messageStatus); // We need to address these problems with message metadata - // 1. Message was deleted while executing, this means local copy is stale + // 1. The Message was deleted while executing; this means local copy is stale // 2. Parallel update is being made [dashboard operation, periodic job (two periodic jobs can // run in parallel due to failure)] if (!messageStatus.isTerminalState() || getRqueueMessage().isPeriodic()) { Duration duration = rqueueConfig.getMessageDurability(getRqueueMessage().getPeriod()); saveMessageMetadata( () -> { - messageMetadataService.save(getMessageMetadata(), duration); + messageMetadataService.save(getMessageMetadata(), duration, false); return null; }); } else { diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/Constants.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/Constants.java index 064d804c..05d42b51 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/Constants.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/Constants.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -54,10 +54,16 @@ public final class Constants { public static final int MIN_CONCURRENCY = 1; public static final long MINIMUM_JOB_PERIOD = 1000L; public static final String QUEUE_CRUD_LOCK_KEY_PREFIX = "q-crud::"; + public static final String MESSAGE_LOCK_KEY_PREFIX = "msg::"; - private Constants() {} + private Constants() { + } public static String getQueueCrudLockKey(RqueueConfig rqueueConfig, String queueName) { return rqueueConfig.getLockKey(QUEUE_CRUD_LOCK_KEY_PREFIX + queueName); } + + public static String getMessageLockName(RqueueConfig rqueueConfig, String messageId) { + return rqueueConfig.getLockKey(MESSAGE_LOCK_KEY_PREFIX + messageId); + } } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/web/service/RqueueMessageMetadataService.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/web/service/RqueueMessageMetadataService.java index 67a75fd0..1dd29585 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/web/service/RqueueMessageMetadataService.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/web/service/RqueueMessageMetadataService.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2019-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -18,12 +18,11 @@ import com.github.sonus21.rqueue.core.RqueueMessage; import com.github.sonus21.rqueue.models.db.MessageMetadata; -import org.springframework.data.redis.core.ZSetOperations.TypedTuple; -import reactor.core.publisher.Mono; - import java.time.Duration; import java.util.Collection; import java.util.List; +import org.springframework.data.redis.core.ZSetOperations.TypedTuple; +import reactor.core.publisher.Mono; public interface RqueueMessageMetadataService { @@ -35,7 +34,7 @@ public interface RqueueMessageMetadataService { List findAll(Collection ids); - void save(MessageMetadata messageMetadata, Duration ttl); + void save(MessageMetadata messageMetadata, Duration ttl, boolean checkUnique); MessageMetadata getByMessageId(String queueName, String messageId); @@ -43,7 +42,7 @@ public interface RqueueMessageMetadataService { MessageMetadata getOrCreateMessageMetadata(RqueueMessage rqueueMessage); - Mono saveReactive(MessageMetadata messageMetadata, Duration ttl); + Mono saveReactive(MessageMetadata messageMetadata, Duration ttl, boolean checkUnique); List> readMessageMetadataForQueue( String queueName, long start, long end); diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/web/service/impl/RqueueMessageMetadataServiceImpl.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/web/service/impl/RqueueMessageMetadataServiceImpl.java index 37bcc019..a84d437d 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/web/service/impl/RqueueMessageMetadataServiceImpl.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/web/service/impl/RqueueMessageMetadataServiceImpl.java @@ -1,38 +1,45 @@ /* - * Copyright 2021 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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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 com.github.sonus21.rqueue.web.service.impl; import com.github.sonus21.rqueue.common.RqueueLockManager; +import com.github.sonus21.rqueue.config.RqueueConfig; import com.github.sonus21.rqueue.core.RqueueMessage; import com.github.sonus21.rqueue.core.support.RqueueMessageUtils; import com.github.sonus21.rqueue.dao.RqueueMessageMetadataDao; import com.github.sonus21.rqueue.dao.RqueueStringDao; import com.github.sonus21.rqueue.models.db.MessageMetadata; import com.github.sonus21.rqueue.models.enums.MessageStatus; +import com.github.sonus21.rqueue.utils.Constants; import com.github.sonus21.rqueue.web.service.RqueueMessageMetadataService; +import java.time.Duration; +import java.util.Collection; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; +import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.DefaultTypedTuple; import org.springframework.data.redis.core.ZSetOperations.TypedTuple; import org.springframework.stereotype.Service; import reactor.core.publisher.Mono; -import java.time.Duration; -import java.util.*; -import java.util.stream.Collectors; @Service @Slf4j @@ -41,15 +48,18 @@ public class RqueueMessageMetadataServiceImpl implements RqueueMessageMetadataSe private final RqueueMessageMetadataDao rqueueMessageMetadataDao; private final RqueueStringDao rqueueStringDao; private final RqueueLockManager lockManager; + private final RqueueConfig rqueueConfig; @Autowired public RqueueMessageMetadataServiceImpl( RqueueMessageMetadataDao rqueueMessageMetadataDao, RqueueStringDao rqueueStringDao, - RqueueLockManager rqueueLockManager) { + RqueueLockManager rqueueLockManager, + RqueueConfig rqueueConfig) { this.rqueueMessageMetadataDao = rqueueMessageMetadataDao; this.rqueueStringDao = rqueueStringDao; this.lockManager = rqueueLockManager; + this.rqueueConfig = rqueueConfig; } @Override @@ -75,8 +85,8 @@ public List findAll(Collection ids) { } @Override - public void save(MessageMetadata messageMetadata, Duration duration) { - rqueueMessageMetadataDao.save(messageMetadata, duration); + public void save(MessageMetadata messageMetadata, Duration duration, boolean checkUnique) { + rqueueMessageMetadataDao.save(messageMetadata, duration, checkUnique); } @Override @@ -88,8 +98,9 @@ public MessageMetadata getByMessageId(String queueName, String messageId) { @Override public boolean deleteMessage(String queueName, String messageId, Duration duration) { String lockValue = UUID.randomUUID().toString(); + String lockKey = Constants.getMessageLockName(rqueueConfig, messageId); try { - if (lockManager.acquireLock(messageId, lockValue, Duration.ofSeconds(1))) { + if (lockManager.acquireLock(lockKey, lockValue, Duration.ofSeconds(1))) { String id = RqueueMessageUtils.getMessageMetaId(queueName, messageId); MessageMetadata messageMetadata = rqueueMessageMetadataDao.get(id); if (messageMetadata == null) { @@ -97,11 +108,14 @@ public boolean deleteMessage(String queueName, String messageId, Duration durati } messageMetadata.setDeleted(true); messageMetadata.setDeletedOn(System.currentTimeMillis()); - save(messageMetadata, duration); + save(messageMetadata, duration, false); + log.debug("message deleted, id: {}", id); return true; + } else { + log.error("Lock could not be acquired, Id: {}", messageId); } } finally { - lockManager.releaseLock(messageId, lockValue); + lockManager.releaseLock(lockKey, lockValue); } return false; } @@ -117,8 +131,9 @@ public MessageMetadata getOrCreateMessageMetadata(RqueueMessage rqueueMessage) { } @Override - public Mono saveReactive(MessageMetadata messageMetadata, Duration duration) { - return rqueueMessageMetadataDao.saveReactive(messageMetadata, duration); + public Mono saveReactive( + MessageMetadata messageMetadata, Duration duration, boolean isUnique) { + return rqueueMessageMetadataDao.saveReactive(messageMetadata, duration, isUnique); } @Override @@ -151,7 +166,7 @@ public List> readMessageMetadataForQueue( public void saveMessageMetadataForQueue( String queueName, MessageMetadata messageMetadata, Long ttlInMillisecond) { messageMetadata.setUpdatedOn(System.currentTimeMillis()); - save(messageMetadata, Duration.ofMillis(ttlInMillisecond)); + save(messageMetadata, Duration.ofMillis(ttlInMillisecond), false); rqueueStringDao.addToOrderedSetWithScore( queueName, messageMetadata.getId(), -(System.currentTimeMillis() + ttlInMillisecond)); } diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/impl/RqueueMessageManagerImplTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/impl/RqueueMessageManagerImplTest.java index 364f3fab..ceca3a83 100644 --- a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/impl/RqueueMessageManagerImplTest.java +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/impl/RqueueMessageManagerImplTest.java @@ -1,28 +1,32 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2021-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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 com.github.sonus21.rqueue.core.impl; +import static org.apache.commons.lang3.reflect.FieldUtils.writeField; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -30,82 +34,313 @@ import com.github.sonus21.TestBase; import com.github.sonus21.rqueue.CoreUnitTest; import com.github.sonus21.rqueue.common.RqueueLockManager; +import com.github.sonus21.rqueue.config.RqueueConfig; import com.github.sonus21.rqueue.core.DefaultRqueueMessageConverter; +import com.github.sonus21.rqueue.core.EndpointRegistry; +import com.github.sonus21.rqueue.core.RqueueMessage; import com.github.sonus21.rqueue.core.RqueueMessageManager; import com.github.sonus21.rqueue.core.RqueueMessageTemplate; import com.github.sonus21.rqueue.exception.LockCanNotBeAcquired; -import com.github.sonus21.rqueue.listener.RqueueMessageHeaders; +import com.github.sonus21.rqueue.listener.QueueDetail; +import com.github.sonus21.rqueue.models.MessageMoveResult; +import com.github.sonus21.rqueue.models.db.MessageMetadata; +import com.github.sonus21.rqueue.utils.MessageMetadataTestUtils; +import com.github.sonus21.rqueue.utils.PriorityUtils; import com.github.sonus21.rqueue.utils.TestUtils; import com.github.sonus21.rqueue.web.service.RqueueMessageMetadataService; +import java.time.Duration; +import java.util.Collections; +import java.util.List; import java.util.UUID; -import org.apache.commons.lang3.reflect.FieldUtils; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.Mockito; import org.mockito.MockitoAnnotations; -import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; @CoreUnitTest class RqueueMessageManagerImplTest extends TestBase { + private final MessageConverter messageConverter = new DefaultRqueueMessageConverter(); private final String messageId = UUID.randomUUID().toString(); - private final String queue = "test-queue"; - MessageConverter messageConverter = new DefaultRqueueMessageConverter(); - MessageHeaders messageHeaders = RqueueMessageHeaders.emptyMessageHeaders(); - @Mock private RqueueMessageTemplate messageTemplate; - @Mock private RqueueLockManager rqueueLockManager; - @Mock private RqueueMessageMetadataService rqueueMessageMetadataService; + private final String message = "Test Message"; + + + private final String queueName = "test-queue"; + private final String deadLetterQueueName = "dead-test-queue"; + private final MessageMetadata messageMetadata = MessageMetadataTestUtils.createMessageMetadata( + messageConverter, + queueName, message); + private final RqueueMessage rqueueMessage = messageMetadata.getRqueueMessage(); + private final QueueDetail queueDetail = TestUtils.createQueueDetail(queueName); + + + private final String queueName2 = "test-queue2"; + private final String priority = "high"; + private final String queueNameWithPriority = PriorityUtils.getQueueNameForPriority(queueName2, + priority); + private final QueueDetail queueDetail2 = TestUtils.createQueueDetail( + queueNameWithPriority, + Collections.singletonMap(priority, 5), 3, 15_000, ""); + + + private final MessageMetadata messageMetadata2 = MessageMetadataTestUtils.createMessageMetadata( + messageConverter, + queueNameWithPriority, message); + private final RqueueMessage rqueueMessage2 = messageMetadata2.getRqueueMessage(); + @Mock + private RqueueLockManager rqueueLockManager; + @Mock + private RqueueMessageMetadataService rqueueMessageMetadataService; + @Mock + private RqueueMessageTemplate rqueueMessageTemplate; + @Mock + private RqueueConfig rqueueConfig; + @Mock + private MessageSweeper messageSweeper; private RqueueMessageManager rqueueMessageManager; + @BeforeEach public void init() throws IllegalAccessException { MockitoAnnotations.openMocks(this); rqueueMessageManager = - new RqueueMessageManagerImpl(messageTemplate, messageConverter, messageHeaders); - FieldUtils.writeField( + new RqueueMessageManagerImpl( + rqueueMessageTemplate, messageConverter, null); + EndpointRegistry.delete(); + EndpointRegistry.register(queueDetail); + EndpointRegistry.register(queueDetail2); + writeField(rqueueMessageManager, "rqueueConfig", rqueueConfig, true); + writeField(rqueueMessageManager, "rqueueLockManager", rqueueLockManager, true); + writeField( rqueueMessageManager, "rqueueMessageMetadataService", rqueueMessageMetadataService, true); - FieldUtils.writeField(rqueueMessageManager, "rqueueLockManager", rqueueLockManager, true); + } + + + @Test + void deleteAllMessages() { + try (MockedStatic messageSweeperMockedStatic = Mockito.mockStatic( + MessageSweeper.class)) { + messageSweeperMockedStatic.when(() -> MessageSweeper.getInstance(any(), any(), any())) + .thenReturn(messageSweeper); + rqueueMessageManager.deleteAllMessages(queueName); + verify(messageSweeper, times(1)).deleteAllMessages(any()); + } + } + + @Test + void deleteAllMessagesWithPriority() { + try (MockedStatic messageSweeperMockedStatic = Mockito.mockStatic( + MessageSweeper.class)) { + messageSweeperMockedStatic.when(() -> MessageSweeper.getInstance(any(), any(), any())) + .thenReturn(messageSweeper); + rqueueMessageManager.deleteAllMessages(queueName2, priority); + verify(messageSweeper, times(1)).deleteAllMessages(any()); + } + } + + + @Test + void getAllMessages() { + doReturn(Collections.emptyList()).when(rqueueMessageTemplate) + .getAllMessages(queueDetail.getQueueName(), queueDetail.getProcessingQueueName(), + queueDetail.getScheduledQueueName()); + assertEquals(0, rqueueMessageManager.getAllMessages(queueName).size()); + doReturn(Collections.singletonList(rqueueMessage)).when(rqueueMessageTemplate) + .getAllMessages(queueDetail.getQueueName(), queueDetail.getProcessingQueueName(), + queueDetail.getScheduledQueueName()); + List messages = rqueueMessageManager.getAllMessages(queueName); + assertEquals(1, messages.size()); + assertEquals(message, messages.get(0)); + } + + @Test + void getAllMessagesWithPriority() { + doReturn(Collections.emptyList()).when(rqueueMessageTemplate) + .getAllMessages(queueDetail2.getQueueName(), queueDetail2.getProcessingQueueName(), + queueDetail2.getScheduledQueueName()); + assertEquals(0, rqueueMessageManager.getAllMessages(queueName2, priority).size()); + + doReturn(Collections.singletonList(rqueueMessage2)).when(rqueueMessageTemplate) + .getAllMessages(queueDetail2.getQueueName(), queueDetail2.getProcessingQueueName(), + queueDetail2.getScheduledQueueName()); + List messages = rqueueMessageManager.getAllMessages(queueName2, priority); + assertEquals(1, messages.size()); + assertEquals(message, messages.get(0)); } @Test - void getMessageDoesNotExist() { - assertNull(rqueueMessageManager.getMessage(queue, messageId)); + void getRqueueMessage() { + assertNull(rqueueMessageManager.getRqueueMessage(queueName, messageId)); verify(rqueueMessageMetadataService, times(1)).getByMessageId(anyString(), anyString()); + + doReturn(messageMetadata) + .when(rqueueMessageMetadataService) + .getByMessageId(queueName, messageId); + assertEquals(rqueueMessage, rqueueMessageManager.getRqueueMessage(queueName, messageId)); + } + + @Test + void getRqueueMessageWithPriority() { + assertNull(rqueueMessageManager.getRqueueMessage(queueName2, priority, messageId)); + verify(rqueueMessageMetadataService, times(1)).getByMessageId(anyString(), anyString()); + + doReturn(messageMetadata2) + .when(rqueueMessageMetadataService) + .getByMessageId(queueNameWithPriority, messageId); + assertEquals(rqueueMessage2, + rqueueMessageManager.getRqueueMessage(queueName2, priority, messageId)); + } + + + @Test + void getAllRqueueMessage() { + assertEquals(0, rqueueMessageManager.getAllRqueueMessage(queueName).size()); + verify(rqueueMessageTemplate, times(1)).getAllMessages(queueDetail.getQueueName(), + queueDetail.getProcessingQueueName(), queueDetail.getScheduledQueueName()); + + doReturn(Collections.singletonList(rqueueMessage)) + .when(rqueueMessageTemplate).getAllMessages(queueDetail.getQueueName(), + queueDetail.getProcessingQueueName(), queueDetail.getScheduledQueueName()); + List messages = rqueueMessageManager.getAllRqueueMessage(queueName); + assertEquals(1, messages.size()); + assertEquals(rqueueMessage, messages.get(0)); } @Test - void getRqueueMessageDoesNotExist() { - assertNull(rqueueMessageManager.getRqueueMessage(queue, messageId)); + void getAllRqueueMessageWithPriority() { + assertEquals(0, rqueueMessageManager.getAllRqueueMessage(queueName2, priority).size()); + verify(rqueueMessageTemplate, times(1)).getAllMessages(queueDetail2.getQueueName(), + queueDetail2.getProcessingQueueName(), queueDetail2.getScheduledQueueName()); + + doReturn(Collections.singletonList(rqueueMessage)) + .when(rqueueMessageTemplate).getAllMessages(queueDetail2.getQueueName(), + queueDetail2.getProcessingQueueName(), queueDetail2.getScheduledQueueName()); + List messages = rqueueMessageManager.getAllRqueueMessage(queueName2, priority); + assertEquals(1, messages.size()); + assertEquals(rqueueMessage, messages.get(0)); + } + + + @Test + void getMessage() { + assertNull(rqueueMessageManager.getMessage(queueName, messageId)); verify(rqueueMessageMetadataService, times(1)).getByMessageId(anyString(), anyString()); + + doReturn(messageMetadata) + .when(rqueueMessageMetadataService) + .getByMessageId(queueName, messageId); + assertNotNull(rqueueMessageManager.getMessage(queueName, messageId)); } @Test - void getMessageExist() { - doReturn(TestUtils.createMessageMetadata(messageConverter, queue)) + void getMessageWithPriority() { + assertNull(rqueueMessageManager.getMessage(queueName2, priority, messageId)); + verify(rqueueMessageMetadataService, times(1)).getByMessageId(anyString(), anyString()); + assertNull(rqueueMessageManager.getRqueueMessage(queueName2, priority, messageId)); + verify(rqueueMessageMetadataService, times(2)).getByMessageId(anyString(), anyString()); + doReturn(messageMetadata2) .when(rqueueMessageMetadataService) - .getByMessageId(queue, messageId); - assertNotNull(rqueueMessageManager.getMessage(queue, messageId)); + .getByMessageId(queueNameWithPriority, messageId); + assertNotNull(rqueueMessageManager.getMessage(queueName2, priority, messageId)); } @Test - void existLockCanNotBeAcquired() { - doReturn(false).when(rqueueLockManager).acquireLock(anyString(), anyString(), any()); - assertThrows(LockCanNotBeAcquired.class, () -> rqueueMessageManager.exist(queue, messageId)); + void exist() { + doReturn(messageMetadata) + .when(rqueueMessageMetadataService) + .getByMessageId(queueName, messageId); + assertTrue(rqueueMessageManager.exist(queueName, messageId)); + + doReturn(null) + .when(rqueueMessageMetadataService) + .getByMessageId(queueName, messageId); + assertFalse(rqueueMessageManager.exist(queueName, messageId)); } @Test - void existLockAcquiredAndMessageExist() { - doReturn(true).when(rqueueLockManager).acquireLock(anyString(), anyString(), any()); - doReturn(TestUtils.createMessageMetadata(messageConverter, queue)) + void existWithPriority() { + // entry wont exist + assertFalse(rqueueMessageManager.exist(queueName2, priority, messageId)); + doReturn(messageMetadata2) .when(rqueueMessageMetadataService) - .getByMessageId(queue, messageId); - assertTrue(rqueueMessageManager.exist(queue, messageId)); + .getByMessageId(queueNameWithPriority, messageId); + assertTrue(rqueueMessageManager.exist(queueName2, priority, messageId)); + } + + + @Test + void deleteMessage() { + assertFalse(rqueueMessageManager.deleteMessage(queueName, messageId)); + + doReturn(Duration.ofSeconds(500)).when(rqueueConfig).getMessageDurability(0L); + doReturn(messageMetadata).when(rqueueMessageMetadataService) + .getByMessageId(queueName, messageId); + assertFalse(rqueueMessageManager.deleteMessage(queueName, messageId)); + + doReturn(true).when(rqueueMessageMetadataService) + .deleteMessage(queueName, messageId, Duration.ofSeconds(500)); + assertTrue(rqueueMessageManager.deleteMessage(queueName, messageId)); + } + + @Test + void deleteMessageWithPriority() { + + assertFalse(rqueueMessageManager.deleteMessage(queueName2, messageId)); + + doReturn(Duration.ofSeconds(500)).when(rqueueConfig).getMessageDurability(0L); + doReturn(messageMetadata2).when(rqueueMessageMetadataService) + .getByMessageId(queueNameWithPriority, messageId); + assertFalse(rqueueMessageManager.deleteMessage(queueName2, priority, messageId)); + + doReturn(true).when(rqueueMessageMetadataService) + .deleteMessage(queueNameWithPriority, messageId, Duration.ofSeconds(500)); + assertTrue(rqueueMessageManager.deleteMessage(queueName2, priority, messageId)); + } + + @Test + void getMessageConverter() { + assertEquals(messageConverter.hashCode(), + rqueueMessageManager.getMessageConverter().hashCode()); + } + + + @Test + void moveMessageFromQueueExceptions() { + assertThrows(IllegalArgumentException.class, + () -> rqueueMessageManager.moveMessageFromDeadLetterToQueue(null, queueName, null)); + assertThrows(IllegalArgumentException.class, + () -> rqueueMessageManager.moveMessageFromDeadLetterToQueue(deadLetterQueueName, null, + null)); + doReturn(new MessageMoveResult(10, true)) + .when(rqueueMessageTemplate) + .moveMessageListToList(anyString(), anyString(), anyInt()); + rqueueMessageManager.moveMessageFromDeadLetterToQueue(deadLetterQueueName, queueName, null); + } + + @Test + void moveMessageFromDeadLetterToQueueDefaultSize() { + doReturn(new MessageMoveResult(100, true)) + .when(rqueueMessageTemplate) + .moveMessageListToList(anyString(), anyString(), anyInt()); + rqueueMessageManager.moveMessageFromDeadLetterToQueue(deadLetterQueueName, queueName, null); + verify(rqueueMessageTemplate, times(1)) + .moveMessageListToList(eq(deadLetterQueueName), eq(queueName), anyInt()); + verify(rqueueMessageTemplate, times(1)) + .moveMessageListToList(deadLetterQueueName, queueName, 100); } @Test - void existLockAcquiredAndMessageDoesNotExist() { - doReturn(true).when(rqueueLockManager).acquireLock(anyString(), anyString(), any()); - assertFalse(rqueueMessageManager.exist(queue, messageId)); + void moveMessageFromDeadLetterToQueueFixedSize() { + doReturn(new MessageMoveResult(10, true)) + .when(rqueueMessageTemplate) + .moveMessageListToList(anyString(), anyString(), anyInt()); + rqueueMessageManager.moveMessageFromDeadLetterToQueue(deadLetterQueueName, queueName, 10); + verify(rqueueMessageTemplate, times(1)) + .moveMessageListToList(eq(deadLetterQueueName), eq(queueName), anyInt()); + verify(rqueueMessageTemplate, times(1)) + .moveMessageListToList(eq(deadLetterQueueName), eq(queueName), eq(10)); } } diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/listener/JobImplTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/listener/JobImplTest.java index ab1c58ed..1e8add28 100644 --- a/rqueue-core/src/test/java/com/github/sonus21/rqueue/listener/JobImplTest.java +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/listener/JobImplTest.java @@ -1,21 +1,38 @@ /* - * Copyright 2022 Sonu Kumar + * Copyright (c) 2021-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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 com.github.sonus21.rqueue.listener; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + import com.github.sonus21.TestBase; import com.github.sonus21.rqueue.CoreUnitTest; import com.github.sonus21.rqueue.common.RqueueLockManager; @@ -27,27 +44,22 @@ import com.github.sonus21.rqueue.dao.RqueueJobDao; import com.github.sonus21.rqueue.models.db.Execution; import com.github.sonus21.rqueue.models.db.MessageMetadata; +import com.github.sonus21.rqueue.models.db.RqueueJob; import com.github.sonus21.rqueue.models.enums.ExecutionStatus; import com.github.sonus21.rqueue.models.enums.JobStatus; import com.github.sonus21.rqueue.models.enums.MessageStatus; import com.github.sonus21.rqueue.utils.TestUtils; import com.github.sonus21.rqueue.web.service.RqueueMessageMetadataService; +import java.time.Duration; import org.apache.commons.lang3.reflect.FieldUtils; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.Mock; import org.mockito.MockitoAnnotations; -import org.mockito.invocation.InvocationOnMock; import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; -import org.mockito.stubbing.Answer; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.messaging.converter.MessageConverter; -import java.time.Duration; - -import static org.junit.jupiter.api.Assertions.*; -import static org.mockito.ArgumentMatchers.*; -import static org.mockito.Mockito.*; @CoreUnitTest @MockitoSettings(strictness = Strictness.LENIENT) @@ -60,16 +72,11 @@ class JobImplTest extends TestBase { private final MessageMetadata messageMetadata = new MessageMetadata(rqueueMessage, MessageStatus.PROCESSING); private final Object userMessage = "Test Object"; - @Mock - private RedisConnectionFactory redisConnectionFactory; - @Mock - private RqueueMessageMetadataService messageMetadataService; - @Mock - private RqueueJobDao rqueueJobDao; - @Mock - private RqueueMessageTemplate rqueueMessageTemplate; - @Mock - private RqueueLockManager rqueueLockManager; + @Mock private RedisConnectionFactory redisConnectionFactory; + @Mock private RqueueMessageMetadataService messageMetadataService; + @Mock private RqueueJobDao rqueueJobDao; + @Mock private RqueueMessageTemplate rqueueMessageTemplate; + @Mock private RqueueLockManager rqueueLockManager; private RqueueConfig rqueueConfig; @BeforeEach @@ -188,8 +195,8 @@ void updateMessageStatus() { job.updateMessageStatus(MessageStatus.PROCESSING); assertEquals(MessageStatus.PROCESSING, job.getMessageMetadata().getStatus()); assertEquals(JobStatus.PROCESSING, job.getStatus()); - verify(rqueueJobDao, times(1)).createJob(any(), any()); - verify(messageMetadataService, times(1)).save(any(), any()); + verify(rqueueJobDao, times(1)).createJob(any(RqueueJob.class), any(Duration.class)); + verify(messageMetadataService, times(1)).save(any(MessageMetadata.class), any(Duration.class), anyBoolean()); verify(rqueueJobDao, times(1)).save(any(), any()); } @@ -303,22 +310,25 @@ void testMessageMetadataIsDeleted() throws IllegalAccessException { @Test void testMessageWasDeletedWhileRunning() throws IllegalAccessException { doReturn(true).when(rqueueLockManager).acquireLock(anyString(), any(), any()); - MessageMetadata metadata = messageMetadata.toBuilder().deleted(true) - .status(MessageStatus.DELETED).build(); + MessageMetadata metadata = + messageMetadata.toBuilder().deleted(true).status(MessageStatus.DELETED).build(); doReturn(metadata).when(messageMetadataService).get(messageMetadata.getId()); JobImpl job = instance(); job.execute(); job.updateMessageStatus(MessageStatus.FAILED); verify(rqueueJobDao, times(1)).createJob(any(), any()); verify(rqueueJobDao, times(2)).save(any(), any()); - doAnswer(invocation -> { - MessageMetadata messageMetadata = invocation.getArgument(0); - assertTrue(messageMetadata.isDeleted()); - assertEquals(MessageStatus.DELETED, messageMetadata.getStatus()); - return null; - }).when(messageMetadataService).save( - any(MessageMetadata.class), - eq(Duration.ofMinutes(rqueueConfig.getMessageDurabilityInMinute()))); - + doAnswer( + invocation -> { + MessageMetadata messageMetadata = invocation.getArgument(0); + assertTrue(messageMetadata.isDeleted()); + assertEquals(MessageStatus.DELETED, messageMetadata.getStatus()); + return null; + }) + .when(messageMetadataService) + .save( + any(MessageMetadata.class), + eq(Duration.ofMinutes(rqueueConfig.getMessageDurabilityInMinute())), + eq(false)); } } diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/utils/MessageMetadataTestUtils.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/utils/MessageMetadataTestUtils.java new file mode 100644 index 00000000..f77ae607 --- /dev/null +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/utils/MessageMetadataTestUtils.java @@ -0,0 +1,45 @@ +/* + * Copyright (c) 2023 Sonu Kumar + * + * Licensed 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 + * + * https://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 com.github.sonus21.rqueue.utils; + + +import com.github.sonus21.rqueue.core.RqueueMessage; +import com.github.sonus21.rqueue.core.support.RqueueMessageUtils; +import com.github.sonus21.rqueue.models.db.MessageMetadata; +import com.github.sonus21.rqueue.models.enums.MessageStatus; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; +import org.springframework.messaging.converter.MessageConverter; + +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public class MessageMetadataTestUtils { + + public static MessageMetadata createMessageMetadata( + MessageConverter messageConverter, String queue) { + return new MessageMetadata(RqueueMessageTestUtils.createMessage(messageConverter, queue), + MessageStatus.ENQUEUED); + } + + public static MessageMetadata createMessageMetadata( + MessageConverter messageConverter, String queue, Object message) { + RqueueMessage rqueueMessage = RqueueMessageUtils.generateMessages( + messageConverter, message, queue, null, null, 1) + .get(0); + return new MessageMetadata(rqueueMessage, MessageStatus.ENQUEUED); + } + +} diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/utils/RqueueMessageTestUtils.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/utils/RqueueMessageTestUtils.java new file mode 100644 index 00000000..d4256677 --- /dev/null +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/utils/RqueueMessageTestUtils.java @@ -0,0 +1,37 @@ +/* + * Copyright (c) 2023 Sonu Kumar + * + * Licensed 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 + * + * https://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 com.github.sonus21.rqueue.utils; + +import com.github.sonus21.rqueue.core.DefaultRqueueMessageConverter; +import com.github.sonus21.rqueue.core.RqueueMessage; +import com.github.sonus21.rqueue.core.support.RqueueMessageUtils; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; +import org.springframework.messaging.converter.MessageConverter; + +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public class RqueueMessageTestUtils { + + public static RqueueMessage createMessage(String queueName) { + return createMessage(new DefaultRqueueMessageConverter(), queueName); + } + + + public static RqueueMessage createMessage(MessageConverter messageConverter, String queue) { + return RqueueMessageUtils.generateMessage(messageConverter, queue); + } +} diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/web/service/RqueueMessageMetadataServiceTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/web/service/RqueueMessageMetadataServiceTest.java index d8ba905e..0384a596 100644 --- a/rqueue-core/src/test/java/com/github/sonus21/rqueue/web/service/RqueueMessageMetadataServiceTest.java +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/web/service/RqueueMessageMetadataServiceTest.java @@ -1,44 +1,55 @@ /* - * Copyright 2021 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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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 com.github.sonus21.rqueue.web.service; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.verifyNoInteractions; + import com.github.sonus21.TestBase; import com.github.sonus21.rqueue.CoreUnitTest; import com.github.sonus21.rqueue.common.RqueueLockManager; +import com.github.sonus21.rqueue.config.RqueueConfig; import com.github.sonus21.rqueue.core.support.RqueueMessageUtils; import com.github.sonus21.rqueue.dao.RqueueMessageMetadataDao; import com.github.sonus21.rqueue.dao.RqueueStringDao; import com.github.sonus21.rqueue.models.db.MessageMetadata; import com.github.sonus21.rqueue.models.enums.MessageStatus; +import com.github.sonus21.rqueue.utils.Constants; import com.github.sonus21.rqueue.web.service.impl.RqueueMessageMetadataServiceImpl; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.mockito.Mock; -import org.mockito.MockitoAnnotations; - import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.UUID; - -import static org.junit.jupiter.api.Assertions.*; -import static org.mockito.ArgumentMatchers.*; -import static org.mockito.Mockito.*; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; @CoreUnitTest class RqueueMessageMetadataServiceTest extends TestBase { @@ -47,6 +58,7 @@ class RqueueMessageMetadataServiceTest extends TestBase { @Mock private RqueueMessageMetadataDao rqueueMessageMetadataDao; @Mock private RqueueStringDao rqueueStringDao; @Mock private RqueueLockManager lockManager; + @Mock private RqueueConfig rqueueConfig; private RqueueMessageMetadataService rqueueMessageMetadataService; @BeforeEach @@ -54,7 +66,7 @@ public void init() { MockitoAnnotations.openMocks(this); rqueueMessageMetadataService = new RqueueMessageMetadataServiceImpl( - rqueueMessageMetadataDao, rqueueStringDao, lockManager); + rqueueMessageMetadataDao, rqueueStringDao, lockManager, rqueueConfig); } @Test @@ -82,8 +94,14 @@ void findAll() { @Test void deleteMessageShouldCreateMessageMetadata() { + doAnswer((Answer) invocationOnMock -> invocationOnMock.getArgument(0)) + .when(rqueueConfig) + .getLockKey(anyString()); String id = UUID.randomUUID().toString(); - doReturn(true).when(lockManager).acquireLock(eq(id), anyString(), eq(Duration.ofSeconds(1))); + doReturn(true) + .when(lockManager) + .acquireLock( + eq(Constants.MESSAGE_LOCK_KEY_PREFIX + id), anyString(), eq(Duration.ofSeconds(1))); doAnswer( invocation -> { MessageMetadata metadata = invocation.getArgument(0); @@ -92,14 +110,20 @@ void deleteMessageShouldCreateMessageMetadata() { return null; }) .when(rqueueMessageMetadataDao) - .save(any(), eq(Duration.ofDays(7))); + .save(any(), eq(Duration.ofDays(7)), eq(false)); assertTrue(rqueueMessageMetadataService.deleteMessage(queueName, id, Duration.ofDays(7))); } @Test void deleteMessage() { String id = UUID.randomUUID().toString(); - doReturn(true).when(lockManager).acquireLock(eq(id), anyString(), eq(Duration.ofSeconds(1))); + doAnswer((Answer) invocationOnMock -> invocationOnMock.getArgument(0)) + .when(rqueueConfig) + .getLockKey(anyString()); + doReturn(true) + .when(lockManager) + .acquireLock( + eq(Constants.MESSAGE_LOCK_KEY_PREFIX + id), anyString(), eq(Duration.ofSeconds(1))); MessageMetadata metadata = new MessageMetadata( RqueueMessageUtils.getMessageMetaId(queueName, id), MessageStatus.ENQUEUED); @@ -115,14 +139,20 @@ void deleteMessage() { return null; }) .when(rqueueMessageMetadataDao) - .save(any(), eq(Duration.ofDays(7))); + .save(any(), eq(Duration.ofDays(7)), eq(false)); assertTrue(rqueueMessageMetadataService.deleteMessage(queueName, id, Duration.ofDays(7))); } @Test void deleteMessageShouldFailDueToLock() { String id = UUID.randomUUID().toString(); - doReturn(false).when(lockManager).acquireLock(eq(id), anyString(), eq(Duration.ofSeconds(1))); + doAnswer((Answer) invocationOnMock -> invocationOnMock.getArgument(0)) + .when(rqueueConfig) + .getLockKey(anyString()); + doReturn(false) + .when(lockManager) + .acquireLock( + eq(Constants.MESSAGE_LOCK_KEY_PREFIX + id), anyString(), eq(Duration.ofSeconds(1))); assertFalse(rqueueMessageMetadataService.deleteMessage(queueName, id, Duration.ofDays(7))); verifyNoInteractions(rqueueMessageMetadataDao); } diff --git a/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/MessageDeduplicationTest.java b/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/MessageDeduplicationTest.java index 0aae722d..58589dbe 100644 --- a/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/MessageDeduplicationTest.java +++ b/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/MessageDeduplicationTest.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -27,10 +27,12 @@ import com.github.sonus21.rqueue.test.dto.Email; import com.github.sonus21.rqueue.test.dto.Notification; import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.TestPropertySource; +import org.springframework.util.Assert; @SpringBootTest @ContextConfiguration(classes = Application.class) @@ -40,6 +42,7 @@ "rqueue.retry.per.poll=20", "rqueue.scheduler.auto.start=true", "spring.redis.port=8009", + "use.system.redis=false", "mysql.db.name=MessageDeduplicationTest", "rqueue.metrics.count.failure=false", "rqueue.metrics.count.execution=false", @@ -57,16 +60,18 @@ void enqueueUnique() throws TimedOutException { @Test void enqueueUniqueIn() throws TimedOutException { Notification notification = Notification.newInstance(); - rqueueMessageEnqueuer.enqueueUniqueIn( - notificationQueue, notification.getId(), notification, 1000L); + Assertions.assertTrue( + rqueueMessageEnqueuer.enqueueUniqueIn( + notificationQueue, notification.getId(), notification, 1000L)); Notification newNotification = Notification.newInstance(); newNotification.setId(notification.getId()); sleep(100); - rqueueMessageEnqueuer.enqueueUniqueIn( - notificationQueue, newNotification.getId(), newNotification, 1000L); - waitFor(() -> getMessageCount(notificationQueue) == 0, 30_000, "notification to be sent"); + Assertions.assertFalse( + rqueueMessageEnqueuer.enqueueUniqueIn( + notificationQueue, newNotification.getId(), newNotification, 1000L)); + waitFor(() -> getMessageCount(notificationQueue) == 0, 60_000, "notification to be sent"); Notification notificationFromDb = - consumedMessageStore.getMessage(newNotification.getId(), Notification.class); - assertEquals(newNotification, notificationFromDb); + consumedMessageStore.getMessage(notification.getId(), Notification.class); + assertEquals(notification, notificationFromDb); } } diff --git a/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/MessageProcessorTest.java b/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/MessageProcessorTest.java index 48b97252..53ea1676 100644 --- a/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/MessageProcessorTest.java +++ b/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/MessageProcessorTest.java @@ -1,16 +1,16 @@ /* - * Copyright 2022 Sonu Kumar + * Copyright (c) 2021-2023 Sonu Kumar * - * Licensed 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 + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -18,6 +18,7 @@ import static com.github.sonus21.rqueue.utils.TimeoutUtils.waitFor; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import com.github.sonus21.rqueue.exception.TimedOutException; import com.github.sonus21.rqueue.spring.boot.application.ApplicationWithMessageProcessor; @@ -29,12 +30,12 @@ import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import org.junitpioneer.jupiter.RetryingTest; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.TestPropertySource; +import org.springframework.util.Assert; @ContextConfiguration(classes = ApplicationWithMessageProcessor.class) @SpringBootTest @@ -113,19 +114,21 @@ void simpleMessageExecution() throws TimedOutException { assertEquals(0, discardMessageProcessor.count()); } - @RetryingTest(2) + @Test void manualDeletionMessageProcessorTest() throws TimedOutException { cleanQueue(notificationQueue); + deleteTestData(); Notification notification = Notification.newInstance(); failureManager.createFailureDetail(notification.getId(), -1, 3); String messageId = enqueueAtGetMessageId(notificationQueue, notification, System.currentTimeMillis() + 1000L); - rqueueMessageManager.deleteMessage(notificationQueue, messageId); + assertTrue(rqueueMessageManager.deleteMessage(notificationQueue, messageId)); waitFor( () -> { List messages = getAllMessages(notificationQueue); return !messages.contains(notification); }, + 30_000L, "message to be ignored"); assertEquals(1, preExecutionMessageProcessor.count()); assertEquals(1, manualDeletionMessageProcessor.count()); diff --git a/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/RqueueMessageTemplateTest.java b/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/RqueueMessageTemplateTest.java index 615c0f30..626d2325 100644 --- a/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/RqueueMessageTemplateTest.java +++ b/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/RqueueMessageTemplateTest.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2020-2023 Sonu Kumar * - * Licensed 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 + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -47,7 +47,8 @@ class RqueueMessageTemplateTest extends SpringTestBase { @Test void moveMessageFromDeadLetterQueueToOriginalQueue() { enqueue(emailDeadLetterQueue, i -> Email.newInstance(), 10, true); - rqueueMessageSender.moveMessageFromDeadLetterToQueue(emailDeadLetterQueue, emailQueue); + assertTrue( + rqueueMessageManager.moveMessageFromDeadLetterToQueue(emailDeadLetterQueue, emailQueue)); assertEquals(10, stringRqueueRedisTemplate.getListSize(emailQueue).intValue()); assertEquals(0, stringRqueueRedisTemplate.getListSize(emailDeadLetterQueue).intValue()); } diff --git a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/application/ApplicationWithMessageProcessor.java b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/application/ApplicationWithMessageProcessor.java index 836ea7a0..9033b0a4 100644 --- a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/application/ApplicationWithMessageProcessor.java +++ b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/application/ApplicationWithMessageProcessor.java @@ -1,16 +1,16 @@ /* - * Copyright 2022 Sonu Kumar + * Copyright (c) 2021-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -30,12 +30,12 @@ public TestMessageProcessor deadLetterQueueMessageProcessor() { @Bean public TestMessageProcessor preExecutionMessageProcessor() { - return new TestMessageProcessor("PreEx"); + return new TestMessageProcessor("Pre"); } @Bean public TestMessageProcessor postExecutionMessageProcessor() { - return new TestMessageProcessor("PostEx"); + return new TestMessageProcessor("Post"); } @Bean @@ -51,7 +51,7 @@ public TestMessageProcessor discardMessageProcessor() { @Bean public SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory( @Qualifier("preExecutionMessageProcessor") TestMessageProcessor preExecutionMessageProcessor, - @Qualifier("postExecutionMessageProcessor") TestMessageProcessor postExecutionMessageProcessor, + @Qualifier("postExecutionMessageProcessor") TestMessageProcessor postExecutionMessageProcessor, @Qualifier("manualDeletionMessageProcessor") TestMessageProcessor manualDeletionMessageProcessor, @Qualifier("discardMessageProcessor") TestMessageProcessor discardMessageProcessor, @Qualifier("deadLetterQueueMessageProcessor") TestMessageProcessor deadLetterQueueMessageProcessor) { diff --git a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/common/SpringTestBase.java b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/common/SpringTestBase.java index ff46ad70..4c6ab889 100644 --- a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/common/SpringTestBase.java +++ b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/common/SpringTestBase.java @@ -1,16 +1,16 @@ /* - * Copyright 2022 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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -27,7 +27,6 @@ import com.github.sonus21.rqueue.core.RqueueMessage; import com.github.sonus21.rqueue.core.RqueueMessageEnqueuer; import com.github.sonus21.rqueue.core.RqueueMessageManager; -import com.github.sonus21.rqueue.core.RqueueMessageSender; import com.github.sonus21.rqueue.core.RqueueMessageTemplate; import com.github.sonus21.rqueue.core.support.RqueueMessageUtils; import com.github.sonus21.rqueue.dao.RqueueJobDao; @@ -57,11 +56,8 @@ @Slf4j public abstract class SpringTestBase extends TestBase { - - @Autowired protected RqueueMessageSender rqueueMessageSender; @Autowired protected RqueueMessageTemplate rqueueMessageTemplate; @Autowired protected RqueueConfig rqueueConfig; - @Autowired protected RqueueWebConfig rqueueWebConfig; @Autowired protected RqueueRedisTemplate stringRqueueRedisTemplate; @Autowired protected ConsumedMessageStore consumedMessageStore; @Autowired protected RqueueMessageListenerContainer rqueueMessageListenerContainer; @@ -258,13 +254,15 @@ protected void cleanQueue(String queue) { } } + protected void deleteTestData() { + consumedMessageStore.deleteAll(); + failureManager.deleteAll(); + } + protected boolean enqueue(String queueName, Object message) { if (reactiveEnabled) { return reactiveRqueueMessageEnqueuer.enqueue(queueName, message).block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueue(queueName, message); - } return rqueueMessageEnqueuer.enqueue(queueName, message) != null; } @@ -272,9 +270,6 @@ protected boolean enqueueAt(String queueName, Object message, Date instant) { if (reactiveEnabled) { return reactiveRqueueMessageEnqueuer.enqueueAt(queueName, message, instant).block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueAt(queueName, message, instant); - } return rqueueMessageEnqueuer.enqueueAt(queueName, message, instant) != null; } @@ -282,9 +277,6 @@ protected boolean enqueueAt(String queueName, Object message, Instant instant) { if (reactiveEnabled) { return reactiveRqueueMessageEnqueuer.enqueueAt(queueName, message, instant).block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueAt(queueName, message, instant); - } return rqueueMessageEnqueuer.enqueueAt(queueName, message, instant) != null; } @@ -292,9 +284,6 @@ protected boolean enqueueAt(String queueName, Object message, long delay) { if (reactiveEnabled) { return reactiveRqueueMessageEnqueuer.enqueueAt(queueName, message, delay).block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueAt(queueName, message, delay); - } return rqueueMessageEnqueuer.enqueueAt(queueName, message, delay) != null; } @@ -309,9 +298,6 @@ protected boolean enqueueIn(String queueName, Object message, long delay) { if (reactiveEnabled) { return reactiveRqueueMessageEnqueuer.enqueueIn(queueName, message, delay).block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueIn(queueName, message, delay); - } return rqueueMessageEnqueuer.enqueueIn(queueName, message, delay) != null; } @@ -320,9 +306,6 @@ protected boolean enqueueIn(String queueName, Object message, long delay, TimeUn return reactiveRqueueMessageEnqueuer.enqueueIn(queueName, message, delay, timeUnit).block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueIn(queueName, message, delay, timeUnit); - } return rqueueMessageEnqueuer.enqueueIn(queueName, message, delay, timeUnit) != null; } @@ -330,9 +313,6 @@ protected boolean enqueueIn(String queueName, Object message, Duration duration) if (reactiveEnabled) { return reactiveRqueueMessageEnqueuer.enqueueIn(queueName, message, duration).block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueIn(queueName, message, duration); - } return rqueueMessageEnqueuer.enqueueIn(queueName, message, duration) != null; } @@ -341,9 +321,6 @@ protected boolean enqueueWithPriority(String queueName, String priority, Object return reactiveRqueueMessageEnqueuer.enqueueWithPriority(queueName, priority, message).block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueWithPriority(queueName, priority, message); - } return rqueueMessageEnqueuer.enqueueWithPriority(queueName, priority, message) != null; } @@ -355,9 +332,6 @@ protected boolean enqueueInWithPriority( .block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueInWithPriority(queueName, priority, message, delay); - } return rqueueMessageEnqueuer.enqueueInWithPriority(queueName, priority, message, delay) != null; } @@ -369,9 +343,6 @@ protected boolean enqueueInWithPriority( .block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueInWithPriority(queueName, priority, message, delay, unit); - } return rqueueMessageEnqueuer.enqueueInWithPriority(queueName, priority, message, delay, unit) != null; } @@ -384,9 +355,6 @@ protected boolean enqueueInWithPriority( .block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueInWithPriority(queueName, priority, message, duration); - } return rqueueMessageEnqueuer.enqueueInWithPriority(queueName, priority, message, duration) != null; } @@ -399,9 +367,6 @@ protected boolean enqueueAtWithPriority( .block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueAtWithPriority(queueName, priority, message, date); - } return rqueueMessageEnqueuer.enqueueAtWithPriority(queueName, priority, message, date) != null; } @@ -413,9 +378,6 @@ protected boolean enqueueAtWithPriority( .block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueAtWithPriority(queueName, priority, message, date); - } return rqueueMessageEnqueuer.enqueueAtWithPriority(queueName, priority, message, date) != null; } @@ -427,9 +389,6 @@ protected boolean enqueueAtWithPriority( .block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueAtWithPriority(queueName, priority, message, instant); - } return rqueueMessageEnqueuer.enqueueAtWithPriority(queueName, priority, message, instant) != null; } @@ -439,18 +398,11 @@ protected boolean enqueueWithRetry(String queueName, Object message, int retry) return reactiveRqueueMessageEnqueuer.enqueueWithRetry(queueName, message, retry).block() != null; } - if (random.nextBoolean()) { - return rqueueMessageSender.enqueueWithRetry(queueName, message, retry); - } return rqueueMessageEnqueuer.enqueueWithRetry(queueName, message, retry) != null; } protected void registerQueue(String queue, String... priorities) { - if (random.nextBoolean()) { - rqueueMessageSender.registerQueue(queue, priorities); - } else { - rqueueEndpointManager.registerQueue(queue, priorities); - } + rqueueEndpointManager.registerQueue(queue, priorities); } protected List getProcessingMessages(String queueName) { @@ -459,16 +411,10 @@ protected List getProcessingMessages(String queueName) { } protected List getAllMessages(String queueName) { - if (random.nextBoolean()) { - return rqueueMessageSender.getAllMessages(queueName); - } return rqueueMessageManager.getAllMessages(queueName); } private void deleteAllMessageInternal(String queueName) throws TimedOutException { - if (random.nextBoolean()) { - rqueueMessageSender.deleteAllMessages(queueName); - } rqueueMessageManager.deleteAllMessages(queueName); TimeoutUtils.waitFor(() -> getMessageCount(queueName) == 0, "message deletion"); } diff --git a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/service/ConsumedMessageStore.java b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/service/ConsumedMessageStore.java index 7ac7ee49..135e2f86 100644 --- a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/service/ConsumedMessageStore.java +++ b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/service/ConsumedMessageStore.java @@ -1,16 +1,16 @@ /* - * Copyright 2022 Sonu Kumar + * Copyright (c) 2019-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -27,6 +27,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import lombok.NonNull; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -38,10 +39,12 @@ @Slf4j public class ConsumedMessageStore { - @NonNull - private final ConsumedMessageRepository consumedMessageRepository; - @NonNull - private final ObjectMapper objectMapper; + @NonNull private final ConsumedMessageRepository consumedMessageRepository; + @NonNull private final ObjectMapper objectMapper; + + public void deleteAll() { + consumedMessageRepository.deleteAll(); + } public void save(BaseQueueMessage message, Object tag, String queueName) throws JsonProcessingException { @@ -95,6 +98,10 @@ public Map getMessages(Collection messageIds, Class tC return idToMessage; } + public List getAllMessages(String messageId) { + return consumedMessageRepository.findByMessageIdIn(Collections.singletonList(messageId)); + } + public Map getMessages(Collection messageIds) { Iterable consumedMessages = consumedMessageRepository.findByMessageIdIn(messageIds); @@ -104,22 +111,29 @@ public Map getMessages(Collection messageIds) { return idToMessage; } - public List getAllMessages() { - List consumedMessages = new ArrayList<>(); - for (ConsumedMessage consumedMessage : consumedMessageRepository.findAll()) { - consumedMessages.add(consumedMessage); + public ConsumedMessage getConsumedMessage(String messageId) { + List messages = getConsumedMessages(messageId); + if (messages.isEmpty()) { + return null; } - return consumedMessages; + if (messages.size() == 1) { + return messages.get(0); + } + throw new IllegalStateException("more than one record found"); } - public ConsumedMessage getConsumedMessage(String messageId) { + public ConsumedMessage getConsumedMessage(String messageId, String tag) { List messages = getConsumedMessages(messageId); if (messages.isEmpty()) { - return null; + throw new IllegalStateException("no message found"); } + messages = messages.stream().filter(e -> tag.equals(e.getTag())).collect(Collectors.toList()); if (messages.size() == 1) { return messages.get(0); } + if (messages.isEmpty()) { + throw new IllegalStateException("no message found"); + } throw new IllegalStateException("more than one record found"); } diff --git a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/service/FailureManager.java b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/service/FailureManager.java index 39751413..9ca489d0 100644 --- a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/service/FailureManager.java +++ b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/service/FailureManager.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2019-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -22,6 +22,7 @@ import lombok.NonNull; import lombok.RequiredArgsConstructor; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.domain.Page; import org.springframework.stereotype.Service; @Service @@ -30,6 +31,10 @@ public class FailureManager { @NonNull private final FailureDetailRepository failureDetailRepository; + public void deleteAll() { + failureDetailRepository.deleteAll(); + } + public boolean shouldFail(String id) { // no entry so no fail FailureDetail failureDetail = failureDetailRepository.findById(id).orElse(null); diff --git a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/tests/RetryTests.java b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/tests/RetryTests.java index f4a621ab..343ab283 100644 --- a/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/tests/RetryTests.java +++ b/rqueue-spring-common-test/src/main/java/com/github/sonus21/rqueue/test/tests/RetryTests.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2019-2023 Sonu Kumar * - * Licensed 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 + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -41,7 +41,7 @@ protected void verifyAfterNRetryTaskIsDeletedFromProcessingQueue() throws TimedO cleanQueue(jobQueue); Job job = Job.newInstance(); failureManager.createFailureDetail(job.getId(), 3, 10); - rqueueMessageSender.put(jobQueue, job); + rqueueMessageEnqueuer.enqueue(jobQueue, job); waitFor( () -> { Job jobInDb = consumedMessageStore.getMessage(job.getId(), Job.class); @@ -61,7 +61,7 @@ protected void verifyMessageMovedToDeadLetterQueue() throws TimedOutException { Email email = Email.newInstance(); failureManager.createFailureDetail(email.getId(), -1, 0); log.debug("queue: {} msg: {}", emailQueue, email); - rqueueMessageSender.put(emailQueue, email, 1000L); + rqueueMessageEnqueuer.enqueueIn(emailQueue, email, 1000L); waitFor( () -> emailRetryCount == failureManager.getFailureCount(email.getId()), 30000000, diff --git a/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/services/FailureDetailRepositoryImpl.java b/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/services/FailureDetailRepositoryImpl.java index 79cd2051..ff71865d 100644 --- a/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/services/FailureDetailRepositoryImpl.java +++ b/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/services/FailureDetailRepositoryImpl.java @@ -1,33 +1,36 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2019-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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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 com.github.sonus21.rqueue.spring.services; +import com.github.sonus21.rqueue.test.entity.ConsumedMessage; import com.github.sonus21.rqueue.test.entity.FailureDetail; import com.github.sonus21.rqueue.test.repository.FailureDetailRepository; -import java.util.List; -import java.util.Optional; import javax.persistence.EntityManager; import javax.persistence.EntityManagerFactory; import javax.persistence.criteria.CriteriaBuilder; +import javax.persistence.criteria.CriteriaDelete; import javax.persistence.criteria.CriteriaQuery; import javax.persistence.criteria.Root; +import java.util.List; +import java.util.Optional; import lombok.AllArgsConstructor; import org.hibernate.Session; import org.hibernate.Transaction; +import org.hibernate.query.Query; import org.springframework.stereotype.Repository; import org.springframework.util.CollectionUtils; @@ -92,14 +95,25 @@ public long count() { } @Override - public void deleteById(String s) {} + public void deleteById(String s) { + } @Override - public void delete(FailureDetail entity) {} + public void delete(FailureDetail entity) { + } @Override - public void deleteAll(Iterable entities) {} + public void deleteAll(Iterable entities) { + } @Override - public void deleteAll() {} + public void deleteAll() { + EntityManager entityManager = entityManagerFactory.createEntityManager(); + Session session = entityManager.unwrap(Session.class); + CriteriaBuilder cb = session.getCriteriaBuilder(); + CriteriaDelete cr = cb.createCriteriaDelete(FailureDetail.class); + Query query = session.createQuery(cr); + query.getSingleResult(); + entityManager.close(); + } } diff --git a/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/tests/integration/SpringAppTest.java b/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/tests/integration/SpringAppTest.java index 3bc831ad..76e73cd8 100644 --- a/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/tests/integration/SpringAppTest.java +++ b/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/tests/integration/SpringAppTest.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 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. - * You may obtain a copy of the License at + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -53,20 +53,20 @@ @WebAppConfiguration @TestPropertySource( properties = { - "spring.redis.port=7004", - "mysql.db.name=SpringAppTest", - "sms.queue.active=true", - "notification.queue.active=false", - "email.queue.active=true", - "job.queue.active=true", - "use.system.redis=false", - "priority.mode=STRICT", - "reservation.queue.active=true", - "feed.generation.queue.active=true", - "chat.indexing.queue.active=true", - "provide.executor=true", - "email.queue.retry.count=-1", - "rqueue.retry.per.poll=10" + "spring.redis.port=7004", + "mysql.db.name=SpringAppTest", + "sms.queue.active=true", + "notification.queue.active=false", + "email.queue.active=true", + "job.queue.active=true", + "use.system.redis=false", + "priority.mode=STRICT", + "reservation.queue.active=true", + "feed.generation.queue.active=true", + "chat.indexing.queue.active=true", + "provide.executor=true", + "email.queue.retry.count=-1", + "rqueue.retry.per.poll=10" }) @SpringIntegrationTest class SpringAppTest extends AllQueueMode { @@ -109,7 +109,7 @@ void verifyDefaultDeadLetterQueueRetry() throws TimedOutException { Email email1 = (Email) RqueueMessageUtils.convertMessageToObject( - messages.get(0), rqueueMessageSender.getMessageConverter()); + messages.get(0), rqueueMessageManager.getMessageConverter()); assertEquals(email.getId(), email1.getId()); } @@ -129,63 +129,64 @@ void onlyPushMode() { registerQueue(emailQueue); registerQueue(smsQueue, critical, high, low); Email[] emails = - new Email[] { - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), - Email.newInstance(), + new Email[]{ + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), + Email.newInstance(), }; Sms[] sms = - new Sms[] { - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), - Sms.newInstance(), + new Sms[]{ + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), + Sms.newInstance(), }; assertTrue(enqueue(emailQueue, emails[0])); assertTrue(rqueueMessageEnqueuer.enqueue(emailQueue, emails[1].getId(), emails[1])); assertTrue(rqueueMessageEnqueuer.enqueueUnique(emailQueue, emails[2].getId(), emails[2])); + assertFalse(rqueueMessageEnqueuer.enqueueUnique(emailQueue, emails[2].getId(), emails[2])); assertNotNull(rqueueMessageEnqueuer.enqueueWithRetry(emailQueue, emails[3], 3)); assertTrue(rqueueMessageEnqueuer.enqueueWithRetry(emailQueue, emails[4].getId(), emails[4], 3)); @@ -233,6 +234,9 @@ void onlyPushMode() { assertTrue( rqueueMessageEnqueuer.enqueueUniqueIn( emailQueue, emails[18].getId(), emails[18], Constants.ONE_MILLI)); + assertFalse( + rqueueMessageEnqueuer.enqueueUniqueIn( + emailQueue, emails[18].getId(), emails[18], Constants.ONE_MILLI)); assertNotNull( rqueueMessageEnqueuer.enqueueInWithRetry(emailQueue, emails[19], 3, Constants.ONE_MILLI)); diff --git a/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/tests/unit/RqueueMessageConfigTest.java b/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/tests/unit/RqueueMessageConfigTest.java index eca53086..01e5cfc9 100644 --- a/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/tests/unit/RqueueMessageConfigTest.java +++ b/rqueue-spring/src/test/java/com/github/sonus21/rqueue/spring/tests/unit/RqueueMessageConfigTest.java @@ -1,16 +1,16 @@ /* - * Copyright 2021 Sonu Kumar + * Copyright (c) 2019-2023 Sonu Kumar * - * Licensed 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 + * Licensed 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 * - * https://www.apache.org/licenses/LICENSE-2.0 + * https://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. + * 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. * */ @@ -45,12 +45,18 @@ class RqueueMessageConfigTest extends TestBase { private final List messageConverterList = new ArrayList<>(); - @Mock RqueueMessageHandler rqueueMessageHandler; - @Mock private SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory; - @Mock private BeanFactory beanFactory; - @Mock private RqueueMessageTemplate rqueueMessageTemplate; - @Mock private RedisConnectionFactory redisConnectionFactory; - @InjectMocks private RqueueListenerConfig rqueueMessageConfig; + @Mock + RqueueMessageHandler rqueueMessageHandler; + @Mock + private SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory; + @Mock + private BeanFactory beanFactory; + @Mock + private RqueueMessageTemplate rqueueMessageTemplate; + @Mock + private RedisConnectionFactory redisConnectionFactory; + @InjectMocks + private RqueueListenerConfig rqueueMessageConfig; @BeforeEach public void init() { @@ -106,7 +112,7 @@ void rqueueMessageSenderWithMessageTemplate() throws IllegalAccessException { doReturn(new DefaultRqueueMessageConverter()).when(rqueueMessageHandler).getMessageConverter(); RqueueListenerConfig messageConfig = new RqueueListenerConfig(); FieldUtils.writeField(messageConfig, "simpleRqueueListenerContainerFactory", factory, true); - assertNotNull(messageConfig.rqueueMessageSender(rqueueMessageHandler, rqueueMessageTemplate)); + assertNotNull(messageConfig.rqueueMessageEnqueuer(rqueueMessageHandler, rqueueMessageTemplate)); assertEquals(factory.getRqueueMessageTemplate().hashCode(), rqueueMessageTemplate.hashCode()); } }