@@ -0,0 +1,1347 @@
+package net.etalia.crepuscolo.queue;%0A%0Aimport org.apache.commons.logging.Log;%0Aimport org.apache.commons.logging.LogFactory;%0A%0Apublic abstract class ReadQueuePoller%3CT%3E extends Thread %7B%0A%0A%09protected Log log = LogFactory.getLog(ReadQueuePoller.class);%0A%0A%09protected ReadQueue%3CT%3E queue = null;%0A%09private int wait = 0;%0A%0A%09private volatile boolean mustStop = false;%0A%0A%09public ReadQueuePoller(ReadQueue%3CT%3E queue) %7B%0A%09%09this(queue, 0);%0A%09%7D%0A%09public ReadQueuePoller(ReadQueue%3CT%3E queue, int wait) %7B%0A%09%09this.queue = queue;%0A%09%09this.wait = wait;%0A%09%7D%0A%0A%09@Override%0A%09public void run() %7B%0A%09%09while (!isInterrupted() && !mustStop) %7B%0A%09%09%09try %7B%0A%09%09%09%09if (wait %3E 0) %7B%0A%09%09%09%09%09sleep(wait);%0A%09%09%09%09%7D%0A%09%09%09%09MessageIterator%3CT%3E iterator = queue.iterator();%0A%09%09%09%09if (iterator.hasNext()) %7B%0A%09%09%09%09%09try %7B%0A%09%09%09%09%09%09parse(iterator);%0A%09%09%09%09%09%7D catch (Throwable e) %7B%0A%09%09%09%09%09%09if (e instanceof InterruptedException) return;%0A%09%09%09%09%09%09log.warn(%22Error while parsing queue messages%22, e);%0A%09%09%09%09%09%7D%0A%09%09%09%09%7D%0A%09%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09%09interrupt();%0A%09%09%09%09return;%0A%09%09%09%7D catch (Throwable t) %7B%0A%09%09%09%09log.warn(%22Error while fecthing SQS messages, will retry soon%22, t);%0A%09%09%09%09try %7B%0A%09%09%09%09%09Thread.sleep(wait * 10);%0A%09%09%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09%09%09interrupt();%0A%09%09%09%09%09return;%0A%09%09%09%09%7D%0A%09%09%09%7D%0A%09%09%7D%0A%09%7D%0A%0A%09protected abstract void parse(MessageIterator%3CT%3E iterator);%0A%0A%09public void kill() %7B%0A%09%09this.interrupt();%0A%09%09this.mustStop  = true;%0A%09%7D%0A%0A%7D%0A

Add queue poller