diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java index 5d12130fe1..ca26c20a0a 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java @@ -20,10 +20,8 @@ import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Queue; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.Message; @@ -57,8 +55,6 @@ abstract class AbstractTwitterMessageSource extends AbstractEndpoint implemen private volatile long lastPollForTweet; - private final CountDownLatch latch = new CountDownLatch(1); - private final TwitterPollingTask twitterPoller = new TwitterPollingTask(); private volatile MetadataStore metadataStore; @@ -135,18 +131,15 @@ abstract class AbstractTwitterMessageSource extends AbstractEndpoint implemen if (tweet == null){ long currentTime = System.currentTimeMillis(); long diff = currentTime - lastPollForTweet; - lastPollForTweet = currentTime; + if (diff < 15000){ - try { - long waitTime = 15000-diff; - this.latch.await(waitTime, TimeUnit.MILLISECONDS); - } catch (Exception e) { - //ignore - } + return null; } twitterPoller.run(); + tweet = this.tweets.poll(); + lastPollForTweet = currentTime; } - tweet = this.tweets.poll(); + if (tweet != null) { this.lastProcessedId = tweet.getId(); this.metadataStore.put(this.metadataKey, String.valueOf(this.lastProcessedId)); diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java index 0a56fa7c9e..c45521f811 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java @@ -167,6 +167,7 @@ public class DirectMessageReceivingMessageSourceTests { Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets"); assertTrue(!CollectionUtils.isEmpty(msg)); assertEquals(1, msg.size()); + //Thread.sleep(15000); Message message = source.receive(); Tweet tweet = (Tweet) message.getPayload(); assertEquals(2000, tweet.getId());