implemented support for SQS long polling with a thread per SQS Queue

Author: aaronwalker

pom.xml

@@ -62,7 +62,7 @@
         <dependency>
             <groupId>com.amazonaws</groupId>
             <artifactId>aws-java-sdk</artifactId>
-            <version>1.3.3</version>
+            <version>1.3.26</version>
         </dependency>
     </dependencies>
 

src/main/java/com/base2services/jenkins/SqsQueueHandler.java

@@ -8,9 +8,11 @@
 import com.cloudbees.jenkins.GitHubWebHook;
 import hudson.Extension;
 import hudson.model.PeriodicWork;
+import hudson.util.SequentialExecutionQueue;
 import hudson.util.TimeUnit2;
 
 import java.util.List;
+import java.util.concurrent.Executors;
 import java.util.logging.Logger;
 
 /**
@@ -23,20 +25,45 @@ public class SqsQueueHandler extends PeriodicWork {
 
     private static final Logger LOGGER = Logger.getLogger(SqsQueueHandler.class.getName());
 
+    private transient final SequentialExecutionQueue queue = new SequentialExecutionQueue(Executors.newFixedThreadPool(2));
+
     @Override
     public long getRecurrencePeriod() {
-        return TimeUnit2.SECONDS.toMillis(30);
+        return TimeUnit2.SECONDS.toMillis(2);
     }
 
     @Override
     protected void doRun() throws Exception {
-        List<SqsProfile> profiles = SqsBuildTrigger.DescriptorImpl.get().getSqsProfiles();
-        for(SqsProfile profile : profiles) {
+        if(queue.getInProgress().size() == 0) {
+            List<SqsProfile> profiles = SqsBuildTrigger.DescriptorImpl.get().getSqsProfiles();
+            queue.setExecutors(Executors.newFixedThreadPool(profiles.size()));
+            for(final SqsProfile profile : profiles) {
+                queue.execute(new SQSQueueReceiver(profile));
+            }
+        } else {
+            logger.fine("Currently Waiting for Messages from Queues");
+        }
+    }
+
+    public static SqsQueueHandler get() {
+        return PeriodicWork.all().get(SqsQueueHandler.class);
+    }
+
+    private class SQSQueueReceiver implements Runnable {
+
+        private SqsProfile profile;
+
+        private SQSQueueReceiver(SqsProfile profile) {
+            this.profile = profile;
+        }
+
+        public void run() {
             LOGGER.fine("looking for build triggers on queue:" + profile.sqsQueue);
             AmazonSQS sqs = profile.getSQSClient();
             String queueUrl = profile.getQueueUrl();
             TriggerProcessor processor = profile.getTriggerProcessor();
             ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(queueUrl);
+            receiveMessageRequest.setWaitTimeSeconds(20);
             List<Message> messages = sqs.receiveMessage(receiveMessageRequest).getMessages();
             for(Message message : messages) {
                 //Process the message payload, it needs to conform to the GitHub Web-Hook JSON format
@@ -52,8 +79,4 @@ protected void doRun() throws Exception {
             }
         }
     }
-
-    public static SqsQueueHandler get() {
-        return PeriodicWork.all().get(SqsQueueHandler.class);
-    }
 }