Custom visibility timeout for each message in SQS queue

nevado-jms/src/main/java/org/skyscreamer/nevado/jms/connector/AbstractSQSConnector.java

@@ -29,26 +29,37 @@
  */
 public abstract class AbstractSQSConnector implements SQSConnector {
     protected static final String AWS_ERROR_CODE_AUTHENTICATION = "InvalidClientTokenId";
-
+   
     protected final Log _log = LogFactory.getLog(getClass());
 
     private final long _receiveCheckIntervalMs;
     private final boolean _isAsync;
+    private final int _visibilityTimeoutOnReset;
 
     protected AbstractSQSConnector(long receiveCheckIntervalMs)
     {
         this(receiveCheckIntervalMs, false);
     }
 
-    protected AbstractSQSConnector(long receiveCheckIntervalMs, boolean isAsync)
+    protected AbstractSQSConnector(long receiveCheckIntervalMs, boolean isAsync, int visibilityTimeoutOnReset)
     {
         _receiveCheckIntervalMs = receiveCheckIntervalMs;
         _isAsync = isAsync;
+        _visibilityTimeoutOnReset = visibilityTimeoutOnReset;
+    }
+
+    protected AbstractSQSConnector(long receiveCheckIntervalMs, boolean isAsync)
+    {
+    	this(receiveCheckIntervalMs, isAsync, 0);
     }
 
     public boolean isAsync() {
         return _isAsync;
     }
+
+    public int getVisibilityTimeoutOnReset() {
+	return _visibilityTimeoutOnReset;
+    }
 
     public void sendMessage(NevadoDestination destination, NevadoMessage message) throws JMSException
     {
@@ -127,7 +138,7 @@ public void resetMessage(NevadoMessage message) throws JMSException {
                     "Did this come from an SQS queue?");
         }
         SQSQueue sqsQueue = getSQSQueue(message.getNevadoDestination());
-        sqsQueue.setMessageVisibilityTimeout(sqsReceiptHandle, 0);
+        sqsQueue.setMessageVisibilityTimeout(sqsReceiptHandle, _visibilityTimeoutOnReset); // Customize message visibility timeout
     }
 
     /**
@@ -179,7 +190,9 @@ protected SQSMessage receiveSQSMessage(NevadoConnection connection, NevadoDestin
                 if (sqsMessage != null && !connection.isRunning()) {
                     // Connection was stopped while the REST call to SQS was being made
                     try {
-                        sqsQueue.setMessageVisibilityTimeout(sqsMessage.getReceiptHandle(), 0); // Make it immediately available to the next requestor
+                        if(sqsMessage.getReceiptHandle() != null && StringUtils.isNotEmpty(sqsMessage.getReceiptHandle()) && sqsMessage.getReceiptHandle().trim().length() > 0) {
+                    		sqsQueue.setMessageVisibilityTimeout(sqsMessage.getReceiptHandle(), _visibilityTimeoutOnReset); // Customize message visibility timeout
+                    	}
                     } catch (JMSException e) {
                         String exMessage = "Unable to reset visibility timeout for message: " + e.getMessage();
                         _log.warn(exMessage, e); // Non-fatal.  Just means the message will disappear until the visibility timeout expires.

nevado-jms/src/main/java/org/skyscreamer/nevado/jms/connector/amazonaws/AmazonAwsSQSConnector.java

@@ -47,16 +47,25 @@
 public class AmazonAwsSQSConnector extends AbstractSQSConnector {
     public static final String MESSAGE_ATTRIBUTE_APPROXIMATE_RECEIVE_COUNT = "ApproximateReceiveCount";
 
-    private final AmazonSQS _amazonSQS;
-    private final AmazonSNS _amazonSNS;
+    private AmazonSQS _amazonSQS;
+    private AmazonSNS _amazonSNS;
 
     public AmazonAwsSQSConnector(String awsAccessKey, String awsSecretKey, boolean isSecure, long receiveCheckIntervalMs) {
         this(awsAccessKey, awsSecretKey, isSecure, receiveCheckIntervalMs, false);
     }
 
     public AmazonAwsSQSConnector(String awsAccessKey, String awsSecretKey, boolean isSecure, long receiveCheckIntervalMs, boolean isAsync) {
         super(receiveCheckIntervalMs, isAsync);
-        AWSCredentials awsCredentials = new BasicAWSCredentials(awsAccessKey, awsSecretKey);
+        initializeConnection(awsAccessKey, awsSecretKey, isSecure, receiveCheckIntervalMs, isAsync);
+    }
+
+    public AmazonAwsSQSConnector(String awsAccessKey, String awsSecretKey, boolean isSecure, long receiveCheckIntervalMs, boolean isAsync, int visibilityTimeoutOnReset) {
+        super(receiveCheckIntervalMs, isAsync, visibilityTimeoutOnReset);
+        initializeConnection(awsAccessKey, awsSecretKey, isSecure, receiveCheckIntervalMs, isAsync);
+    }
+
+    private void initializeConnection(String awsAccessKey, String awsSecretKey, boolean isSecure, long receiveCheckIntervalMs, boolean isAsync){
+    	AWSCredentials awsCredentials = new BasicAWSCredentials(awsAccessKey, awsSecretKey);
         ClientConfiguration clientConfiguration = new ClientConfiguration();
         String proxyHost = System.getProperty("http.proxyHost");
         String proxyPort = System.getProperty("http.proxyPort");

nevado-jms/src/main/java/org/skyscreamer/nevado/jms/connector/amazonaws/AmazonAwsSQSConnectorFactory.java

@@ -10,11 +10,12 @@
  */
 public class AmazonAwsSQSConnectorFactory extends AbstractSQSConnectorFactory {
     protected boolean _useAsyncSend = false;
+    protected int _visibilityTimeoutOnReset = 0;
 
     @Override
     public AmazonAwsSQSConnector getInstance(String awsAccessKey, String awsSecretKey, String awsSQSEndpoint, String awsSNSEndpoint) {
         AmazonAwsSQSConnector amazonAwsSQSConnector = new AmazonAwsSQSConnector(awsAccessKey, awsSecretKey, _isSecure,
-                _receiveCheckIntervalMs, _useAsyncSend);
+                _receiveCheckIntervalMs, _useAsyncSend, _visibilityTimeoutOnReset);
         if (StringUtils.isNotEmpty(awsSQSEndpoint)) {
             amazonAwsSQSConnector.getAmazonSQS().setEndpoint(awsSQSEndpoint);
         }
@@ -31,4 +32,12 @@ public void setUseAsyncSend(boolean useAsyncSend) {
     public boolean isUseAsyncSend() {
         return _useAsyncSend;
     }
+
+   public int getVisibilityTimeoutOnReset() {
+	return _visibilityTimeoutOnReset;
+    }
+
+   public void setVisibilityTimeoutOnReset(int visibilityTimeoutOnReset) {
+	_visibilityTimeoutOnReset = visibilityTimeoutOnReset;
+    }
 }