Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Mark Fisher
2010-11-11 18:42:28 -05:00
parent 82ddb01cfd
commit e02582f079
7 changed files with 28 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/Twitter4jTemplate.java
View File

@@ -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;
}

26
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/TwitterOperations.java
View File

@@ -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<Tweet> getDirectMessages();
List<Tweet> getDirectMessages(long sinceId);

30
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java
View File

@@ -61,9 +61,9 @@ abstract class AbstractTwitterMessageSource<T> 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<T> 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<T> 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<T> 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<Tweet> pollForTweets();
protected abstract List<Tweet> pollForTweets(long sinceId);
// Lifecycle methods
@@ -191,7 +185,7 @@ abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint implemen
public void run() {
try {
if (tweets.size() <= prefetchThreshold) {
List<Tweet> tweets = pollForTweets();
List<Tweet> tweets = pollForTweets(lastEnqueuedId);
if (!CollectionUtils.isEmpty(tweets)) {
enqueueAll(tweets);
}

7
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSource.java
View File

@@ -42,9 +42,10 @@ public class DirectMessageReceivingMessageSource extends AbstractTwitterMessageS
}
@Override
protected List<Tweet> pollForTweets() {
long sinceId = getMarkerId();
return hasMarkedStatus() ? this.getTwitterOperations().getDirectMessages(sinceId) : this.getTwitterOperations().getDirectMessages();
protected List<Tweet> pollForTweets(long sinceId) {
return (sinceId > 0)
? this.getTwitterOperations().getDirectMessages(sinceId)
: this.getTwitterOperations().getDirectMessages();
}
}

5
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionsReceivingMessageSource.java
View File

@@ -40,9 +40,8 @@ public class MentionsReceivingMessageSource extends AbstractTwitterMessageSource
}
@Override
protected List<Tweet> pollForTweets() {
long sinceId = getMarkerId();
return hasMarkedStatus() ? this.getTwitterOperations().getMentions(sinceId) : this.getTwitterOperations().getMentions();
protected List<Tweet> pollForTweets(long sinceId) {
return (sinceId > 0) ? this.getTwitterOperations().getMentions(sinceId) : this.getTwitterOperations().getMentions();
}
}

6
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java
View File

@@ -49,8 +49,10 @@ public class SearchReceivingMessageSource extends AbstractTwitterMessageSource<T
}
@Override
protected List<Tweet> pollForTweets() {
SearchResults results = this.getTwitterOperations().search(query);
protected List<Tweet> pollForTweets(long sinceId) {
SearchResults results = (sinceId > 0)
? this.getTwitterOperations().search(query, 1, sinceId)
: this.getTwitterOperations().search(query);
return (results != null) ? results.getTweets() : null;
}

5
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSource.java
View File

@@ -42,9 +42,8 @@ public class TimelineReceivingMessageSource extends AbstractTwitterMessageSource
}
@Override
protected List<Tweet> pollForTweets() {
long sinceId = getMarkerId();
return hasMarkedStatus() ? this.getTwitterOperations().getTimeline(sinceId) : this.getTwitterOperations().getTimeline();
protected List<Tweet> pollForTweets(long sinceId) {
return(sinceId > 0) ? this.getTwitterOperations().getTimeline(sinceId) : this.getTwitterOperations().getTimeline();
}
}