INT-1471 removed custom thread pools used by inbound adapters in favor of task scheduler with RateLimitStatusTRiger, polishing the code and javadocs. . . more to come

spring-integration-twitter/pom.xml

@@ -81,5 +81,9 @@
 			<artifactId>commons-io</artifactId>
 			<version>1.4</version>
 		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-all</artifactId>
+		</dependency>
 	</dependencies>
 </project>

spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/twitter/Twitter4jDirectMessage.java

@@ -1,37 +1,39 @@
 /*
- * Copyright 2010 the original author or authors
+ * Copyright 2002-2010 the original author or authors.
  *
- *     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
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ *      http://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 org.springframework.integration.twitter.core.twitter;
 
 import org.springframework.integration.twitter.core.User;
+import org.springframework.util.Assert;
+
 import twitter4j.DirectMessage;
 
 import java.util.Date;
 
-
 /**
- * implementation of the {@link org.springframework.integration.twitter.core.DirectMessage} interface
- * that wraps, and works with, a {@link twitter4j.DirectMessage} instance.
+ * {@link twitter4j.DirectMessage} wrapper for {@link org.springframework.integration.twitter.core.DirectMessage}.
  *
  * @author Josh Long
+ * @author Oleg ZHurakousky
  * @since 2.0 
  */
 public class Twitter4jDirectMessage implements org.springframework.integration.twitter.core.DirectMessage {
 	private DirectMessage directMessage;
 
 	public Twitter4jDirectMessage(DirectMessage directMessage) {
+		Assert.notNull(directMessage, "'directMessage' must not be null");
 		this.directMessage = directMessage;
 	}
 

spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java

@@ -1,98 +1,75 @@
 /*
- * Copyright 2010 the original author or authors
+ * Copyright 2002-2010 the original author or authors.
  *
- *     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
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ *      http://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 org.springframework.integration.twitter.inbound;
 
-import org.apache.commons.lang.exception.ExceptionUtils;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ScheduledFuture;
 
 import org.springframework.context.Lifecycle;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.history.HistoryWritingMessagePostProcessor;
 import org.springframework.integration.history.TrackableComponent;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.twitter.oauth.OAuthConfiguration;
-
 import org.springframework.util.Assert;
 
-import twitter4j.RateLimitStatus;
-import twitter4j.ResponseList;
 import twitter4j.Twitter;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-
 
 /**
- * There are a lot of operations that are common to receiving the various types of messages when using the
- * Twitter API, and this class abstracts most of them for you. Implementers must take note of
- * {@link AbstractInboundTwitterEndpointSupport#runAsAPIRateLimitsPermit(AbstractInboundTwitterEndpointSupport.ApiCallback)}
- * which will invoke the instance of {@link AbstractInboundTwitterEndpointSupport.ApiCallback} when the
- * rate-limit API deems that its OK to do so. This class handles keeping tabs on that and on spacing out requests
- * as required.
- * <p/>
- * Simialarly, this class handles keeping track on the latest inbound message its received and avoiding, where
+ * Abstract class that defines common operations for receiving various types of messages when using the
+ * Twitter API. 
+ * This class also handles keeping track on the latest inbound message its received and avoiding, where
  * possible, redelivery of common messages. This functionality is enabled using the
  * {@link org.springframework.integration.context.metadata.MetadataStore} implementation
  *
  * @author Josh Long
+ * @author Oleg Zhurakousky
+ * 
  * @since 2.0
  */
-public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessageProducerSupport implements Lifecycle, TrackableComponent, Runnable {
-
+public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessageProducerSupport implements Lifecycle, TrackableComponent{
+	
     protected volatile OAuthConfiguration configuration;
     protected volatile long markerId = -1;
     protected Twitter twitter;
     private final Object markerGuard = new Object();
-    private final Object apiPermitGuard = new Object();
+    private volatile ScheduledFuture<?> twitterUpdatePollingTask;
 
     private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor();
 
-	protected Executor taskExecutor;
-	protected int poolSize = 1;
-
-    public void setPoolSize(int poolSize) {
-        this.poolSize = poolSize;
-    }
-
-    protected void checkTaskExecutor(final String threadName) {
-        if (this.taskExecutor == null) {
-            this.taskExecutor = Executors.newFixedThreadPool(this.poolSize,
-                    new ThreadFactory() {
-                        public Thread newThread(Runnable runner) {
-                            Thread thread = new Thread(runner);
-                            thread.setName(threadName);
-                            thread.setDaemon(true);
-
-                            return thread;
-                        }
-                    });
-        }
-    }
-
-    public void setConfiguration(OAuthConfiguration configuration) {
-        this.configuration = configuration;
-    }
-
     abstract protected void markLastStatusId(T statusId);
 
     abstract protected List<T> sort(List<T> rl);
+    
+    abstract Runnable getApiCallback();
+    
+    public void setConfiguration(OAuthConfiguration configuration) {
+        this.configuration = configuration;
+    }
+    
+    public void setShouldTrack(boolean shouldTrack) {
+        this.historyWritingPostProcessor.setShouldTrack(shouldTrack);
+    }
+    
+    public long getMarkerId() {
+        return markerId;
+    }
 
     @Override
     protected void onInit() {
@@ -105,28 +82,28 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessagePr
     protected void forwardAll(List<T> tResponses) {
         List<T> stats = new ArrayList<T>();
 
-        for (T t : tResponses)
-            stats.add(t);
-
-        for (T twitterResponse : sort(stats))
-            forward(twitterResponse);
+        for (T t : tResponses){
+        	stats.add(t);
+        }
+           
+        for (T twitterResponse : sort(stats)) {
+        	 forward(twitterResponse);
+        }         
     }
 
-    public long getMarkerId() {
-        return markerId;
-    }
-
-    abstract public String getComponentType();
-
     @Override
     protected void doStart() {
-        historyWritingPostProcessor.setTrackableComponent(this);
-
-        checkTaskExecutor(getClass().getName() + "-taskExecutor");
-
-        taskExecutor.execute(this);
+        historyWritingPostProcessor.setTrackableComponent(this);  
+    	RateLimitStatusTrigger trigger = new RateLimitStatusTrigger(this.twitter); 	
+    	Runnable apiCallback = this.getApiCallback();
+    	twitterUpdatePollingTask = this.getTaskScheduler().schedule(apiCallback, trigger);
     }
 
+    @Override
+    protected void doStop() {
+    	twitterUpdatePollingTask.cancel(true);
+    }
+    
     protected void forward(T status) {
         synchronized (this.markerGuard) {
             Message<T> twtMsg = MessageBuilder.withPayload(status).build();
@@ -136,107 +113,7 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessagePr
             markLastStatusId(status);
         }
     }
-
-    /**
-     * this is execu
-     */
-    public void run() {
-        try {
-            beginPolling();
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    abstract protected void beginPolling() throws Exception;
-
-    protected void forwardAll(ResponseList<T> tResponses) {
-        List<T> stats = new ArrayList<T>();
-
-        for (T t : tResponses)
-            stats.add(t);
-
-        for (T twitterResponse : sort(stats))
-            forward(twitterResponse);
-    }
-
-    @SuppressWarnings("unchecked")
-    protected void runAsAPIRateLimitsPermit(ApiCallback apiCallback)
-        throws Exception {
-        synchronized (this.apiPermitGuard) {
-            while (waitUntilPullAvailable()) {
-                if (logger.isDebugEnabled()) {
-                    logger.debug("have room to make an API request now");
-                }
-
-                apiCallback.run(this, twitter);
-            }
-        }
-    }
-
-    protected boolean handleReceivingRateLimitStatus(
-        RateLimitStatus rateLimitStatus) {
-        try {
-            int secondsUntilReset = rateLimitStatus.getSecondsUntilReset();
-            int remainingHits = rateLimitStatus.getRemainingHits();
-
-            if (remainingHits == 0) {
-                logger.debug(
-                    "rate status limit service returned 0 for the remaining hits value");
-
-                return false;
-            }
-
-            if (secondsUntilReset == 0) {
-                logger.debug(
-                    "rate status limit service returned 0 for the seconds until reset period value");
-
-                return false;
-            }
-
-            int secondsUntilWeCanPullAgain = secondsUntilReset / remainingHits;
-            long msUntilWeCanPullAgain = secondsUntilWeCanPullAgain * 1000;
-
-            logger.debug("need to Thread.sleep() " +
-                secondsUntilWeCanPullAgain +
-                " seconds until the next timeline pull. Have " + remainingHits +
-                " remaining pull this rate period. The period ends in " +
-                secondsUntilReset);
-
-            Thread.sleep(msUntilWeCanPullAgain);
-        } catch (Throwable throwable) {
-            logger.debug(
-                "encountered an error when trying to refresh the timeline: " +
-                ExceptionUtils.getFullStackTrace(throwable));
-        }
-
-        return true;
-    }
-
-    protected boolean waitUntilPullAvailable() throws Exception {
-        return this.handleReceivingRateLimitStatus(this.twitter.getRateLimitStatus());
-    }
-
     protected boolean hasMarkedStatus() {
         return markerId > -1;
     }
-
-    @Override
-    protected void doStop() {
-    }
-
-    public void setShouldTrack(boolean shouldTrack) {
-        this.historyWritingPostProcessor.setShouldTrack(shouldTrack);
-    }
-
-    /**
-     * Hook for clients to run logic when the API rate limiting lets us
-     * <p/>
-     * Simply register your callback using #runAsAPIRateLimitsPermit
-     *
-     * @param <C>
-     */
-    public static interface ApiCallback<C> {
-        void run(C t, Twitter twitter) throws Exception;
-    }
 }

spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterStatusEndpointSupport.java

@@ -1,17 +1,17 @@
 /*
- * Copyright 2010 the original author or authors
+ * Copyright 2002-2010 the original author or authors.
  *
- *     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
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ *      http://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 org.springframework.integration.twitter.inbound;
 import org.springframework.integration.twitter.core.Status;
@@ -31,8 +31,6 @@ import java.util.List;
  */
 abstract public class AbstractInboundTwitterStatusEndpointSupport extends AbstractInboundTwitterEndpointSupport<Status> {
 
-
-
 	private Comparator<Status> statusComparator = new Comparator<Status>() {
 		public int compare(Status status, Status status1) {
 			return status.getCreatedAt().compareTo(status1.getCreatedAt());
@@ -41,8 +39,9 @@ abstract public class AbstractInboundTwitterStatusEndpointSupport extends Abstra
 
 	protected List<Status> fromTwitter4jStatuses(List<twitter4j.Status> stats) {
 		List<Status> fwd = new ArrayList<Status>();
-		for (twitter4j.Status s : stats)
+		for (twitter4j.Status s : stats) {
 			fwd.add(new Twitter4jStatus(s));
+		}		
 		return fwd;
 	}
 

spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundDirectMessageEndpoint.java

@@ -1,35 +1,37 @@
 /*
- * Copyright 2010 the original author or authors
+ * Copyright 2002-2010 the original author or authors.
  *
- *     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
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ *      http://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 org.springframework.integration.twitter.inbound;
 
-import org.springframework.integration.twitter.core.DirectMessage;
-import org.springframework.integration.twitter.core.twitter.Twitter4jDirectMessage;
-import twitter4j.Paging;
-import twitter4j.Twitter;
-
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.twitter.core.DirectMessage;
+import org.springframework.integration.twitter.core.twitter.Twitter4jDirectMessage;
+
+import twitter4j.Paging;
 
 /**
  * This class handles support for receiving DMs (direct messages) using Twitter.
  *
  * @author Josh Long
+ * @author Oleg Zhurakousky
+ * @since 2.0
  */
 public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpointSupport<DirectMessage> {
 	
@@ -62,22 +64,35 @@ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpoint
 	}
 
 	@Override
-	protected void beginPolling() throws Exception {
+	Runnable getApiCallback() {
+		Runnable apiCallback = new Runnable() {	
+			@Override
+			public void run() {
+				try {
+					long sinceId = getMarkerId();
+					
+					List<twitter4j.DirectMessage> dms = !hasMarkedStatus() 
+								? twitter.getDirectMessages() 
+								: twitter.getDirectMessages(new Paging(sinceId));
 
-		this.runAsAPIRateLimitsPermit(new ApiCallback<InboundDirectMessageEndpoint>() {
-			public void run(InboundDirectMessageEndpoint t, Twitter twitter)
-					throws Exception {
-
-				List<twitter4j.DirectMessage> dms = !hasMarkedStatus() ? t.twitter.getDirectMessages() : t.twitter.getDirectMessages(new Paging(t.getMarkerId()));
-
-				List<DirectMessage> dmsToFwd = new ArrayList<DirectMessage>();
-
-				for( twitter4j.DirectMessage dm : dms)
-					dmsToFwd.add( new Twitter4jDirectMessage( dm));
-
-				forwardAll( dmsToFwd );
+					List<DirectMessage> dmsToFwd = new ArrayList<DirectMessage>();
+			
+					for( twitter4j.DirectMessage dm : dms) {
+						dmsToFwd.add( new Twitter4jDirectMessage( dm));
+					}
+					forwardAll(dmsToFwd);
+				} catch (Exception e) {
+					e.printStackTrace();
+					if (e instanceof RuntimeException){
+						throw (RuntimeException)e;
+					}
+					else {
+						throw new MessagingException("Failed to poll for Twitter mentions updates", e);
+					}
+				}
 			}
-		});
+		};
+		return apiCallback;
 	}
 
 }

spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundMentionEndpoint.java

@@ -1,30 +1,31 @@
 /*
- * Copyright 2010 the original author or authors
+ * Copyright 2002-2010 the original author or authors.
  *
- *     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
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ *      http://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 org.springframework.integration.twitter.inbound;
 
-import twitter4j.Paging;
-import twitter4j.Twitter;
-
 import java.util.List;
 
+import org.springframework.integration.MessagingException;
+
+import twitter4j.Paging;
 
 /**
  * Handles forwarding all new {@link twitter4j.Status} that are 'replies' or 'mentions' to some other tweet.
  *
  * @author Josh Long
+ * @author Oleg Zhurakousky
  */
 public class InboundMentionEndpoint extends AbstractInboundTwitterStatusEndpointSupport {
 
@@ -32,20 +33,29 @@ public class InboundMentionEndpoint extends AbstractInboundTwitterStatusEndpoint
 	public String getComponentType() {
 		return  "twitter:inbound-mention-channel-adapter";
 	}
-
 	@Override
-	protected void beginPolling() throws Exception {
-		 this.runAsAPIRateLimitsPermit(new ApiCallback<InboundMentionEndpoint>() {
-
-			public void run(InboundMentionEndpoint ctx, Twitter twitter) throws Exception {
-				List<twitter4j.Status> stats = (!hasMarkedStatus())
-						? twitter.getMentions()
-						: twitter.getMentions(new Paging(ctx.getMarkerId()));
-				forwardAll( fromTwitter4jStatuses( stats));
+	Runnable getApiCallback() {
+		Runnable apiCallback = new Runnable() {	
+			@Override
+			public void run() {
+				try {
+					long sinceId = getMarkerId();
+					List<twitter4j.Status> stats = (!hasMarkedStatus())
+					? twitter.getMentions()
+					: twitter.getMentions(new Paging(sinceId));
+					System.out.println("Polling. . . .");
+					forwardAll( fromTwitter4jStatuses( stats));
+				} catch (Exception e) {
+					if (e instanceof RuntimeException){
+						throw (RuntimeException)e;
+					}
+					else {
+						throw new MessagingException("Failed to poll for Twitter mentions updates", e);
+					}
+				}
 			}
-
-		});
+		};
+		return apiCallback;
 	}
 
-
 }

spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundTimelineUpdateEndpoint.java

@@ -1,22 +1,23 @@
 /*
- * Copyright 2010 the original author or authors
+ * Copyright 2002-2010 the original author or authors.
  *
- *     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
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ *      http://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 org.springframework.integration.twitter.inbound;
 
+import org.springframework.integration.MessagingException;
+
 import twitter4j.Paging;
-import twitter4j.Twitter;
 
 
 /**
@@ -24,6 +25,7 @@ import twitter4j.Twitter;
  * as messages. It has support for dynamic throttling of API requests.
  *
  * @author Josh Long
+ * @author Oleg Zhurakousky
  * @since 2.0
  */
 public class InboundTimelineUpdateEndpoint extends AbstractInboundTwitterStatusEndpointSupport {
@@ -34,14 +36,25 @@ public class InboundTimelineUpdateEndpoint extends AbstractInboundTwitterStatusE
 	}
 
 	@Override
-	protected void beginPolling() throws Exception {
-		this.runAsAPIRateLimitsPermit(new ApiCallback<InboundTimelineUpdateEndpoint>() {
-			public void run(InboundTimelineUpdateEndpoint t, Twitter twitter)
-					throws Exception {
-				forwardAll( fromTwitter4jStatuses(!t.hasMarkedStatus()
-						? twitter.getFriendsTimeline() :
-						twitter.getFriendsTimeline(new Paging(t.getMarkerId()))));
+	Runnable getApiCallback() {
+		Runnable apiCallback = new Runnable() {	
+			@Override
+			public void run() {
+				try {
+					long sinceId = getMarkerId();
+					forwardAll( fromTwitter4jStatuses(!hasMarkedStatus()
+							? twitter.getFriendsTimeline() :
+							twitter.getFriendsTimeline(new Paging(sinceId))));
+				} catch (Exception e) {
+					if (e instanceof RuntimeException){
+						throw (RuntimeException)e;
+					}
+					else {
+						throw new MessagingException("Failed to poll for Twitter mentions updates", e);
+					}
+				}
 			}
-		});
+		};
+		return apiCallback;
 	}
 }

spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/RateLimitStatusTrigger.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * 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
+ *
+ *      http://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 org.springframework.integration.twitter.inbound;
+
+import java.util.Date;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.scheduling.SchedulingException;
+import org.springframework.scheduling.Trigger;
+import org.springframework.scheduling.TriggerContext;
+import org.springframework.util.Assert;
+
+import twitter4j.RateLimitStatus;
+import twitter4j.Twitter;
+import twitter4j.TwitterException;
+
+/**
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+class RateLimitStatusTrigger implements Trigger {
+	protected final Log logger = LogFactory.getLog(getClass());
+	private Twitter twitter;
+
+	public RateLimitStatusTrigger(Twitter twitter){
+		Assert.notNull(twitter, "'twitter' must not be null");
+		this.twitter = twitter;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.scheduling.Trigger#nextExecutionTime(org.springframework.scheduling.TriggerContext)
+	 */
+	@Override
+	public Date nextExecutionTime(TriggerContext triggerContext) {
+		if (triggerContext.lastCompletionTime() == null){
+			return new Date(System.currentTimeMillis());
+		}
+		try {
+			RateLimitStatus rateLimitStatus = twitter.getRateLimitStatus();
+			int secondsUntilReset = rateLimitStatus.getSecondsUntilReset();
+			int remainingHits = rateLimitStatus.getRemainingHits();
+			 if (remainingHits == 0) {
+                logger.debug(
+                    "rate status limit service returned 0 for the remaining hits value");
+                return null;
+             }
+	         if (secondsUntilReset == 0) {
+	        	 logger.debug(
+                 	"rate status limit service returned 0 for the seconds until reset period value");
+	        	 return null;
+	         }
+	         int secondsUntilWeCanPullAgain = secondsUntilReset / remainingHits;
+	         long msUntilWeCanPullAgain = secondsUntilWeCanPullAgain * 1000;
+	         logger.debug("need to Thread.sleep() " + secondsUntilWeCanPullAgain +
+	        		 " seconds until the next timeline pull. Have " + remainingHits +
+	        		 " remaining pull this rate period. The period ends in " +
+	        		 secondsUntilReset);
+	         return new Date(System.currentTimeMillis() + msUntilWeCanPullAgain);
+		} catch (TwitterException e) {
+			throw new SchedulingException("Failed to schedule the next Twitter update", e);
+		}
+	}
+}

spring-integration-twitter/src/test/java/org/springframework/integration/twitter/InboundDirectMessageStatusEndpointTests.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * 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
+ *
+ *      http://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 org.springframework.integration.twitter;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertNotNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.Message;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.twitter.core.twitter.Twitter4jDirectMessage;
+import org.springframework.integration.twitter.inbound.InboundDirectMessageEndpoint;
+import org.springframework.integration.twitter.oauth.OAuthConfiguration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+import twitter4j.DirectMessage;
+import twitter4j.Paging;
+import twitter4j.RateLimitStatus;
+import twitter4j.ResponseList;
+import twitter4j.Twitter;
+
+/**
+ * @author Oleg Zhurakousky
+ *
+ */
+public class InboundDirectMessageStatusEndpointTests {
+	private DirectMessage firstMessage;
+	private DirectMessage secondMessage;
+	private Twitter twitter;
+	
+	@Test
+	@Ignore
+	/*
+	 * In order to run this test you need to provide values to the twitter.properties file
+	 */
+	public void testUpdatesWithRealTwitter() throws Exception{
+		CountDownLatch latch = new CountDownLatch(1);
+		new ClassPathXmlApplicationContext("TestRecievingUsingNamespace-context.xml", this.getClass());
+		latch.await(10000, TimeUnit.SECONDS);
+	}
+
+	@Test
+	public void testTwitterMockedUpdates() throws Exception{
+		QueueChannel channel = new QueueChannel();
+		
+		InboundDirectMessageEndpoint endpoint = new InboundDirectMessageEndpoint();
+		endpoint.setOutputChannel(channel);
+		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+		scheduler.afterPropertiesSet();
+		endpoint.setTaskScheduler(scheduler);
+		
+		endpoint.setConfiguration(this.getTestConfigurationForDirectMessages());
+		endpoint.afterPropertiesSet();
+		endpoint.start();
+		Message message1 = channel.receive(3000);
+		assertNotNull(message1);
+		assertEquals(secondMessage, ((Twitter4jDirectMessage)message1.getPayload()).getDirectMessage());
+		Message message2 = channel.receive(3000);
+		assertNotNull(message2);
+		assertEquals(firstMessage, ((Twitter4jDirectMessage)message2.getPayload()).getDirectMessage());
+	}
+	
+	@Before
+	public void prepare(){
+		twitter = mock(Twitter.class);
+		firstMessage = mock(DirectMessage.class);
+		when(firstMessage.getCreatedAt()).thenReturn(new Date(5555555555L));
+		when(firstMessage.getId()).thenReturn(200);
+		secondMessage = mock(DirectMessage.class);
+		when(secondMessage.getCreatedAt()).thenReturn(new Date(2222222222L));
+		when(secondMessage.getId()).thenReturn(2000);
+	}
+	
+	@SuppressWarnings("unchecked")
+	private OAuthConfiguration getTestConfigurationForDirectMessages() throws Exception{
+		OAuthConfiguration configuration = mock(OAuthConfiguration.class);
+		RateLimitStatus rateLimitStatus = mock(RateLimitStatus.class);
+		when(twitter.getRateLimitStatus()).thenReturn(rateLimitStatus);
+		when(configuration.getTwitter()).thenReturn(twitter);
+		
+		when(rateLimitStatus.getSecondsUntilReset()).thenReturn(2464);
+		when(rateLimitStatus.getRemainingHits()).thenReturn(250);
+		
+		ResponseList<DirectMessage> responses = mock(ResponseList.class);
+		List<DirectMessage> testMessages = new ArrayList<DirectMessage>();
+		
+		testMessages.add(firstMessage);
+		testMessages.add(secondMessage);
+		
+		when(responses.iterator()).thenReturn(testMessages.iterator());
+		when(twitter.getDirectMessages()).thenReturn(responses);
+		
+		when(twitter.getDirectMessages(Mockito.any(Paging.class))).thenReturn(responses);
+		return configuration;
+	}
+}

spring-integration-twitter/src/test/java/org/springframework/integration/twitter/SimpleTwitterTestClient.java

@@ -1,40 +0,0 @@
-package org.springframework.integration.twitter;
-
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.integration.twitter.oauth.OAuthConfiguration;
-import org.springframework.test.context.ContextConfiguration;
-import twitter4j.Status;
-import twitter4j.Twitter;
-
-import java.util.Collection;
-
-
-/**
- * This class is used to simply demonstrating correctly factory-ing a {@link twitter4j.Twitter} instance
- *
- * @author Josh Long
- */
-@ContextConfiguration
-public class SimpleTwitterTestClient {
-	private Twitter twitter;
-	@Autowired
-	private volatile OAuthConfiguration oAuthConfiguration;
-
-	@Before
-	public void begin() throws Exception {
-		this.twitter = oAuthConfiguration.getTwitter();
-	}
-
-	@Test
-	@Ignore
-	public void testConnectivity() throws Throwable {
-		Collection<Status> responses;
-		Assert.assertNotNull(this.twitter);
-		Assert.assertNotNull(responses = this.twitter.getFriendsTimeline());
-		Assert.assertTrue(responses.size() > 0);
-	}
-}

spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace-context.xml

@@ -39,7 +39,7 @@
             base-package="org.springframework.integration.twitter"/>
 
     <context:property-placeholder
-            location="file://${user.home}/Desktop/twitter.properties"
+            location="classpath:twitter.properties"
             ignore-unresolvable="true"/>
 
     <channel id="inbound_dm"/>

spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TwitterAnnouncer.java

@@ -1,3 +1,18 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * 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
+ *
+ *      http://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 org.springframework.integration.twitter;
 
 import org.springframework.integration.twitter.core.DirectMessage;

spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/RateLimitStatusTriggerTests.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * 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
+ *
+ *      http://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 org.springframework.integration.twitter.inbound;
+
+import static junit.framework.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Date;
+
+import org.junit.Test;
+import org.springframework.scheduling.TriggerContext;
+
+import twitter4j.RateLimitStatus;
+import twitter4j.Twitter;
+
+/**
+ * @author Oleg Zhurakousky
+ *
+ */
+public class RateLimitStatusTriggerTests {
+
+	@Test
+	public void testTriggerImediateAndSubsequentExecutionTime() throws Exception{
+		Twitter twitter = mock(Twitter.class);
+		RateLimitStatusTrigger trigger = new RateLimitStatusTrigger(twitter);
+		TriggerContext context = mock(TriggerContext.class);
+		Date currentDate = new Date(System.currentTimeMillis());
+		Date nextDate = trigger.nextExecutionTime(context);
+		// as long as its within 1 msec we can consider it right away for the purpose of testing
+		assertTrue(nextDate.getTime() - currentDate.getTime() < 100);
+		
+		
+		RateLimitStatus rateLimitStatis = mock(RateLimitStatus.class);
+		when(twitter.getRateLimitStatus()).thenReturn(rateLimitStatis);
+		when(rateLimitStatis.getRemainingHits()).thenReturn(2000);
+		when(rateLimitStatis.getSecondsUntilReset()).thenReturn(4000);
+		
+		when(context.lastCompletionTime()).thenReturn(nextDate);
+		// based on the above values the next execution time should be at least 2000 msec
+		assertTrue(trigger.nextExecutionTime(context).getTime() - nextDate.getTime() > 2000);
+	}
+}

spring-integration-twitter/src/test/java/twitter.properties

@@ -0,0 +1,5 @@
+twitter.oauth.consumerKey=
+twitter.oauth.consumerSecret=
+twitter.oauth.pin=
+twitter.oauth.accessToken=
+twitter.oauth.accessTokenSecret=