From 4f3b887dddb026f619ec85fc702b359bae039baa Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 11 Nov 2010 17:44:48 -0500 Subject: [PATCH] INT-1604 encapsulation --- .../inbound/AbstractTwitterMessageSource.java | 123 ++++++++++-------- .../DirectMessageReceivingMessageSource.java | 3 +- .../MentionsReceivingMessageSource.java | 2 +- .../inbound/SearchReceivingMessageSource.java | 2 +- .../TimelineReceivingMessageSource.java | 2 +- ...archReceivingMessageSourceParserTests.java | 25 +--- .../twitter/ignored/TwitterAnnouncer.java | 12 +- ...ectMessageReceivingMessageSourceTests.java | 4 +- .../SearchReceivingMessageSourceTests.java | 25 ++-- 9 files changed, 103 insertions(+), 95 deletions(-) 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 1514493969..de0daaef73 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 @@ -42,9 +42,8 @@ import org.springframework.util.StringUtils; * Abstract class that defines common operations for receiving various types of * messages when using the Twitter API. This class also handles keeping track of * the latest inbound message it has received and avoiding, where possible, - * redelivery of common messages. This functionality is enabled using the - * {@link org.springframework.integration.store.MetadataStore} - * strategy. + * redelivery of duplicate messages. This functionality is enabled using the + * {@link org.springframework.integration.store.MetadataStore} strategy. * * @author Josh Long * @author Oleg Zhurakousky @@ -58,25 +57,29 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i private volatile String metadataKey; - protected final Queue tweets = new LinkedBlockingQueue(); + private final Queue tweets = new LinkedBlockingQueue(); - protected volatile int prefetchThreshold = 0; + private volatile int prefetchThreshold = 0; - protected volatile long markerId = -1; + private volatile long markerId = -1; - protected volatile long processedId = -1; + //private volatile long processedId = -1; - protected final TwitterOperations twitter; + private final TwitterOperations twitterOperations; - private volatile ScheduledFuture twitterUpdatePollingTask; + private final TweetComparator tweetComparator = new TweetComparator(); + + private volatile ScheduledFuture twitterPollingTask; private final Object markerGuard = new Object(); - public AbstractTwitterMessageSource(TwitterOperations twitter){ - this.twitter = twitter; + public AbstractTwitterMessageSource(TwitterOperations twitterOperations) { + Assert.notNull(twitterOperations, "twitterOperations must not be null"); + this.twitterOperations = twitterOperations; } + public long getMarkerId() { return this.markerId; } @@ -85,20 +88,20 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i return this.markerId > -1; } + protected TwitterOperations getTwitterOperations() { + return this.twitterOperations; + } + @Override protected void onInit() throws Exception{ Assert.notNull(this.getTaskScheduler(), - "Can not locate TaskScheduler. You must inject one explicitly or define a bean by the name 'taskScheduler'"); + "Unable to locate TaskScheduler. You must inject one explicitly or define a bean by the name 'taskScheduler'."); super.onInit(); - if (this.metadataStore == null) { - // first try to look for a 'messageStore' in the context + // first try to look for a 'metadataStore' in the context BeanFactory beanFactory = this.getBeanFactory(); if (beanFactory != null) { - MetadataStore metadataStore = IntegrationContextUtils.getMetadataStore(beanFactory); - if (metadataStore != null) { - this.metadataStore = metadataStore; - } + this.metadataStore = IntegrationContextUtils.getMetadataStore(beanFactory); } if (this.metadataStore == null) { this.metadataStore = new SimpleMetadataStore(); @@ -114,47 +117,18 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i else if (logger.isWarnEnabled()) { logger.warn(this.getClass().getSimpleName() + " has no name. MetadataStore key might not be unique."); } - String profileId = twitter.getProfileId(); - metadataKeyBuilder.append(profileId); + String profileId = this.twitterOperations.getProfileId(); + if (profileId != null) { + metadataKeyBuilder.append(profileId); + } this.metadataKey = metadataKeyBuilder.toString(); String lastId = this.metadataStore.get(this.metadataKey); // initialize the last status ID from the metadataStore - if (StringUtils.hasText(lastId)){ + if (StringUtils.hasText(lastId)) { this.markerId = Long.parseLong(lastId); } } - @SuppressWarnings("unchecked") - protected void forwardAll(List tResponses) { - Collections.sort(tResponses, this.getComparator()); - for (Tweet twitterResponse : tResponses) { - forward(twitterResponse); - } - } - - private Comparator getComparator() { - return new Comparator() { - public int compare(Tweet tweet1, Tweet tweet2) { - return tweet1.getCreatedAt().compareTo(tweet2.getCreatedAt()); - } - }; - } - - @Override - protected void doStart(){ - 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 twitterPollingTask = new TwitterPollingTask(); - twitterUpdatePollingTask = this.getTaskScheduler().schedule(twitterPollingTask, trigger); - } - - @Override - protected void doStop(){ - twitterUpdatePollingTask.cancel(true); - } - public Message receive() { Tweet tweet = this.tweets.poll(); if (tweet != null) { @@ -164,27 +138,54 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i return null; } - protected void forward(Tweet tweet) { + @SuppressWarnings("unchecked") + private void enqueueAll(List tweets) { + Collections.sort(tweets, this.tweetComparator); + for (Tweet tweet : tweets) { + enqueue(tweet); + } + } + + private void enqueue(Tweet tweet) { synchronized (this.markerGuard) { long id = tweet.getId(); if (id > this.markerId) { this.markerId = id; - tweets.add(tweet); + this.tweets.add(tweet); } } } - protected void markProcessedId(long statusId) { - this.processedId = statusId; + private void markProcessedId(long statusId) { + //this.processedId = statusId; this.metadataStore.put(this.metadataKey, String.valueOf(statusId)); } + /** * Subclasses must implement this to return tweets. */ protected abstract List pollForTweets(); + // 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() { @@ -192,7 +193,7 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i if (tweets.size() <= prefetchThreshold) { List tweets = pollForTweets(); if (!CollectionUtils.isEmpty(tweets)) { - forwardAll(tweets); + enqueueAll(tweets); } } } @@ -205,4 +206,12 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint i } } + + private static class TweetComparator implements Comparator { + + public int compare(Tweet tweet1, Tweet tweet2) { + return tweet1.getCreatedAt().compareTo(tweet2.getCreatedAt()); + } + } + } 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 2895cc8ebf..5511e0e8be 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 @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.twitter.inbound; import java.util.List; @@ -43,7 +44,7 @@ public class DirectMessageReceivingMessageSource extends AbstractTwitterMessageS @Override protected List pollForTweets() { long sinceId = getMarkerId(); - return hasMarkedStatus() ? twitter.getDirectMessages(sinceId) : twitter.getDirectMessages(); + return hasMarkedStatus() ? 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 982612b908..a95c6bd72b 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 @@ -42,7 +42,7 @@ public class MentionsReceivingMessageSource extends AbstractTwitterMessageSource @Override protected List pollForTweets() { long sinceId = getMarkerId(); - return hasMarkedStatus() ? twitter.getMentions(sinceId) : twitter.getMentions(); + return hasMarkedStatus() ? 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 800011c315..b463aef86c 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 @@ -50,7 +50,7 @@ public class SearchReceivingMessageSource extends AbstractTwitterMessageSource pollForTweets() { - SearchResults results = this.twitter.search(query); + SearchResults results = 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 6622ab5ec6..500b45864f 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 @@ -44,7 +44,7 @@ public class TimelineReceivingMessageSource extends AbstractTwitterMessageSource @Override protected List pollForTweets() { long sinceId = getMarkerId(); - return hasMarkedStatus() ? twitter.getTimeline(sinceId) : twitter.getTimeline(); + return hasMarkedStatus() ? this.getTwitterOperations().getTimeline(sinceId) : this.getTwitterOperations().getTimeline(); } } diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestSearchReceivingMessageSourceParserTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestSearchReceivingMessageSourceParserTests.java index 314a0a1389..81311e0cda 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestSearchReceivingMessageSourceParserTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestSearchReceivingMessageSourceParserTests.java @@ -13,32 +13,24 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.twitter.config; import static junit.framework.Assert.assertEquals; import static junit.framework.Assert.assertFalse; -import static junit.framework.Assert.assertTrue; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import java.lang.reflect.Proxy; - -import net.sf.cglib.proxy.Enhancer; import org.junit.Test; -import org.springframework.beans.factory.FactoryBean; + import org.springframework.context.ApplicationContext; 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.Twitter4jTemplate; import org.springframework.integration.twitter.core.TwitterOperations; -import org.springframework.integration.twitter.inbound.AbstractTwitterMessageSource; import org.springframework.integration.twitter.inbound.SearchReceivingMessageSource; /** * @author Oleg Zhurakousky - * */ public class TestSearchReceivingMessageSourceParserTests { @@ -48,23 +40,20 @@ public class TestSearchReceivingMessageSourceParserTests { SourcePollingChannelAdapter spca = ac.getBean("searchAdapter", SourcePollingChannelAdapter.class); SearchReceivingMessageSource ms = (SearchReceivingMessageSource) TestUtils.getPropertyValue(spca, "source"); assertFalse(ms.isAutoStartup()); - assertFalse(ms.isAutoStartup()); - Twitter4jTemplate template = (Twitter4jTemplate) TestUtils.getPropertyValue(ms, "twitter"); - - assertFalse(template.getUnderlyingTwitter().isOAuthEnabled()); // verify that anonymous Twitte + Twitter4jTemplate template = (Twitter4jTemplate) TestUtils.getPropertyValue(ms, "twitterOperations"); + assertFalse(template.getUnderlyingTwitter().isOAuthEnabled()); // verify anonymous Twitter } + @Test public void testSearchReceivingCustomTemplate(){ 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()); - TwitterOperations template = (TwitterOperations) TestUtils.getPropertyValue(ms, "twitter"); + TwitterOperations template = (TwitterOperations) TestUtils.getPropertyValue(ms, "twitterOperations"); assertEquals(ac.getBean("twitter"), template); } - - + } diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TwitterAnnouncer.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TwitterAnnouncer.java index 3548faffea..65b593646b 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TwitterAnnouncer.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TwitterAnnouncer.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.twitter.ignored; import org.springframework.integration.Message; @@ -20,28 +21,29 @@ import org.springframework.integration.history.MessageHistory; import org.springframework.integration.twitter.core.Tweet; import org.springframework.stereotype.Component; - @Component public class TwitterAnnouncer { + public void dm(Tweet directMessage) { System.out.println("A direct message has been received from " + directMessage.getFromUser() + " with text " + directMessage.getText()); } - - public void search(Message search) { + + public void search(Message search) { MessageHistory history = MessageHistory.read(search); System.out.println(history); Tweet tweet = (Tweet) search.getPayload(); - System.out.println("A search item was received " + tweet.getCreatedAt() + " with text " + tweet.getText()); } public void mention(Tweet s) { - System.out.println("A tweet mentioning (or replying) to " + "you was received having text " + s.getFromUser() + "-" + s.getText() + " from " + s.getSource()); + System.out.println("A tweet mentioning (or replying) to you was received having text " + + s.getFromUser() + "-" + s.getText() + " from " + s.getSource()); } public void updates(Tweet t) { System.out.println("Received timeline update: " + t.getText() + " from " + t.getSource()); } + } diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java index c48a4da6c8..b0f6084343 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java @@ -134,12 +134,12 @@ public class DirectMessageReceivingMessageSourceTests { assertEquals(2000, message.getId()); Thread.sleep(1000); verify(twitter, times(1)).getDirectMessages(); - // based on the Mock, the Queue shoud now have 1 more messages third and fourth + // 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(); } /** - * This test will validate that last status is initilaized from the metadatastore + * This test will validate that last status is initialized from the metadatastore * @throws Exception */ @SuppressWarnings("rawtypes") 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 index bb14791b3b..aa8ccc8305 100644 --- 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 @@ -1,6 +1,19 @@ -/** - * +/* + * 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 org.junit.Ignore; @@ -8,7 +21,6 @@ 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; @@ -19,7 +31,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author ozhurakousky - * */ public class SearchReceivingMessageSourceTests { @@ -30,9 +41,7 @@ public class SearchReceivingMessageSourceTests { 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 { @@ -45,14 +54,12 @@ public class SearchReceivingMessageSourceTests { adapter.setOutputChannel(channel); adapter.afterPropertiesSet(); adapter.start(); - ms.setBeanFactory(bf); ms.setQuery("#springintegration"); - ms.setTaskScheduler(scheduler); ms.afterPropertiesSet(); ms.start(); - System.in.read(); } + }