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

INT-1604 refactored reusable polling logic into AbstractTwitterMessageSource

Browse Source
This commit is contained in:
Mark Fisher
2010-11-11 17:21:28 -05:00
parent 6f5719f352
commit 79b8b6b8b5
5 changed files with 77 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<T> extends AbstractEndpoint i
forward(twitterResponse);
}
}
abstract Runnable getApiCallback();
protected Comparator getComparator() {
private Comparator getComparator() {
return new Comparator<Tweet>() {
public int compare(Tweet tweet1, Tweet tweet2) {
return tweet1.getCreatedAt().compareTo(tweet2.getCreatedAt());
@@ -142,13 +142,12 @@ public abstract class AbstractTwitterMessageSource<T> 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<T> 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<T> extends AbstractEndpoint i
this.metadataStore.put(this.metadataKey, String.valueOf(statusId));
}
/**
* Subclasses must implement this to return tweets.
*/
protected abstract List<Tweet> pollForTweets();
private class TwitterPollingTask implements Runnable {
public void run() {
try {
if (tweets.size() <= prefetchThreshold) {
List<Tweet> tweets = pollForTweets();
if (!CollectionUtils.isEmpty(tweets)) {
forwardAll(tweets);
}
}
}
catch (RuntimeException e) {
throw e;
}
catch (Exception e) {
throw new MessagingException("failed while polling Twitter", e);
}
}
}
}

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

@@ -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<Tweet> {
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<Tweet> 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<Tweet> pollForTweets() {
long sinceId = getMarkerId();
return hasMarkedStatus() ? twitter.getDirectMessages(sinceId) : twitter.getDirectMessages();
}
@SuppressWarnings("rawtypes")
protected Comparator getComparator() {
return new Comparator<Tweet>() {
public int compare(Tweet tweet1, Tweet tweet2) {
return tweet1.getCreatedAt().compareTo(tweet2.getCreatedAt());
}
};
}
}

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

@@ -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<Tweet> 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<Tweet> pollForTweets() {
long sinceId = getMarkerId();
return hasMarkedStatus() ? twitter.getMentions(sinceId) : twitter.getMentions();
}
}

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

@@ -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<Tweet> {
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<Tweet> 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<Tweet> pollForTweets() {
SearchResults results = this.twitter.search(query);
return (results != null) ? results.getTweets() : null;
}
}

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

@@ -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<Tweet> {
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<Tweet> 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<Tweet> pollForTweets() {
long sinceId = getMarkerId();
return hasMarkedStatus() ? twitter.getTimeline(sinceId) : twitter.getTimeline();
}
}