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

AbstractTwitterMessageSource no longer implements Lifecycle (no longer extends AbstractEndpoint). @Ignore-d irrelevant tests, and removed the RateLimitStatusTrigger.

Browse Source
This commit is contained in:
Mark Fisher
2010-11-18 19:44:48 -05:00
parent 5265dad85d
commit 382e074f59
9 changed files with 62 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterInboundChannelAdapterParser.java
View File

@@ -48,7 +48,6 @@ public class TwitterInboundChannelAdapterParser extends AbstractPollingInboundCh
BASE_PACKAGE + ".core.Twitter4jTemplate");
builder.addConstructorArgValue(templateBuilder.getBeanDefinition());
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "query");
return builder.getBeanDefinition();
}

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

@@ -21,14 +21,13 @@ import java.util.Comparator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
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.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.store.MetadataStore;
import org.springframework.integration.store.SimpleMetadataStore;
import org.springframework.integration.support.MessageBuilder;
@@ -51,11 +50,11 @@ import org.springframework.util.StringUtils;
* @since 2.0
*/
@SuppressWarnings("rawtypes")
abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint implements MessageSource {
abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport implements MessageSource {
private volatile long lastPollForTweet;
private final TwitterPollingTask twitterPoller = new TwitterPollingTask();
private final TwitterPollingTask twitterPollingTask = new TwitterPollingTask();
private volatile MetadataStore metadataStore;
@@ -73,8 +72,6 @@ abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint implemen
private final TweetComparator tweetComparator = new TweetComparator();
private volatile ScheduledFuture<?> twitterPollingTask;
private final Object markerGuard = new Object();
@@ -91,8 +88,6 @@ abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint implemen
@Override
protected void onInit() throws Exception{
super.onInit();
Assert.notNull(this.getTaskScheduler(),
"Unable to locate TaskScheduler. You must inject one explicitly or define a bean by the name 'taskScheduler'.");
if (this.metadataStore == null) {
// first try to look for a 'metadataStore' in the context
BeanFactory beanFactory = this.getBeanFactory();
@@ -128,18 +123,17 @@ abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint implemen
public Message<?> receive() {
Tweet tweet = this.tweets.poll();
if (tweet == null){
if (tweet == null) {
long currentTime = System.currentTimeMillis();
long diff = currentTime - lastPollForTweet;
if (diff < 15000){
long elapsedTime = currentTime - this.lastPollForTweet;
if (elapsedTime < 15000) {
// need to wait longer
return null;
}
twitterPoller.run();
this.twitterPollingTask.run();
tweet = this.tweets.poll();
lastPollForTweet = currentTime;
this.lastPollForTweet = currentTime;
}
if (tweet != null) {
this.lastProcessedId = tweet.getId();
this.metadataStore.put(this.metadataKey, String.valueOf(this.lastProcessedId));
@@ -173,24 +167,6 @@ abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint implemen
protected abstract List<Tweet> pollForTweets(long sinceId);
// Lifecycle methods
@Override
protected void doStart() {
// 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.twitterOperations.getUnderlyingTwitter());
this.twitterPollingTask = this.getTaskScheduler().schedule(new TwitterPollingTask(), trigger);
}
@Override
protected void doStop() {
if (this.twitterPollingTask != null) {
this.twitterPollingTask.cancel(true);
}
}
private class TwitterPollingTask implements Runnable {
public void run() {

85
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/RateLimitStatusTrigger.java
View File

@@ -1,85 +0,0 @@
/*
* 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.Date;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.scheduling.SchedulingException;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.util.Assert;
import twitter4j.RateLimitStatus;
import twitter4j.Twitter;
import twitter4j.TwitterException;
/**
* Trigger implementation that takes the Twitter rate limit into consideration.
*
* @author Oleg Zhurakousky
* @since 2.0
*/
class RateLimitStatusTrigger implements Trigger {
private final Log logger = LogFactory.getLog(getClass());
private final Twitter twitter;
public RateLimitStatusTrigger(Twitter twitter) {
Assert.notNull(twitter, "'twitter' must not be null");
this.twitter = twitter;
}
/**
* Returns the next time the task may execute. Considers the Twitter rate limit.
*/
public Date nextExecutionTime(TriggerContext triggerContext) {
if (triggerContext.lastCompletionTime() == null) {
return new Date();
}
try {
RateLimitStatus rateLimitStatus = this.twitter.getRateLimitStatus();
int secondsUntilReset = rateLimitStatus.getSecondsUntilReset();
int remainingHits = rateLimitStatus.getRemainingHits();
if (remainingHits == 0) {
if (logger.isDebugEnabled()) {
logger.debug("rate status limit service returned 0 for the remaining hits value");
}
return null;
}
if (secondsUntilReset == 0) {
if (logger.isDebugEnabled()) {
logger.debug("rate status limit service returned 0 for the seconds until reset period value");
}
return null;
}
int secondsUntilWeCanPullAgain = secondsUntilReset / remainingHits;
long msUntilWeCanPullAgain = secondsUntilWeCanPullAgain * 1000;
logger.debug("Waiting for " + secondsUntilWeCanPullAgain
+ " seconds until the next timeline pull. Have " + remainingHits
+ " remaining pull this rate period. The period ends in " + secondsUntilReset);
return new Date(System.currentTimeMillis() + msUntilWeCanPullAgain);
}
catch (TwitterException e) {
throw new SchedulingException("Failed to schedule the next Twitter update", e);
}
}
}

40
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingMessageSourceParserTests.java
View File

@@ -16,18 +16,10 @@
package org.springframework.integration.twitter.config;
import static junit.framework.Assert.assertFalse;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import org.junit.Test;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.SmartLifecycle;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.twitter.core.TwitterOperations;
/**
@@ -35,21 +27,23 @@ import org.springframework.integration.twitter.core.TwitterOperations;
*/
public class TestReceivingMessageSourceParserTests {
@Test
public void testRecievingAdapterConfigurationAutoStartup(){
ApplicationContext ac = new ClassPathXmlApplicationContext("TestReceivingMessageSourceParser-context.xml", this.getClass());
SourcePollingChannelAdapter spca = ac.getBean("mentionAdapter", SourcePollingChannelAdapter.class);
SmartLifecycle ms = TestUtils.getPropertyValue(spca, "source", SmartLifecycle.class);
assertFalse(ms.isAutoStartup());
spca = ac.getBean("dmAdapter", SourcePollingChannelAdapter.class);
ms = TestUtils.getPropertyValue(spca, "source", SmartLifecycle.class);
assertFalse(ms.isAutoStartup());
spca = ac.getBean("updateAdapter", SourcePollingChannelAdapter.class);
ms = TestUtils.getPropertyValue(spca, "source", SmartLifecycle.class);
assertFalse(ms.isAutoStartup());
}
// NO LONGER RELEVANT...
// @Test
// public void testRecievingAdapterConfigurationAutoStartup(){
// ApplicationContext ac = new ClassPathXmlApplicationContext("TestReceivingMessageSourceParser-context.xml", this.getClass());
// SourcePollingChannelAdapter spca = ac.getBean("mentionAdapter", SourcePollingChannelAdapter.class);
// SmartLifecycle ms = TestUtils.getPropertyValue(spca, "source", SmartLifecycle.class);
// assertFalse(ms.isAutoStartup());
//
// spca = ac.getBean("dmAdapter", SourcePollingChannelAdapter.class);
// ms = TestUtils.getPropertyValue(spca, "source", SmartLifecycle.class);
// assertFalse(ms.isAutoStartup());
//
// spca = ac.getBean("updateAdapter", SourcePollingChannelAdapter.class);
// ms = TestUtils.getPropertyValue(spca, "source", SmartLifecycle.class);
// assertFalse(ms.isAutoStartup());
// }
public static class TwitterTemplateFactoryBean implements FactoryBean<TwitterOperations>{

6
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestSearchReceivingMessageSourceParserTests.java
View File

@@ -39,8 +39,7 @@ public class TestSearchReceivingMessageSourceParserTests {
ApplicationContext ac = new ClassPathXmlApplicationContext("TestSearchReceivingMessageSourceParser-context.xml", this.getClass());
SourcePollingChannelAdapter spca = ac.getBean("searchAdapter", SourcePollingChannelAdapter.class);
SearchReceivingMessageSource ms = (SearchReceivingMessageSource) TestUtils.getPropertyValue(spca, "source");
assertFalse(ms.isAutoStartup());
assertFalse(ms.isAutoStartup());
//assertFalse(ms.isAutoStartup());
Twitter4jTemplate template = (Twitter4jTemplate) TestUtils.getPropertyValue(ms, "twitterOperations");
assertFalse(template.getUnderlyingTwitter().isOAuthEnabled()); // verify anonymous Twitter
}
@@ -50,8 +49,7 @@ public class TestSearchReceivingMessageSourceParserTests {
ApplicationContext ac = new ClassPathXmlApplicationContext("TestSearchReceivingMessageSourceParser-context.xml", this.getClass());
SourcePollingChannelAdapter spca = ac.getBean("searchAdapterWithTemplate", SourcePollingChannelAdapter.class);
SearchReceivingMessageSource ms = (SearchReceivingMessageSource) TestUtils.getPropertyValue(spca, "source");
assertFalse(ms.isAutoStartup());
assertFalse(ms.isAutoStartup());
//assertFalse(ms.isAutoStartup());
TwitterOperations template = (TwitterOperations) TestUtils.getPropertyValue(ms, "twitterOperations");
assertEquals(ac.getBean("twitter"), template);
}

27
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java
View File

@@ -31,6 +31,7 @@ import java.util.Date;
import java.util.Queue;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.Mockito;
@@ -105,27 +106,28 @@ public class DirectMessageReceivingMessageSourceTests {
DirectMessageReceivingMessageSource source = new DirectMessageReceivingMessageSource(twitter);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
source.setTaskScheduler(scheduler);
//source.setTaskScheduler(scheduler);
source.setBeanName("twitterEndpoint");
source.afterPropertiesSet();
source.start();
//source.start();
assertEquals("twitter:dm-inbound-channel-adapter.twitterEndpoint.kermit", TestUtils.getPropertyValue(source, "metadataKey"));
assertTrue(source.isRunning());
source.stop();
//assertTrue(source.isRunning());
//source.stop();
}
@SuppressWarnings("rawtypes")
@Test
@Ignore
public void testSuccessfullInitializationWithMessages() throws Exception{
this.setUpMockScenarioForMessagePolling();
DirectMessageReceivingMessageSource source = new DirectMessageReceivingMessageSource(twitter);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
source.setTaskScheduler(scheduler);
//source.setTaskScheduler(scheduler);
source.setBeanName("twitterEndpoint");
source.afterPropertiesSet();
source.start();
//source.start();
Thread.sleep(1000);
Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
assertTrue(!CollectionUtils.isEmpty(msg));
@@ -136,7 +138,7 @@ public class DirectMessageReceivingMessageSourceTests {
verify(twitter, times(1)).getDirectMessages();
// based on the Mock, the Queue should now have 2 more messages third and fourth
assertTrue(((Queue)TestUtils.getPropertyValue(source, "tweets")).size() == 2);
source.stop();
//source.stop();
}
/**
* This test will validate that last status is initialized from the metadatastore
@@ -144,6 +146,7 @@ public class DirectMessageReceivingMessageSourceTests {
*/
@SuppressWarnings("rawtypes")
@Test
@Ignore
public void testSuccessfullInitializationWithMessagesWithPersistentMetadata() throws Exception{
String fileName = System.getProperty("java.io.tmpdir") + "/spring-integration/metadata-store.properties";
File file = new File(fileName);
@@ -159,10 +162,10 @@ public class DirectMessageReceivingMessageSourceTests {
source.setBeanFactory(bf);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
source.setTaskScheduler(scheduler);
//source.setTaskScheduler(scheduler);
source.setBeanName("twitterEndpoint");
source.afterPropertiesSet();
source.start();
//source.start();
Thread.sleep(1000);
Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
assertTrue(!CollectionUtils.isEmpty(msg));
@@ -171,7 +174,7 @@ public class DirectMessageReceivingMessageSourceTests {
Message message = source.receive();
Tweet tweet = (Tweet) message.getPayload();
assertEquals(2000, tweet.getId());
source.stop();
//source.stop();
Thread.sleep(3000);
@@ -187,10 +190,10 @@ public class DirectMessageReceivingMessageSourceTests {
source.setBeanFactory(bf);
scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
source.setTaskScheduler(scheduler);
//source.setTaskScheduler(scheduler);
source.setBeanName("twitterEndpoint");
source.afterPropertiesSet();
source.start();
//source.start();
Thread.sleep(1000);
msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
assertTrue(!CollectionUtils.isEmpty(msg));

59
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/RateLimitStatusTriggerTests.java
View File

@@ -1,59 +0,0 @@
/*
* 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 static junit.framework.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.Date;
import org.junit.Test;
import org.springframework.integration.twitter.core.TwitterOperations;
import org.springframework.scheduling.TriggerContext;
import twitter4j.RateLimitStatus;
import twitter4j.Twitter;
/**
* @author Oleg Zhurakousky
*
*/
public class RateLimitStatusTriggerTests {
@Test
public void testTriggerImediateAndSubsequentExecutionTime() throws Exception{
TwitterOperations twitter = mock(TwitterOperations.class);
Twitter tw = mock(Twitter.class);
when(twitter.getUnderlyingTwitter()).thenReturn(tw);
RateLimitStatusTrigger trigger = new RateLimitStatusTrigger(twitter.getUnderlyingTwitter());
TriggerContext context = mock(TriggerContext.class);
Date currentDate = new Date(System.currentTimeMillis());
Date nextDate = trigger.nextExecutionTime(context);
// as long as its within 1 msec we can consider it right away for the purpose of testing
assertTrue(nextDate.getTime() - currentDate.getTime() < 100);
RateLimitStatus rateLimitStatis = mock(RateLimitStatus.class);
when(tw.getRateLimitStatus()).thenReturn(rateLimitStatis);
when(rateLimitStatis.getRemainingHits()).thenReturn(2000);
when(rateLimitStatis.getSecondsUntilReset()).thenReturn(4000);
when(context.lastCompletionTime()).thenReturn(nextDate);
// based on the above values the next execution time should be at least 2000 msec
assertTrue(trigger.nextExecutionTime(context).getTime() - nextDate.getTime() > 2000);
}
}

4
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java
View File

@@ -56,9 +56,9 @@ public class SearchReceivingMessageSourceTests {
adapter.start();
ms.setBeanFactory(bf);
ms.setQuery("#springintegration");
ms.setTaskScheduler(scheduler);
//ms.setTaskScheduler(scheduler);
ms.afterPropertiesSet();
ms.start();
//ms.start();
System.in.read();
}

27
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSourceTests.java
View File

@@ -31,6 +31,7 @@ import java.util.Date;
import java.util.Queue;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
@@ -104,27 +105,28 @@ public class TimelineReceivingMessageSourceTests {
TimelineReceivingMessageSource source = new TimelineReceivingMessageSource(twitter);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
source.setTaskScheduler(scheduler);
//source.setTaskScheduler(scheduler);
source.setBeanName("twitterEndpoint");
source.afterPropertiesSet();
source.start();
//source.start();
assertEquals("twitter:inbound-channel-adapter.twitterEndpoint.kermit", TestUtils.getPropertyValue(source, "metadataKey"));
assertTrue(source.isRunning());
source.stop();
//assertTrue(source.isRunning());
//source.stop();
}
@SuppressWarnings("rawtypes")
@Test
@Ignore
public void testSuccessfulInitializationWithMessages() throws Exception{
this.setUpMockScenarioForMessagePolling();
TimelineReceivingMessageSource source = new TimelineReceivingMessageSource(twitter);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
source.setTaskScheduler(scheduler);
//source.setTaskScheduler(scheduler);
source.setBeanName("twitterEndpoint");
source.afterPropertiesSet();
source.start();
//source.start();
Thread.sleep(1000);
Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
assertTrue(!CollectionUtils.isEmpty(msg));
@@ -135,7 +137,7 @@ public class TimelineReceivingMessageSourceTests {
verify(twitter, times(1)).getTimeline(2000);
// based on the Mock, the Queue shoud now have 2 mopre messages third and fourth
assertTrue(((Queue)TestUtils.getPropertyValue(source, "tweets")).size() == 2);
source.stop();
//source.stop();
}
/**
* This test will validate that last status is initilaized from the metadatastore
@@ -143,6 +145,7 @@ public class TimelineReceivingMessageSourceTests {
*/
@SuppressWarnings("rawtypes")
@Test
@Ignore
public void testSuccessfulInitializationWithMessagesWithPersistentMetadata() throws Exception{
String fileName = System.getProperty("java.io.tmpdir") + "/spring-integration/metadata-store.properties";
File file = new File(fileName);
@@ -158,10 +161,10 @@ public class TimelineReceivingMessageSourceTests {
source.setBeanFactory(bf);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
source.setTaskScheduler(scheduler);
//source.setTaskScheduler(scheduler);
source.setBeanName("twitterEndpoint");
source.afterPropertiesSet();
source.start();
//source.start();
Thread.sleep(1000);
Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
assertTrue(!CollectionUtils.isEmpty(msg));
@@ -169,7 +172,7 @@ public class TimelineReceivingMessageSourceTests {
Message message = source.receive();
Tweet tweet = (Tweet) message.getPayload();
assertEquals(2000, tweet.getId());
source.stop();
//source.stop();
Thread.sleep(2000);
@@ -185,10 +188,10 @@ public class TimelineReceivingMessageSourceTests {
source.setBeanFactory(bf);
scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
source.setTaskScheduler(scheduler);
//source.setTaskScheduler(scheduler);
source.setBeanName("twitterEndpoint");
source.afterPropertiesSet();
source.start();
//source.start();
Thread.sleep(1000);
msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
assertTrue(!CollectionUtils.isEmpty(msg));