From e02582f07955a0df80a45cbbe1b620e253413615 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 11 Nov 2010 18:42:28 -0500 Subject: [PATCH] INT-1604 removed unused methods from TwitterOperations and Twitter4JTemplate, renamed 'markerId' to 'lastEnqueuedId', and now passing the 'sinceId' into subclass' pollForTweets(..) callback, and the search adapter now takes 'sinceId' into account as well --- .../twitter/core/Twitter4jTemplate.java | 13 +------- .../twitter/core/TwitterOperations.java | 26 ++-------------- .../inbound/AbstractTwitterMessageSource.java | 30 ++++++++----------- .../DirectMessageReceivingMessageSource.java | 7 +++-- .../MentionsReceivingMessageSource.java | 5 ++-- .../inbound/SearchReceivingMessageSource.java | 6 ++-- .../TimelineReceivingMessageSource.java | 5 ++-- 7 files changed, 28 insertions(+), 64 deletions(-) diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/Twitter4jTemplate.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/Twitter4jTemplate.java index 0c9d84fa7b..250ec443e0 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/Twitter4jTemplate.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/Twitter4jTemplate.java @@ -200,24 +200,13 @@ public class Twitter4jTemplate implements TwitterOperations { return this.search(q); } - public SearchResults search(String query, int page, int sinceId) { + public SearchResults search(String query, int page, long sinceId) { Assert.hasText(query, "'query' must not be null"); Query q = new Query(query); - q.setPage(page); q.setSinceId(sinceId); return this.search(q); } - public SearchResults search(String query, int page, int resultsPerPage, int sinceId, int maxId) { - Assert.hasText(query, "'query' must not be null"); - Query q = new Query(query); - q.setPage(page); - q.setSinceId(sinceId); - q.setMaxId(maxId); - q.setRpp(resultsPerPage); - return this.search(q); - } - public Twitter getUnderlyingTwitter() { return this.twitter; } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/TwitterOperations.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/TwitterOperations.java index 055eee6f35..f52fb1fe7f 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/TwitterOperations.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/TwitterOperations.java @@ -81,34 +81,14 @@ public interface TwitterOperations { * The search query string * @param page * The page to return - * @param pageSize - * The number of {@link Tweet}s per page - * - * @return a {@link SearchResults} containing {@link Tweet}s - * - */ - SearchResults search(String query, int page, int sinceId); - - /** - * Searches Twitter, returning a specific page out of the complete set of - * results. Results are filtered to those whose ID falls between sinceId and - * maxId - * - * @param query - * The search query string - * @param page - * The page to return - * @param pageSize - * The number of {@link Tweet}s per page * @param sinceId * The minimum {@link Tweet} ID to return in the results - * @param maxId - * The maximum {@link Tweet} ID to return in the results * * @return a {@link SearchResults} containing {@link Tweet}s + * */ - SearchResults search(String query, int page, int resultsPerPage, int sinceId, int maxId); - + SearchResults search(String query, int page, long sinceId); + List getDirectMessages(); List getDirectMessages(long sinceId); 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 408be88291..ab56b311c9 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 @@ -61,9 +61,9 @@ abstract class AbstractTwitterMessageSource extends AbstractEndpoint implemen private volatile int prefetchThreshold = 0; - private volatile long markerId = -1; + private volatile long lastEnqueuedId = -1; - //private volatile long processedId = -1; + private volatile long lastProcessedId = -1; private final TwitterOperations twitterOperations; @@ -80,14 +80,6 @@ abstract class AbstractTwitterMessageSource extends AbstractEndpoint implemen } - public long getMarkerId() { - return this.markerId; - } - - protected boolean hasMarkedStatus() { - return this.markerId > -1; - } - protected TwitterOperations getTwitterOperations() { return this.twitterOperations; } @@ -125,14 +117,15 @@ abstract class AbstractTwitterMessageSource extends AbstractEndpoint implemen String lastId = this.metadataStore.get(this.metadataKey); // initialize the last status ID from the metadataStore if (StringUtils.hasText(lastId)) { - this.markerId = Long.parseLong(lastId); + this.lastProcessedId = Long.parseLong(lastId); + this.lastEnqueuedId = this.lastProcessedId; } } public Message receive() { Tweet tweet = this.tweets.poll(); if (tweet != null) { - this.markProcessedId(tweet.getId()); + this.setLastProcessedId(tweet.getId()); return MessageBuilder.withPayload(tweet).build(); } return null; @@ -149,23 +142,24 @@ abstract class AbstractTwitterMessageSource extends AbstractEndpoint implemen private void enqueue(Tweet tweet) { synchronized (this.markerGuard) { long id = tweet.getId(); - if (id > this.markerId) { - this.markerId = id; + if (id > this.lastEnqueuedId) { + this.lastEnqueuedId = id; this.tweets.add(tweet); } } } - private void markProcessedId(long statusId) { - //this.processedId = statusId; + private void setLastProcessedId(long statusId) { + this.lastProcessedId = statusId; this.metadataStore.put(this.metadataKey, String.valueOf(statusId)); } /** * Subclasses must implement this to return tweets. + * The 'sinceId' value will be negative if no last id is known. */ - protected abstract List pollForTweets(); + protected abstract List pollForTweets(long sinceId); // Lifecycle methods @@ -191,7 +185,7 @@ abstract class AbstractTwitterMessageSource extends AbstractEndpoint implemen public void run() { try { if (tweets.size() <= prefetchThreshold) { - List tweets = pollForTweets(); + List tweets = pollForTweets(lastEnqueuedId); if (!CollectionUtils.isEmpty(tweets)) { enqueueAll(tweets); } 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 5511e0e8be..918f88a2bb 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 @@ -42,9 +42,10 @@ public class DirectMessageReceivingMessageSource extends AbstractTwitterMessageS } @Override - protected List pollForTweets() { - long sinceId = getMarkerId(); - return hasMarkedStatus() ? this.getTwitterOperations().getDirectMessages(sinceId) : this.getTwitterOperations().getDirectMessages(); + protected List pollForTweets(long sinceId) { + return (sinceId > 0) + ? this.getTwitterOperations().getDirectMessages(sinceId) + : this.getTwitterOperations().getDirectMessages(); } } 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 a95c6bd72b..7555815928 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 @@ -40,9 +40,8 @@ public class MentionsReceivingMessageSource extends AbstractTwitterMessageSource } @Override - protected List pollForTweets() { - long sinceId = getMarkerId(); - return hasMarkedStatus() ? this.getTwitterOperations().getMentions(sinceId) : this.getTwitterOperations().getMentions(); + protected List pollForTweets(long sinceId) { + return (sinceId > 0) ? this.getTwitterOperations().getMentions(sinceId) : this.getTwitterOperations().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 b463aef86c..d3f51d78f2 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 @@ -49,8 +49,10 @@ public class SearchReceivingMessageSource extends AbstractTwitterMessageSource pollForTweets() { - SearchResults results = this.getTwitterOperations().search(query); + protected List pollForTweets(long sinceId) { + SearchResults results = (sinceId > 0) + ? this.getTwitterOperations().search(query, 1, sinceId) + : this.getTwitterOperations().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 500b45864f..09cd359288 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 @@ -42,9 +42,8 @@ public class TimelineReceivingMessageSource extends AbstractTwitterMessageSource } @Override - protected List pollForTweets() { - long sinceId = getMarkerId(); - return hasMarkedStatus() ? this.getTwitterOperations().getTimeline(sinceId) : this.getTwitterOperations().getTimeline(); + protected List pollForTweets(long sinceId) { + return(sinceId > 0) ? this.getTwitterOperations().getTimeline(sinceId) : this.getTwitterOperations().getTimeline(); } }