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(); } }