From fa135f9c6cc3c66a71ad0a0704a0c02c68b73a59 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 11 Nov 2010 12:58:05 -0500 Subject: [PATCH] INT-1603 added SearchReceivingMessageSource --- .../twitter/core/Twitter4jTemplate.java | 7 +- .../inbound/SearchReceivingMessageSource.java | 93 +++++++++++++++++++ .../twitter/core/Twitter4jTemplateTests.java | 2 +- .../SearchReceivingMessageSourceTests.java | 58 ++++++++++++ 4 files changed, 158 insertions(+), 2 deletions(-) create mode 100644 spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java create mode 100644 spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java 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 7dafbf34be..a01f8c047c 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 @@ -70,7 +70,12 @@ public class Twitter4jTemplate implements TwitterOperations{ @Override public String getProfileId() { try { - return twitter.getScreenName(); + if (twitter.isOAuthEnabled()){ + return twitter.getScreenName(); + } + else { + return "twitter-anonymous"; + } } catch (Exception e) { throw new TwitterOperationException("Failed to obtain Profile ID. ", e); 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 new file mode 100644 index 0000000000..23e263c54b --- /dev/null +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java @@ -0,0 +1,93 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * 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 + * @since 2.0 + */ +public class SearchReceivingMessageSource extends AbstractTwitterMessageSource { + /* since Twitter return 15 entries per page we need to be able to manage + * how many pages deep are we willing to go. Not sure yet about exposing this attribute via namespace + * but setting default to 10. + */ + private volatile int pageDepth = 10; + + private volatile int currentPage = 1; + private volatile String query; + + public SearchReceivingMessageSource(TwitterOperations twitter){ + super(twitter); + } + + public void setQuery(String query) { + 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){ + if (currentPage == pageDepth){ + currentPage = 1; + } + SearchResults results = twitter.search(query, currentPage, 0); + List twetList = results.getTweets(); + if (currentPage == 1){ + forwardAll(twetList); + } + else { + for (Tweet tweet : twetList) { + tweets.add(tweet); + } + } + if (twetList != null && twetList.size() > 0){ + currentPage++; + } + else { + currentPage = 1; + } + } + } catch (Exception e) { + if (e instanceof RuntimeException){ + throw (RuntimeException)e; + } + else { + throw new MessagingException("Failed to poll for Twitter mentions updates", e); + } + } + } + }; + return apiCallback; + } +} diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/core/Twitter4jTemplateTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/core/Twitter4jTemplateTests.java index faf9d4094e..bd4236cf62 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/core/Twitter4jTemplateTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/core/Twitter4jTemplateTests.java @@ -135,6 +135,6 @@ public class Twitter4jTemplateTests { assertTrue(tweets.get(0) instanceof Tweet); assertTrue(tweets.get(1) instanceof Tweet); assertTrue(tweets.get(2) instanceof Tweet); - } + } diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java new file mode 100644 index 0000000000..bb14791b3b --- /dev/null +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java @@ -0,0 +1,58 @@ +/** + * + */ +package org.springframework.integration.twitter.inbound; + +import org.junit.Ignore; +import org.junit.Test; + +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessagingException; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.twitter.core.Tweet; +import org.springframework.integration.twitter.core.Twitter4jTemplate; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +/** + * @author ozhurakousky + * + */ +public class SearchReceivingMessageSourceTests { + + @Test + @Ignore + public void testSearchReceiving() throws Exception{ + DefaultListableBeanFactory bf = new DefaultListableBeanFactory(); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + bf.registerSingleton("taskScheduler", scheduler); + + SearchReceivingMessageSource ms = new SearchReceivingMessageSource(new Twitter4jTemplate()); + + DirectChannel channel = new DirectChannel(); + channel.subscribe(new MessageHandler() { + public void handleMessage(Message message) throws MessagingException { + System.out.println("Message: " + ((Tweet)message.getPayload()).getCreatedAt() + " - " + ((Tweet)message.getPayload()).getText()); + } + }); + SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); + adapter.setSource(ms); + adapter.setBeanFactory(bf); + adapter.setOutputChannel(channel); + adapter.afterPropertiesSet(); + adapter.start(); + + ms.setBeanFactory(bf); + ms.setQuery("#springintegration"); + + ms.setTaskScheduler(scheduler); + ms.afterPropertiesSet(); + ms.start(); + + System.in.read(); + } +}