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

INT-3147: (3167,3173,1941) Improve MetadataStore

Browse Source 
Previously, `MetadataStore` couldn't be configured for Twitter Adapters
- only a global one could be used.
The `metadataKey` was generated automatically with a 'difficult' value.

* Register all `MessageSource` for `SourcePollingChannelAdapter`
as beans with id based on adapter id and prefix '.source' (INT-3147)
* Polishing parser to get rid of explicit `MessageSource` beans. (INT-3147)
* Make Feed and Twitter adapters `id` attribute as required -
now it presents a `metadataKey` for `MetadataStore` (INT-3147)
* Add to Twitter adapters a reference attribute for `MetadataStore` (INT-3173)
* Add Twitter adapters `poll-skip-period` attribute (INT-3167)
* Add and implement `MetadataStore#remove` (INT-1941)
* Make `MetadataStore` as `@ManagedResource` (INT-1941)
* Polishing tests

JIRAs:
https://jira.springsource.org/browse/INT-3147
https://jira.springsource.org/browse/INT-3167
https://jira.springsource.org/browse/INT-3173
https://jira.springsource.org/browse/INT-1941

INT-3147: Polishing and fixes

* add domain suffix to `metadataKey`
* change contract of `MetadataStore.remove`
* remove timeout window from `AbstractTwitterMessageSource`
* polishing and fix `SearchReceivingMessageSourceWithRedisTests`

INT-3147: Rebasing and polishing

INT-3147: fix 'metadata' package tangle

INT-3147 Doc Polishing
This commit is contained in:
Artem Bilan
2013-10-03 11:27:36 -04:00
committed by Gary Russell
parent 1fb838dd1a
commit 5be8ef3fd8
40 changed files with 393 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -32,7 +32,7 @@ import org.springframework.util.StringUtils;
/**
* Parser for inbound Twitter Channel Adapters.
*
*
* @author Oleg Zhurakousky
* @since 2.0
*/
@@ -50,7 +50,10 @@ public class TwitterInboundChannelAdapterParser extends AbstractPollingInboundCh
BeanDefinitionBuilder templateBuilder = BeanDefinitionBuilder.genericBeanDefinition(TwitterTemplate.class);
builder.addConstructorArgValue(templateBuilder.getBeanDefinition());
}
builder.addConstructorArgValue(element.getAttribute(ID_ATTRIBUTE));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "query");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "metadata-store");
return builder.getBeanDefinition();
}

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

@@ -28,9 +28,11 @@ 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.store.metadata.MetadataStore;
import org.springframework.integration.store.metadata.SimpleMetadataStore;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.social.twitter.api.DirectMessage;
import org.springframework.social.twitter.api.Tweet;
import org.springframework.social.twitter.api.Twitter;
@@ -44,24 +46,29 @@ import org.springframework.util.StringUtils;
* 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 duplicate messages. This functionality is enabled using the
* {@link org.springframework.integration.store.MetadataStore} strategy.
* {@link org.springframework.integration.metadata.MetadataStore} strategy.
*
* @author Josh Long
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Gunnar Hillert
* @author Artem Bilan
*
* @since 2.0
*/
@SuppressWarnings("rawtypes")
abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport implements MessageSource {
private volatile long lastPollForTweet;
private final Twitter twitter;
private final TweetComparator tweetComparator = new TweetComparator();
private final Object lastEnqueuedIdMonitor = new Object();
private final String metadataKey;
private volatile MetadataStore metadataStore;
private volatile String metadataKey;
private final Queue<T> tweets = new LinkedBlockingQueue<T>();
private volatile int prefetchThreshold = 0;
@@ -70,25 +77,36 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
private volatile long lastProcessedId = -1;
private final Twitter twitter;
private final TweetComparator tweetComparator = new TweetComparator();
private final Object lastEnqueuedIdMonitor = new Object();
public AbstractTwitterMessageSource(Twitter twitter) {
public AbstractTwitterMessageSource(Twitter twitter, String metadataKey) {
Assert.notNull(twitter, "twitter must not be null");
Assert.notNull(metadataKey, "metadataKey must not be null");
this.twitter = twitter;
if (this.twitter.isAuthorized()){
UserOperations userOperations = this.twitter.userOperations();
String profileId = String.valueOf(userOperations.getProfileId());
if (profileId != null) {
metadataKey += "." + profileId;
}
}
this.metadataKey = metadataKey;
}
public void setMetadataStore(MetadataStore metadataStore) {
this.metadataStore = metadataStore;
}
public void setPrefetchThreshold(int prefetchThreshold) {
this.prefetchThreshold = prefetchThreshold;
}
protected Twitter getTwitter() {
return this.twitter;
}
@Override
protected void onInit() throws Exception{
protected void onInit() throws Exception {
super.onInit();
if (this.metadataStore == null) {
// first try to look for a 'metadataStore' in the context
@@ -100,26 +118,7 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
this.metadataStore = new SimpleMetadataStore();
}
}
StringBuilder metadataKeyBuilder = new StringBuilder();
if (StringUtils.hasText(this.getComponentType())) {
metadataKeyBuilder.append(this.getComponentType());
}
if (StringUtils.hasText(this.getComponentName())) {
metadataKeyBuilder.append("." + this.getComponentName());
}
else if (logger.isWarnEnabled()) {
logger.warn(this.getClass().getSimpleName() + " has no name. MetadataStore key might not be unique.");
}
if (this.twitter.isAuthorized()){
UserOperations userOperations = this.twitter.userOperations();
String profileId = String.valueOf(userOperations.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)) {
@@ -132,16 +131,10 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
public Message<?> receive() {
T tweet = this.tweets.poll();
if (tweet == null) {
long currentTime = System.currentTimeMillis();
long elapsedTime = currentTime - this.lastPollForTweet;
if (elapsedTime < 15000) {
// need to wait longer
return null;
}
this.refreshTweetQueueIfNecessary();
tweet = this.tweets.poll();
this.lastPollForTweet = currentTime;
}
if (tweet != null) {
this.lastProcessedId = this.getIdForTweet(tweet);
this.metadataStore.put(this.metadataKey, String.valueOf(this.lastProcessedId));
@@ -204,6 +197,27 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
}
/**
* Remove the metadata key and the corresponding value from the Metadata Store.
*/
@ManagedOperation(description="Remove the metadata key and the corresponding value from the Metadata Store.")
void resetMetadataStore() {
synchronized(this) {
this.metadataStore.remove(this.metadataKey);
this.lastProcessedId = -1L;
this.lastEnqueuedId = -1L;
}
}
/**
*
* @return {@code -1} if lastProcessedId is not set, yet.
*/
@ManagedAttribute
public long getLastProcessedId() {
return this.lastProcessedId;
}
private class TweetComparator implements Comparator<T> {
public int compare(T tweet1, T tweet2) {
@@ -230,6 +244,7 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
throw new IllegalArgumentException("Uncomparable Twitter objects: " + tweet1 + " and " + tweet2);
}
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2013 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.
@@ -31,14 +31,13 @@ import org.springframework.social.twitter.api.Twitter;
*/
public class DirectMessageReceivingMessageSource extends AbstractTwitterMessageSource<DirectMessage> {
public DirectMessageReceivingMessageSource(Twitter twitter) {
super(twitter);
public DirectMessageReceivingMessageSource(Twitter twitter, String metadataKey) {
super(twitter, metadataKey);
}
@Override
public String getComponentType() {
return "twitter:dm-inbound-channel-adapter";
return "twitter:dm-inbound-channel-adapter";
}
@Override

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2013 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.
@@ -30,11 +30,10 @@ import org.springframework.social.twitter.api.Twitter;
*/
public class MentionsReceivingMessageSource extends AbstractTwitterMessageSource<Tweet> {
public MentionsReceivingMessageSource(Twitter twitter) {
super(twitter);
public MentionsReceivingMessageSource(Twitter twitter, String metadataKey) {
super(twitter, metadataKey);
}
@Override
public String getComponentType() {
return "twitter:mentions-inbound-channel-adapter";

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2013 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.
@@ -35,11 +35,10 @@ public class SearchReceivingMessageSource extends AbstractTwitterMessageSource<T
private volatile String query;
public SearchReceivingMessageSource(Twitter twitter) {
super(twitter);
public SearchReceivingMessageSource(Twitter twitter, String metadataKey) {
super(twitter, metadataKey);
}
public void setQuery(String query) {
Assert.hasText(query, "'query' must not be null");
this.query = query;
@@ -47,7 +46,7 @@ public class SearchReceivingMessageSource extends AbstractTwitterMessageSource<T
@Override
public String getComponentType() {
return "twitter:search-inbound-channel-adapter";
return "twitter:search-inbound-channel-adapter";
}
@Override

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

@@ -22,7 +22,7 @@ import org.springframework.social.twitter.api.Tweet;
import org.springframework.social.twitter.api.Twitter;
/**
* This {@link org.springframework.integration.core.MessageSource} lets Spring Integration consume
* 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
@@ -31,14 +31,13 @@ import org.springframework.social.twitter.api.Twitter;
*/
public class TimelineReceivingMessageSource extends AbstractTwitterMessageSource<Tweet> {
public TimelineReceivingMessageSource(Twitter twitter) {
super(twitter);
public TimelineReceivingMessageSource(Twitter twitter, String metadataKey) {
super(twitter, metadataKey);
}
@Override
public String getComponentType() {
return "twitter:inbound-channel-adapter";
return "twitter:inbound-channel-adapter";
}
@Override

38
spring-integration-twitter/src/main/resources/org/springframework/integration/twitter/config/spring-integration-twitter-3.0.xsd
View File

@@ -123,7 +123,28 @@
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
</xsd:sequence>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="id" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>
The bean id of this Polling Endpoint; the MessageSource is also registered with this id
plus a suffix '.source'; also used as the
MetaDataStore key with suffix '.' + the profileId from the authorized Twitter user.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Identifies the channel the attached to this adapter, to which messages will be sent.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
<xsd:attribute name="twitter-template" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
@@ -136,6 +157,21 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="metadata-store" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to a MetadataStore instance for storing metadata associated with
the retrieved feeds. If the implementation is persistent, it can help to
prevent duplicates between restarts. If shared, it can help coordinate multiple
instances of an adapter across different processes.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.metadata.MetadataStore" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="outbound-twitter-type">

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2013 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.
@@ -16,7 +16,10 @@
package org.springframework.integration.twitter.config;
import static org.junit.Assert.assertNotNull;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
@@ -25,16 +28,13 @@ import org.springframework.integration.twitter.inbound.DirectMessageReceivingMes
import org.springframework.integration.twitter.inbound.MentionsReceivingMessageSource;
import org.springframework.integration.twitter.inbound.TimelineReceivingMessageSource;
import static org.junit.Assert.assertNotNull;
/**
* @author Oleg Zhurakousky
* @author Gunnar Hillert
*/
public class TestReceivingMessageSourceParserTests {
@Test
public void testReceivingAdapterConfigurationAutoStartup(){
ApplicationContext ac = new ClassPathXmlApplicationContext("TestReceivingMessageSourceParser-context.xml", this.getClass());
@@ -47,10 +47,24 @@ public class TestReceivingMessageSourceParserTests {
assertNotNull(dms);
spca = ac.getBean("updateAdapter", SourcePollingChannelAdapter.class);
spca = ac.getBean("updateAdapter", SourcePollingChannelAdapter.class);
TimelineReceivingMessageSource tms = TestUtils.getPropertyValue(spca, "source", TimelineReceivingMessageSource.class);
assertNotNull(tms);
}
@Test
public void testThatMessageSourcesAreRegisteredAsBeans(){
ApplicationContext ac = new ClassPathXmlApplicationContext("TestReceivingMessageSourceParser-context.xml", this.getClass());
MentionsReceivingMessageSource ms = ac.getBean("mentionAdapter.source", MentionsReceivingMessageSource.class);
assertNotNull(ms);
DirectMessageReceivingMessageSource dms = ac.getBean("dmAdapter.source", DirectMessageReceivingMessageSource.class);
assertNotNull(dms);
TimelineReceivingMessageSource tms = ac.getBean("updateAdapter.source", TimelineReceivingMessageSource.class);
assertNotNull(tms);
}
}

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

@@ -31,7 +31,7 @@ import org.springframework.social.twitter.api.impl.TwitterTemplate;
*/
public class DirectMessageReceivingMessageSourceTests {
@SuppressWarnings("unchecked")
@Test @Ignore
public void demoReceiveDm() throws Exception{
@@ -40,11 +40,11 @@ public class DirectMessageReceivingMessageSourceTests {
pf.afterPropertiesSet();
Properties prop = pf.getObject();
System.out.println(prop);
TwitterTemplate template = new TwitterTemplate(prop.getProperty("z_oleg.oauth.consumerKey"),
prop.getProperty("z_oleg.oauth.consumerSecret"),
prop.getProperty("z_oleg.oauth.accessToken"),
TwitterTemplate template = new TwitterTemplate(prop.getProperty("z_oleg.oauth.consumerKey"),
prop.getProperty("z_oleg.oauth.consumerSecret"),
prop.getProperty("z_oleg.oauth.accessToken"),
prop.getProperty("z_oleg.oauth.accessTokenSecret"));
DirectMessageReceivingMessageSource tSource = new DirectMessageReceivingMessageSource(template);
DirectMessageReceivingMessageSource tSource = new DirectMessageReceivingMessageSource(template, "foo");
tSource.afterPropertiesSet();
for (int i = 0; i < 50; i++) {
Message<DirectMessage> message = (Message<DirectMessage>) tSource.receive();

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

@@ -35,7 +35,7 @@ import org.junit.Test;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.integration.Message;
import org.springframework.integration.store.metadata.SimpleMetadataStore;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.social.twitter.api.SearchMetadata;
import org.springframework.social.twitter.api.SearchOperations;
@@ -66,7 +66,7 @@ public class SearchReceivingMessageSourceTests {
prop.getProperty("z_oleg.oauth.consumerSecret"),
prop.getProperty("z_oleg.oauth.accessToken"),
prop.getProperty("z_oleg.oauth.accessTokenSecret"));
SearchReceivingMessageSource tSource = new SearchReceivingMessageSource(template);
SearchReceivingMessageSource tSource = new SearchReceivingMessageSource(template, "foo");
tSource.setQuery(SEARCH_QUERY);
tSource.afterPropertiesSet();
for (int i = 0; i < 50; i++) {
@@ -84,14 +84,14 @@ public class SearchReceivingMessageSourceTests {
@Test
public void testSearchReceivingMessageSourceInit() {
final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(new TwitterTemplate("test"));
final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(new TwitterTemplate("test"), "foo");
messageSource.setComponentName("twitterSearchMessageSource");
final Object metadataStore = TestUtils.getPropertyValue(messageSource, "metadataStore");
final Object metadataKey = TestUtils.getPropertyValue(messageSource, "metadataKey");
assertNull(metadataStore);
assertNull(metadataKey);
assertNotNull(metadataKey);
messageSource.afterPropertiesSet();
@@ -101,7 +101,7 @@ public class SearchReceivingMessageSourceTests {
assertNotNull(metadataStoreInitialized);
assertTrue(metadataStoreInitialized instanceof SimpleMetadataStore);
assertNotNull(metadataKeyInitialized);
assertEquals("twitter:search-inbound-channel-adapter.twitterSearchMessageSource", metadataKeyInitialized);
assertEquals("foo", metadataKeyInitialized);
final Twitter twitter = TestUtils.getPropertyValue(messageSource, "twitter", Twitter.class);
@@ -123,7 +123,7 @@ public class SearchReceivingMessageSourceTests {
when(twitterTemplate.searchOperations()).thenReturn(so);
when(twitterTemplate.searchOperations().search(SEARCH_QUERY, 20, 0, 0)).thenReturn(null);
final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(twitterTemplate);
final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(twitterTemplate, "foo");
messageSource.setQuery(SEARCH_QUERY);
final String setQuery = TestUtils.getPropertyValue(messageSource, "query", String.class);
@@ -165,7 +165,7 @@ public class SearchReceivingMessageSourceTests {
SearchParameters params = new SearchParameters(SEARCH_QUERY).count(20).sinceId(0);
when(twitterTemplate.searchOperations().search(params)).thenReturn(results);
final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(twitterTemplate);
final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(twitterTemplate, "foo");
messageSource.setQuery(SEARCH_QUERY);

14
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests-context.xml
View File

@@ -1,27 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans
xmlns="http://www.springframework.org/schema/beans"
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-twitter="http://www.springframework.org/schema/integration/twitter"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration/twitter http://www.springframework.org/schema/integration/twitter/spring-integration-twitter.xsd">
<int:channel id="inbound_twitter"/>
<int:channel id="inbound_twitter">
<int:queue/>
</int:channel>
<int-twitter:search-inbound-channel-adapter id="twitterSearchAdapter"
query="springintegration"
twitter-template="twitterTemplate"
channel="inbound_twitter"
metadata-store="redisMetadataStore"
auto-startup="false">
<int:poller fixed-rate="5000" max-messages-per-poll="3"/>
<int:poller fixed-rate="100" max-messages-per-poll="3"/>
</int-twitter:search-inbound-channel-adapter>
<bean id="metadataStore" class="org.springframework.integration.redis.store.metadata.RedisMetadataStore">
<bean id="redisMetadataStore" class="org.springframework.integration.redis.metadata.RedisMetadataStore">
<constructor-arg name="connectionFactory" ref="redisConnectionFactory"/>
</bean>

103
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.twitter.inbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
@@ -30,47 +31,61 @@ import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.integration.Message;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.redis.store.metadata.RedisMetadataStore;
import org.springframework.integration.store.metadata.MetadataStore;
import org.springframework.integration.redis.metadata.RedisMetadataStore;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.social.twitter.api.SearchMetadata;
import org.springframework.social.twitter.api.SearchOperations;
import org.springframework.social.twitter.api.SearchResults;
import org.springframework.social.twitter.api.Tweet;
import org.springframework.social.twitter.api.UserOperations;
import org.springframework.social.twitter.api.impl.SearchParameters;
import org.springframework.social.twitter.api.impl.TwitterTemplate;
/**
* @author Gunnar Hillert
* @author Artem Bilan
* @since 3.0
*/
public class SearchReceivingMessageSourceWithRedisTests extends RedisAvailableTests {
private SourcePollingChannelAdapter twitterSearchAdapter;
private RedisConnectionFactory redisConnectionFactory;
private StringRedisTemplate redisTemplate;
private AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
private AbstractTwitterMessageSource twitterMessageSource;
private MetadataStore metadataStore;
private String metadataKey;
private PollableChannel tweets;
@Before
public void setup() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(SearchReceivingMessageSourceWithRedisTestsConfig.class);
context.registerShutdownHook();
context.refresh();
this.redisConnectionFactory = context.getBean(RedisConnectionFactory.class);
this.twitterSearchAdapter = context.getBean(SourcePollingChannelAdapter.class);
this.redisTemplate = new StringRedisTemplate(redisConnectionFactory);
this.twitterMessageSource = context.getBean(AbstractTwitterMessageSource.class);
this.metadataStore = context.getBean(MetadataStore.class);
this.tweets = context.getBean("inbound_twitter", PollableChannel.class);
this.metadataKey = TestUtils.getPropertyValue(twitterSearchAdapter, "source.metadataKey", String.class);
// There is need to set a value, not 'remove' and re-init 'twitterMessageSource'
this.metadataStore.put(metadataKey, "-1");
this.twitterMessageSource.afterPropertiesSet();
}
/**
@@ -80,51 +95,43 @@ public class SearchReceivingMessageSourceWithRedisTests extends RedisAvailableTe
@Test
@RedisAvailable
public void testPollForTweetsThreeResultsWithRedisMetadataStore() throws Exception {
final MetadataStore metadataStore = TestUtils.getPropertyValue(twitterSearchAdapter, "source.metadataStore", MetadataStore.class);
MetadataStore metadataStore = TestUtils.getPropertyValue(this.twitterSearchAdapter, "source.metadataStore", MetadataStore.class);
assertTrue("Exptected metadataStore to be an instance of RedisMetadataStore", metadataStore instanceof RedisMetadataStore);
assertSame(this.metadataStore, metadataStore);
/*
* The metadataKey is automatically generated. To ensure that we use the
* the correct key, we retrieve it from the adapter.
*/
final String metadataKey = TestUtils.getPropertyValue(twitterSearchAdapter, "source.metadataKey", String.class);
assertEquals("twitterSearchAdapter.74", metadataKey);
/*
* As we had to retrieve the metadataKey from the adapter. The metdataStore
* was already invoked and the id retrieved from Redis before we had a chance
* to reset possibly pre-existing values.
*
* Rather than deleting the value, we have to set a value, because "null" values
* returned from the MetadataStore are ignored by the onInit() method in
* the AbstractTwitterMessageSource. */
redisTemplate.opsForValue().set(metadataKey, "-1");
assertEquals("-1", redisTemplate.opsForValue().get(metadataKey));
this.twitterSearchAdapter.start();
final SearchReceivingMessageSource source = TestUtils.getPropertyValue(twitterSearchAdapter, "source", SearchReceivingMessageSource.class);
/* We need to call onInit() in order to update the id from the metadataStore. */
source.onInit();
final Message<?> message1 = source.receive();
final Message<?> message2 = source.receive();
final Message<?> message3 = source.receive();
assertNotNull(this.tweets.receive(10000));
assertNotNull(this.tweets.receive(1000));
assertNotNull(this.tweets.receive(1000));
/* We received 3 messages so far. When invoking receive() again the search
* will return again the 3 test Tweets but as we already processed them
* no message (null) is returned. */
final Message<?> message4 = source.receive();
assertNull(this.tweets.receive(0));
assertNotNull(message1);
assertNotNull(message2);
assertNotNull(message3);
assertNull(message4);
final String persistedMetadataStoreValue = redisTemplate.opsForValue().get(metadataKey);
String persistedMetadataStoreValue = this.metadataStore.get(metadataKey);
assertNotNull(persistedMetadataStoreValue);
assertEquals("3", redisTemplate.opsForValue().get(metadataKey));
assertEquals("3", persistedMetadataStoreValue);
redisTemplate.delete(metadataKey);
this.twitterSearchAdapter.stop();
this.metadataStore.put(metadataKey, "1");
this.twitterMessageSource.afterPropertiesSet();
this.twitterSearchAdapter.start();
assertNotNull(this.tweets.receive(1000));
assertNotNull(this.tweets.receive(1000));
assertNull(this.tweets.receive(0));
persistedMetadataStoreValue = this.metadataStore.get(metadataKey);
assertNotNull(persistedMetadataStoreValue);
assertEquals("3", persistedMetadataStoreValue);
}
@Configuration
@@ -152,7 +159,15 @@ public class SearchReceivingMessageSourceWithRedisTests extends RedisAvailableTe
when(twitterTemplate.searchOperations()).thenReturn(so);
when(twitterTemplate.searchOperations().search(any(SearchParameters.class))).thenReturn(results);
when(twitterTemplate.isAuthorized()).thenReturn(true);
final UserOperations userOperations = mock(UserOperations.class);
when(twitterTemplate.userOperations()).thenReturn(userOperations);
when(userOperations.getProfileId()).thenReturn(74L);
return twitterTemplate;
}
}
}

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

@@ -32,7 +32,7 @@ import org.springframework.social.twitter.api.impl.TwitterTemplate;
*/
public class TimelineReceivingMessageSourceTests {
@SuppressWarnings("unchecked")
@Test @Ignore
public void demoReceiveTimeline() throws Exception{
@@ -41,11 +41,11 @@ public class TimelineReceivingMessageSourceTests {
pf.afterPropertiesSet();
Properties prop = pf.getObject();
System.out.println(prop);
TwitterTemplate template = new TwitterTemplate(prop.getProperty("z_oleg.oauth.consumerKey"),
prop.getProperty("z_oleg.oauth.consumerSecret"),
prop.getProperty("z_oleg.oauth.accessToken"),
TwitterTemplate template = new TwitterTemplate(prop.getProperty("z_oleg.oauth.consumerKey"),
prop.getProperty("z_oleg.oauth.consumerSecret"),
prop.getProperty("z_oleg.oauth.accessToken"),
prop.getProperty("z_oleg.oauth.accessTokenSecret"));
TimelineReceivingMessageSource tSource = new TimelineReceivingMessageSource(template);
TimelineReceivingMessageSource tSource = new TimelineReceivingMessageSource(template, "foo");
tSource.afterPropertiesSet();
for (int i = 0; i < 50; i++) {
Message<Tweet> message = (Message<Tweet>) tSource.receive();