From 79b8b6b8b59f7542af36ab8cc6d26913acf07b32 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 11 Nov 2010 17:21:28 -0500 Subject: [PATCH] INT-1604 refactored reusable polling logic into AbstractTwitterMessageSource --- .../inbound/AbstractTwitterMessageSource.java | 49 ++++++++++++++----- .../DirectMessageReceivingMessageSource.java | 48 +++--------------- .../MentionsReceivingMessageSource.java | 33 ++++--------- .../inbound/SearchReceivingMessageSource.java | 41 +++++----------- .../TimelineReceivingMessageSource.java | 37 ++++---------- 5 files changed, 77 insertions(+), 131 deletions(-) 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 cfd85664ac..1514493969 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 @@ -25,6 +25,7 @@ import java.util.concurrent.ScheduledFuture; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.core.MessageSource; import org.springframework.integration.endpoint.AbstractEndpoint; @@ -34,6 +35,7 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.twitter.core.Tweet; import org.springframework.integration.twitter.core.TwitterOperations; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; /** @@ -129,10 +131,8 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i forward(twitterResponse); } } - - abstract Runnable getApiCallback(); - protected Comparator getComparator() { + private Comparator getComparator() { return new Comparator() { public int compare(Tweet tweet1, Tweet tweet2) { return tweet1.getCreatedAt().compareTo(tweet2.getCreatedAt()); @@ -142,13 +142,12 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i @Override protected void doStart(){ - Assert.notNull(this.twitter, "'twitter' instance can't be null"); - // temporarily injecting Twitter into a trigger so it can deal with Rate Limits. will be changed - // once we switch to Spring Social + Assert.notNull(this.twitter, "'twitter' instance must not be null"); + // temporarily injecting Twitter into a trigger so it can deal with Rate Limits. + // This will likely change once we switch to Spring Social. RateLimitStatusTrigger trigger = new RateLimitStatusTrigger(this.twitter.getUnderlyingTwitter()); - // - Runnable apiCallback = this.getApiCallback(); - twitterUpdatePollingTask = this.getTaskScheduler().schedule(apiCallback, trigger); + Runnable twitterPollingTask = new TwitterPollingTask(); + twitterUpdatePollingTask = this.getTaskScheduler().schedule(twitterPollingTask, trigger); } @Override @@ -157,14 +156,14 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i } public Message receive() { - Tweet tweet = tweets.poll(); - if (tweet != null){ + Tweet tweet = this.tweets.poll(); + if (tweet != null) { this.markProcessedId(tweet.getId()); return MessageBuilder.withPayload(tweet).build(); } return null; } - + protected void forward(Tweet tweet) { synchronized (this.markerGuard) { long id = tweet.getId(); @@ -180,4 +179,30 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i this.metadataStore.put(this.metadataKey, String.valueOf(statusId)); } + /** + * Subclasses must implement this to return tweets. + */ + protected abstract List pollForTweets(); + + + private class TwitterPollingTask implements Runnable { + + public void run() { + try { + if (tweets.size() <= prefetchThreshold) { + List tweets = pollForTweets(); + if (!CollectionUtils.isEmpty(tweets)) { + forwardAll(tweets); + } + } + } + catch (RuntimeException e) { + throw e; + } + catch (Exception e) { + throw new MessagingException("failed while polling Twitter", e); + } + } + } + } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSource.java index 612a2a09e2..2895cc8ebf 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSource.java @@ -15,67 +15,35 @@ */ package org.springframework.integration.twitter.inbound; -import java.util.Comparator; import java.util.List; -import org.springframework.integration.MessagingException; import org.springframework.integration.twitter.core.Tweet; import org.springframework.integration.twitter.core.TwitterOperations; -import org.springframework.util.CollectionUtils; /** * This class handles support for receiving DMs (direct messages) using Twitter. * * @author Josh Long * @author Oleg Zhurakousky + * @author Mark Fisher * @since 2.0 */ public class DirectMessageReceivingMessageSource extends AbstractTwitterMessageSource { - - public DirectMessageReceivingMessageSource(TwitterOperations twitter){ + + public DirectMessageReceivingMessageSource(TwitterOperations twitter) { super(twitter); } - + + @Override public String getComponentType() { return "twitter:dm-inbound-channel-adapter"; } @Override - Runnable getApiCallback() { - Runnable apiCallback = new Runnable() { - public void run() { - try { - long sinceId = getMarkerId(); - if (tweets.size() <= prefetchThreshold){ - List dms = !hasMarkedStatus() - ? twitter.getDirectMessages() - : twitter.getDirectMessages(sinceId); - - if (!CollectionUtils.isEmpty(dms)){ - forwardAll(dms); - } - } - } - catch (Exception e) { - if (e instanceof RuntimeException){ - throw (RuntimeException)e; - } - else { - throw new MessagingException("Failed to poll for Twitter mentions updates", e); - } - } - } - }; - return apiCallback; + protected List pollForTweets() { + long sinceId = getMarkerId(); + return hasMarkedStatus() ? twitter.getDirectMessages(sinceId) : twitter.getDirectMessages(); } - @SuppressWarnings("rawtypes") - protected Comparator getComparator() { - return new Comparator() { - public int compare(Tweet tweet1, Tweet tweet2) { - return tweet1.getCreatedAt().compareTo(tweet2.getCreatedAt()); - } - }; - } } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionsReceivingMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionsReceivingMessageSource.java index 5c4330d5a1..982612b908 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionsReceivingMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionsReceivingMessageSource.java @@ -13,11 +13,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.twitter.inbound; import java.util.List; -import org.springframework.integration.MessagingException; import org.springframework.integration.twitter.core.Tweet; import org.springframework.integration.twitter.core.TwitterOperations; @@ -32,32 +32,17 @@ public class MentionsReceivingMessageSource extends AbstractTwitterMessageSource public MentionsReceivingMessageSource(TwitterOperations twitter){ super(twitter); } + + @Override public String getComponentType() { - return "twitter:inbound-mention-channel-adapter"; + return "twitter:mention-inbound-channel-adapter"; } + @Override - Runnable getApiCallback() { - Runnable apiCallback = new Runnable() { - public void run() { - try { - long sinceId = getMarkerId(); - if (tweets.size() <= prefetchThreshold){ - List stats = (!hasMarkedStatus()) - ? twitter.getMentions() - : twitter.getMentions(sinceId); - forwardAll(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; + protected List pollForTweets() { + long sinceId = getMarkerId(); + return hasMarkedStatus() ? twitter.getMentions(sinceId) : twitter.getMentions(); } + } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java index 8ded71b0ac..800011c315 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java @@ -13,60 +13,45 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.twitter.inbound; import java.util.List; -import org.springframework.integration.MessagingException; import org.springframework.integration.twitter.core.SearchResults; import org.springframework.integration.twitter.core.Tweet; import org.springframework.integration.twitter.core.TwitterOperations; import org.springframework.util.Assert; /** - * * @author Oleg Zhurakousky + * @author Mark Fisher * @since 2.0 */ public class SearchReceivingMessageSource extends AbstractTwitterMessageSource { private volatile String query; - - public SearchReceivingMessageSource(TwitterOperations twitter){ + + + public SearchReceivingMessageSource(TwitterOperations twitter) { super(twitter); } - + + public void setQuery(String query) { - Assert.hasText(query, "'query' must no be null"); + Assert.hasText(query, "query must no be null"); this.query = query; } - + @Override public String getComponentType() { return "twitter:search-inbound-channel-adapter"; } @Override - Runnable getApiCallback() { - Runnable apiCallback = new Runnable() { - public void run() { - try { - if (tweets.size() <= prefetchThreshold){ - SearchResults results = twitter.search(query); - - List twetList = results.getTweets(); - forwardAll(twetList); - } - } catch (Exception e) { - if (e instanceof RuntimeException){ - throw (RuntimeException)e; - } - else { - throw new MessagingException("Failed to poll for Twitter mentions updates", e); - } - } - } - }; - return apiCallback; + protected List pollForTweets() { + SearchResults results = this.twitter.search(query); + return (results != null) ? results.getTweets() : null; } + } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSource.java index 054d4b34aa..6622ab5ec6 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSource.java @@ -13,55 +13,38 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.twitter.inbound; import java.util.List; -import org.springframework.integration.MessagingException; import org.springframework.integration.twitter.core.Tweet; import org.springframework.integration.twitter.core.TwitterOperations; - /** - * This {@link org.springframework.integration.core.MessageSource} lets Spring Integration consume a given account's timeline - * as messages. It has support for dynamic throttling of API requests. + * This {@link org.springframework.integration.core.MessageSource} lets Spring Integration consume + * given account's timeline as messages. It has support for dynamic throttling of API requests. * * @author Josh Long * @author Oleg Zhurakousky * @since 2.0 */ public class TimelineReceivingMessageSource extends AbstractTwitterMessageSource { - + public TimelineReceivingMessageSource(TwitterOperations twitter){ super(twitter); } + + @Override public String getComponentType() { return "twitter:inbound-channel-adapter"; } @Override - Runnable getApiCallback() { - Runnable apiCallback = new Runnable() { - public void run() { - try { - long sinceId = getMarkerId(); - if (tweets.size() <= prefetchThreshold){ - List tweets = !hasMarkedStatus() - ? twitter.getTimeline() - : twitter.getTimeline(sinceId); - forwardAll(tweets); - } - } catch (Exception e) { - if (e instanceof RuntimeException){ - throw (RuntimeException)e; - } - else { - throw new MessagingException("Failed to poll for Twitter mentions updates", e); - } - } - } - }; - return apiCallback; + protected List pollForTweets() { + long sinceId = getMarkerId(); + return hasMarkedStatus() ? twitter.getTimeline(sinceId) : twitter.getTimeline(); } + }