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 21858cd911..21aa4f7bcf 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 @@ -22,7 +22,6 @@ import org.springframework.util.Assert; import twitter4j.DirectMessage; import twitter4j.Paging; -import twitter4j.RateLimitStatus; import twitter4j.ResponseList; import twitter4j.Status; import twitter4j.StatusUpdate; @@ -204,12 +203,13 @@ public class Twitter4jTemplate implements TwitterOperations{ private Tweet buildTweetFromStatus(Status status){ Tweet tweet = new Tweet(); tweet.setCreatedAt(status.getCreatedAt()); - tweet.setFromUser(status.getInReplyToScreenName()); - tweet.setFromUserId(status.getInReplyToUserId()); + if (status.getUser() != null){ + tweet.setFromUser(status.getUser().getScreenName()); + tweet.setFromUserId(status.getUser().getId()); + } tweet.setId(status.getId()); tweet.setSource(status.getSource()); tweet.setText(status.getText()); - tweet.setToUserId((long)status.getUser().getId()); return tweet; } } 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 213ecd3528..f9f29b3cc2 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 @@ -34,14 +34,10 @@ import org.springframework.integration.store.MetadataStore; import org.springframework.integration.store.SimpleMetadataStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.twitter.core.Tweet; -import org.springframework.integration.twitter.core.Twitter4jTemplate; import org.springframework.integration.twitter.core.TwitterOperations; import org.springframework.util.Assert; import org.springframework.util.StringUtils; -import twitter4j.DirectMessage; -import twitter4j.Status; - /** * Abstract class that defines common operations for receiving various types of * messages when using the Twitter API. This class also handles keeping track of @@ -125,6 +121,11 @@ public abstract class AbstractTwitterMessageSource extends AbstractEndpoint String profileId = twitter.getProfileId(); 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)){ + this.markLastStatusId(Long.parseLong(lastId)); + } } @SuppressWarnings("unchecked") 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 14eb348a89..4c283f3b30 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 @@ -23,8 +23,6 @@ import org.springframework.integration.twitter.core.Tweet; import org.springframework.integration.twitter.core.TwitterOperations; import org.springframework.util.CollectionUtils; -import twitter4j.Paging; - /** * This class handles support for receiving DMs (direct messages) using Twitter. * diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionReceivingMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionReceivingMessageSource.java index d0f54b38c0..28f41e8fd8 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionReceivingMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionReceivingMessageSource.java @@ -21,8 +21,6 @@ import org.springframework.integration.MessagingException; import org.springframework.integration.twitter.core.Tweet; import org.springframework.integration.twitter.core.TwitterOperations; -import twitter4j.Paging; - /** * Handles forwarding all new {@link twitter4j.Status} that are 'replies' or 'mentions' to some other tweet. * diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSource.java index 69b953e985..ab4c592c00 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSource.java @@ -21,8 +21,6 @@ import org.springframework.integration.MessagingException; import org.springframework.integration.twitter.core.Tweet; import org.springframework.integration.twitter.core.TwitterOperations; -import twitter4j.Paging; - /** * This {@link org.springframework.integration.core.MessageSource} lets Spring Integration consume a given account's timeline diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/outbound/OutboundTweetMessageMapper.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/outbound/OutboundTweetMessageMapper.java index d0d330d73d..3d1231efb5 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/outbound/OutboundTweetMessageMapper.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/outbound/OutboundTweetMessageMapper.java @@ -45,30 +45,13 @@ public class OutboundTweetMessageMapper implements OutboundMessageMapper Tweet tweet = null; if (payload instanceof String) { tweet = new Tweet(); + tweet.setText((String) payload); if (message.getHeaders().containsKey(TwitterHeaders.IN_REPLY_TO_STATUS_ID)) { Long replyId = (Long) message.getHeaders().get(TwitterHeaders.IN_REPLY_TO_STATUS_ID); if ((replyId != null) && (replyId > 0)) { tweet.setToUserId(replyId); } } -// if (message.getHeaders().containsKey(TwitterHeaders.PLACE_ID)) { -// String placeId = (String) message.getHeaders().get(TwitterHeaders.PLACE_ID); -// if (StringUtils.hasText(placeId)) { -// statusUpdate.placeId(placeId); -// } -// } -// if (message.getHeaders().containsKey(TwitterHeaders.GEOLOCATION)) { -// GeoLocation geoLocation = (GeoLocation) message.getHeaders().get(TwitterHeaders.GEOLOCATION); -// if (null != geoLocation) { -// statusUpdate.location(geoLocation); -// } -// } -// if (message.getHeaders().containsKey(TwitterHeaders.DISPLAY_COORDINATES)) { -// Boolean displayCoords = (Boolean) message.getHeaders().get(TwitterHeaders.DISPLAY_COORDINATES); -// if (displayCoords != null) { -// statusUpdate.displayCoordinates(displayCoords); -// } -// } } else if (payload instanceof Tweet) { tweet = (Tweet) payload; diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingMessageSourceParserTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingMessageSourceParserTests.java index 687c11d210..cb080d0be2 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingMessageSourceParserTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingMessageSourceParserTests.java @@ -25,7 +25,6 @@ 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; diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestReceivingUsingNamespace-context.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestReceivingUsingNamespace-context.xml index eb09db03cc..b08710a045 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestReceivingUsingNamespace-context.xml +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestReceivingUsingNamespace-context.xml @@ -39,15 +39,15 @@ - - - - + + + + - - - - + + + + diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingDMsUsingNamespace-context.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingDMsUsingNamespace-context.xml index 69cd62b3cd..32e6ff2dfd 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingDMsUsingNamespace-context.xml +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingDMsUsingNamespace-context.xml @@ -21,17 +21,17 @@ location="classpath:twitter.sender.properties" ignore-unresolvable="true"/> - + + + + + + - + diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingUpdatesUsingNamespace-context.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingUpdatesUsingNamespace-context.xml index d076b3dcea..a9bcdf6b3f 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingUpdatesUsingNamespace-context.xml +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingUpdatesUsingNamespace-context.xml @@ -16,27 +16,23 @@ http://www.springframework.org/schema/tool http://www.springframework.org/schema/tool/spring-tool-3.0.xsd http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-3.0.xsd http://www.springframework.org/schema/integration/twitter http://www.springframework.org/schema/integration/twitter/spring-integration-twitter.xsd"> - - + - + + + + + + - - - + + diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingUpdatesUsingNamespace.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingUpdatesUsingNamespace.java index 841aa1924a..0577a635de 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingUpdatesUsingNamespace.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/ignored/TestSendingUpdatesUsingNamespace.java @@ -16,6 +16,8 @@ package org.springframework.integration.twitter.ignored; +import java.util.Date; + import org.junit.Ignore; import org.junit.Test; import org.springframework.beans.factory.annotation.Value; @@ -40,9 +42,8 @@ public class TestSendingUpdatesUsingNamespace extends AbstractJUnit4SpringContex @Test @Ignore public void testSendingATweet() throws Throwable { - MessageBuilder mb = MessageBuilder.withPayload("simple test demonstrating the ability to encode location information") + MessageBuilder mb = MessageBuilder.withPayload("Aligning #springintegration Twitter adapter with #SpringSocial API - " + new Date(System.currentTimeMillis())) .setHeader(TwitterHeaders.IN_REPLY_TO_STATUS_ID, 21927437001L) - //.setHeader(TwitterHeaders.TWITTER_GEOLOCATION, new Twitter4jGeoLocation(-76.226823, 23.642465)) // antarctica .setHeader(TwitterHeaders.DISPLAY_COORDINATES, true); Message m = mb.build(); this.messagingTemplate.send(this.channel, m); 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 bea91ed01b..df10888fcd 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 @@ -27,7 +27,7 @@ public class TwitterAnnouncer { } public void mention(Tweet s) { - System.out.println("A tweet mentioning (or replying) to " + "you was received having text " + 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) { 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 b1d433ce0a..65f4e5ffb5 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 @@ -19,20 +19,26 @@ package org.springframework.integration.twitter.inbound; import static junit.framework.Assert.assertEquals; import static junit.framework.Assert.assertTrue; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.File; +import java.lang.reflect.Field; import java.util.ArrayList; -import java.util.Collection; import java.util.Date; import java.util.Queue; import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.store.PropertiesPersistingMetadataStore; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.twitter.core.Tweet; +import org.springframework.integration.twitter.core.Twitter4jTemplate; import org.springframework.integration.twitter.core.TwitterOperations; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.CollectionUtils; @@ -42,19 +48,20 @@ import twitter4j.Paging; import twitter4j.RateLimitStatus; import twitter4j.ResponseList; import twitter4j.Twitter; +import twitter4j.TwitterFactory; /** * @author Oleg Zhurakousky */ public class DirectMessageReceivingMessageSourceTests { - private Tweet firstMessage; + private DirectMessage firstMessage; - private Tweet secondMessage; + private DirectMessage secondMessage; - private Tweet thirdMessage; + private DirectMessage thirdMessage; - private Tweet fourthMessage; + private DirectMessage fourthMessage; private TwitterOperations twitter; @@ -63,25 +70,31 @@ public class DirectMessageReceivingMessageSourceTests { @Before public void prepare() throws Exception{ - twitter = mock(TwitterOperations.class); - firstMessage = mock(Tweet.class); + twitter = new Twitter4jTemplate(); + + firstMessage = mock(DirectMessage.class); when(firstMessage.getCreatedAt()).thenReturn(new Date(5555555555L)); - when(firstMessage.getId()).thenReturn((long) 200); - secondMessage = mock(Tweet.class); + when(firstMessage.getId()).thenReturn( 200); + + secondMessage = mock(DirectMessage.class); when(secondMessage.getCreatedAt()).thenReturn(new Date(2222222222L)); - when(secondMessage.getId()).thenReturn((long) 2000); + when(secondMessage.getId()).thenReturn( 2000); - thirdMessage = mock(Tweet.class); + thirdMessage = mock(DirectMessage.class); when(thirdMessage.getCreatedAt()).thenReturn(new Date(66666666666L)); - when(thirdMessage.getId()).thenReturn((long) 3000); + when(thirdMessage.getId()).thenReturn( 3000); - fourthMessage = mock(Tweet.class); + fourthMessage = mock(DirectMessage.class); when(fourthMessage.getCreatedAt()).thenReturn(new Date(77777777777L)); - when(fourthMessage.getId()).thenReturn((long) 4000); + when(fourthMessage.getId()).thenReturn( 4000); - when(twitter.getProfileId()).thenReturn("kermit"); - Twitter tw = mock(Twitter.class); - when(twitter.getUnderlyingTwitter()).thenReturn(tw); + tw = mock(Twitter.class); + Field twField = Twitter4jTemplate.class.getDeclaredField("twitter"); + twField.setAccessible(true); + twField.set(twitter, tw); + when(tw.getScreenName()).thenReturn("kermit"); + + twitter = spy(twitter); } @@ -97,6 +110,7 @@ public class DirectMessageReceivingMessageSourceTests { source.start(); assertEquals("twitter:inbound-dm-channel-adapter.twitterEndpoint.kermit", TestUtils.getPropertyValue(source, "metadataKey")); assertTrue(source.isRunning()); + source.stop(); } @SuppressWarnings("rawtypes") @@ -112,24 +126,83 @@ public class DirectMessageReceivingMessageSourceTests { source.afterPropertiesSet(); source.start(); Thread.sleep(1000); - System.out.println("Tweets: " + TestUtils.getPropertyValue(source, "tweets")); Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets"); assertTrue(!CollectionUtils.isEmpty(msg)); assertEquals(1, msg.size()); // because the other message has a older timestamp and is assumed to be read by Tweet message = (Tweet) msg.poll(); - assertEquals(secondMessage, message); + assertEquals(2000, message.getId()); Thread.sleep(1000); verify(twitter, times(1)).getDirectMessages(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(); + } + /** + * This test will validate that last status is initilaized from the metadatastore + * @throws Exception + */ + @SuppressWarnings("rawtypes") + @Test + public void testSuccessfullInitializationWithMessagesWithPersistentMetadata() throws Exception{ + String fileName = System.getProperty("java.io.tmpdir") + "/spring-integration/metadata-store.properties"; + File file = new File(fileName); + if (file.exists()){ + file.delete(); + } + this.setUpMockScenarioForMessagePolling(); + DefaultListableBeanFactory bf = new DefaultListableBeanFactory(); + PropertiesPersistingMetadataStore store = new PropertiesPersistingMetadataStore(); + store.afterPropertiesSet(); + bf.registerSingleton(IntegrationContextUtils.METADATA_STORE_BEAN_NAME, store); + DirectMessageReceivingMessageSource source = new DirectMessageReceivingMessageSource(twitter); + source.setBeanFactory(bf); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + source.setTaskScheduler(scheduler); + source.setBeanName("twitterEndpoint"); + source.afterPropertiesSet(); + source.start(); + Thread.sleep(1000); + Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets"); + assertTrue(!CollectionUtils.isEmpty(msg)); + assertEquals(1, msg.size()); // because the other message has a older timestamp and is assumed to be read by + Tweet message = (Tweet) msg.poll(); + assertEquals(2000, message.getId()); + source.stop(); + Thread.sleep(3000); + + + // Resuming + this.prepare(); + this.setUpMockScenarioForMessagePolling(); + store.destroy(); + bf = new DefaultListableBeanFactory(); + store = new PropertiesPersistingMetadataStore(); + store.afterPropertiesSet(); + bf.registerSingleton(IntegrationContextUtils.METADATA_STORE_BEAN_NAME, store); + source = new DirectMessageReceivingMessageSource(twitter); + source.setBeanFactory(bf); + scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + source.setTaskScheduler(scheduler); + source.setBeanName("twitterEndpoint"); + source.afterPropertiesSet(); + source.start(); + Thread.sleep(1000); + msg = (Queue) TestUtils.getPropertyValue(source, "tweets"); + assertTrue(!CollectionUtils.isEmpty(msg)); + message = (Tweet) msg.poll(); + assertEquals(3000, message.getId()); + message = (Tweet) msg.poll(); + assertEquals(4000, message.getId()); + file.delete(); } @SuppressWarnings("unchecked") private void setUpMockScenarioForMessagePolling() throws Exception{ RateLimitStatus rateLimitStatus = mock(RateLimitStatus.class); - Twitter tw = mock(Twitter.class); - when(twitter.getUnderlyingTwitter()).thenReturn(tw); + when(tw.getRateLimitStatus()).thenReturn(rateLimitStatus); when(rateLimitStatus.getSecondsUntilReset()).thenReturn(1000); when(rateLimitStatus.getRemainingHits()).thenReturn(1000); @@ -137,12 +210,12 @@ public class DirectMessageReceivingMessageSourceTests { SampleResoponceList testMessages = new SampleResoponceList(); testMessages.add(firstMessage); testMessages.add(secondMessage); - when(twitter.getDirectMessages()).thenReturn(testMessages); + when(tw.getDirectMessages()).thenReturn(testMessages); testMessages = new SampleResoponceList(); testMessages.add(thirdMessage); testMessages.add(fourthMessage); - when(twitter.getDirectMessages((long)2000)).thenReturn(testMessages); + when(tw.getDirectMessages(Mockito.any(Paging.class))).thenReturn(testMessages); } @SuppressWarnings({ "rawtypes", "serial" }) diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSourceTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSourceTests.java new file mode 100644 index 0000000000..a585fd6372 --- /dev/null +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSourceTests.java @@ -0,0 +1,234 @@ +/* + * 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.assertEquals; +import static junit.framework.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.Date; +import java.util.Queue; + +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.store.PropertiesPersistingMetadataStore; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.twitter.core.Tweet; +import org.springframework.integration.twitter.core.Twitter4jTemplate; +import org.springframework.integration.twitter.core.TwitterOperations; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.CollectionUtils; + +import twitter4j.Paging; +import twitter4j.RateLimitStatus; +import twitter4j.ResponseList; +import twitter4j.Status; +import twitter4j.Twitter; + +/** + * @author Oleg Zhurakousky + */ +public class TimelineUpdateReceivingMessageSourceTests { + + private Status firstMessage; + + private Status secondMessage; + + private Status thirdMessage; + + private Status fourthMessage; + + private TwitterOperations twitter; + + Twitter tw; + + + @Before + public void prepare() throws Exception{ + twitter = new Twitter4jTemplate(); + + firstMessage = mock(Status.class); + when(firstMessage.getCreatedAt()).thenReturn(new Date(5555555555L)); + when(firstMessage.getId()).thenReturn( (long) 200); + + secondMessage = mock(Status.class); + when(secondMessage.getCreatedAt()).thenReturn(new Date(2222222222L)); + when(secondMessage.getId()).thenReturn((long) 2000); + + thirdMessage = mock(Status.class); + when(thirdMessage.getCreatedAt()).thenReturn(new Date(66666666666L)); + when(thirdMessage.getId()).thenReturn((long) 3000); + + fourthMessage = mock(Status.class); + when(fourthMessage.getCreatedAt()).thenReturn(new Date(77777777777L)); + when(fourthMessage.getId()).thenReturn( (long)4000); + + tw = mock(Twitter.class); + Field twField = Twitter4jTemplate.class.getDeclaredField("twitter"); + twField.setAccessible(true); + twField.set(twitter, tw); + when(tw.getScreenName()).thenReturn("kermit"); + + twitter = spy(twitter); + } + + + @Test + public void testSuccessfullInitialization() throws Exception{ + + TimelineUpdateReceivingMessageSource source = new TimelineUpdateReceivingMessageSource(twitter); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + source.setTaskScheduler(scheduler); + source.setBeanName("twitterEndpoint"); + source.afterPropertiesSet(); + source.start(); + assertEquals("twitter:inbound-update-channel-adapter.twitterEndpoint.kermit", TestUtils.getPropertyValue(source, "metadataKey")); + assertTrue(source.isRunning()); + source.stop(); + } + + @SuppressWarnings("rawtypes") + @Test + public void testSuccessfullInitializationWithMessages() throws Exception{ + this.setUpMockScenarioForMessagePolling(); + + TimelineUpdateReceivingMessageSource source = new TimelineUpdateReceivingMessageSource(twitter); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + source.setTaskScheduler(scheduler); + source.setBeanName("twitterEndpoint"); + source.afterPropertiesSet(); + source.start(); + Thread.sleep(1000); + Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets"); + assertTrue(!CollectionUtils.isEmpty(msg)); + assertEquals(1, msg.size()); // because the other message has a older timestamp and is assumed to be read by + Tweet message = (Tweet) msg.poll(); + assertEquals(2000, message.getId()); + Thread.sleep(1000); + verify(twitter, times(1)).getFriendsTimeline(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(); + } + /** + * This test will validate that last status is initilaized from the metadatastore + * @throws Exception + */ + @SuppressWarnings("rawtypes") + @Test + public void testSuccessfullInitializationWithMessagesWithPersistentMetadata() throws Exception{ + String fileName = System.getProperty("java.io.tmpdir") + "/spring-integration/metadata-store.properties"; + File file = new File(fileName); + if (file.exists()){ + file.delete(); + } + this.setUpMockScenarioForMessagePolling(); + DefaultListableBeanFactory bf = new DefaultListableBeanFactory(); + PropertiesPersistingMetadataStore store = new PropertiesPersistingMetadataStore(); + store.afterPropertiesSet(); + bf.registerSingleton(IntegrationContextUtils.METADATA_STORE_BEAN_NAME, store); + TimelineUpdateReceivingMessageSource source = new TimelineUpdateReceivingMessageSource(twitter); + source.setBeanFactory(bf); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + source.setTaskScheduler(scheduler); + source.setBeanName("twitterEndpoint"); + source.afterPropertiesSet(); + source.start(); + Thread.sleep(1000); + Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets"); + assertTrue(!CollectionUtils.isEmpty(msg)); + assertEquals(1, msg.size()); // because the other message has a older timestamp and is assumed to be read by + Tweet message = (Tweet) msg.poll(); + assertEquals(2000, message.getId()); + source.stop(); + Thread.sleep(3000); + + + // Resuming + this.prepare(); + this.setUpMockScenarioForMessagePolling(); + store.destroy(); + bf = new DefaultListableBeanFactory(); + store = new PropertiesPersistingMetadataStore(); + store.afterPropertiesSet(); + bf.registerSingleton(IntegrationContextUtils.METADATA_STORE_BEAN_NAME, store); + source = new TimelineUpdateReceivingMessageSource(twitter); + source.setBeanFactory(bf); + scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + source.setTaskScheduler(scheduler); + source.setBeanName("twitterEndpoint"); + source.afterPropertiesSet(); + source.start(); + Thread.sleep(1000); + msg = (Queue) TestUtils.getPropertyValue(source, "tweets"); + assertTrue(!CollectionUtils.isEmpty(msg)); + message = (Tweet) msg.poll(); + assertEquals(3000, message.getId()); + message = (Tweet) msg.poll(); + assertEquals(4000, message.getId()); + file.delete(); + } + + + @SuppressWarnings("unchecked") + private void setUpMockScenarioForMessagePolling() throws Exception{ + RateLimitStatus rateLimitStatus = mock(RateLimitStatus.class); + + when(tw.getRateLimitStatus()).thenReturn(rateLimitStatus); + when(rateLimitStatus.getSecondsUntilReset()).thenReturn(1000); + when(rateLimitStatus.getRemainingHits()).thenReturn(1000); + + SampleResoponceList testMessages = new SampleResoponceList(); + testMessages.add(firstMessage); + testMessages.add(secondMessage); + when(tw.getFriendsTimeline()).thenReturn(testMessages); + + testMessages = new SampleResoponceList(); + testMessages.add(thirdMessage); + testMessages.add(fourthMessage); + when(tw.getFriendsTimeline(Mockito.any(Paging.class))).thenReturn(testMessages); + } + + @SuppressWarnings({ "rawtypes", "serial" }) + public static class SampleResoponceList extends ArrayList implements ResponseList { + + @Override + public RateLimitStatus getRateLimitStatus() { + return mock(RateLimitStatus.class); + } + + @Override + public RateLimitStatus getFeatureSpecificRateLimitStatus() { + return mock(RateLimitStatus.class); + } + + } +}