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

INT-4368: Twitter - move init to start()

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4368

Test fails if redis not available.

Also fix synchronization of `lastEnqueuedId`.
This commit is contained in:
Gary Russell
2017-12-19 15:01:11 -05:00
committed by Artem Bilan
parent bd872846fb
commit e585bb949f
1 changed files with 41 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -24,6 +24,7 @@ import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
@@ -57,7 +58,8 @@ import org.springframework.util.StringUtils;
* @since 2.0
*/
@SuppressWarnings("rawtypes")
abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport implements MessageSource {
abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport implements MessageSource,
Lifecycle {
private static final int DEFAULT_PAGE_SIZE = 20;
@@ -69,10 +71,10 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
private final String metadataKey;
private volatile MetadataStore metadataStore;
private final Queue<T> tweets = new LinkedBlockingQueue<T>();
private volatile MetadataStore metadataStore;
private volatile int prefetchThreshold = 0;
private volatile long lastEnqueuedId = -1;
@@ -81,6 +83,7 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
private volatile int pageSize = DEFAULT_PAGE_SIZE;
private volatile boolean running;
protected AbstractTwitterMessageSource(Twitter twitter, String metadataKey) {
Assert.notNull(twitter, "twitter must not be null");
@@ -132,13 +135,31 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
}
}
String lastId = this.metadataStore.get(this.metadataKey);
// initialize the last status ID from the metadataStore
if (StringUtils.hasText(lastId)) {
this.lastProcessedId = Long.parseLong(lastId);
this.lastEnqueuedId = this.lastProcessedId;
}
}
@Override
public synchronized void start() {
if (!this.running) {
String lastId = this.metadataStore.get(this.metadataKey);
// initialize the last status ID from the metadataStore
if (StringUtils.hasText(lastId)) {
this.lastProcessedId = Long.parseLong(lastId);
synchronized (this.lastEnqueuedIdMonitor) {
this.lastEnqueuedId = this.lastProcessedId;
}
}
this.running = true;
}
}
@Override
public synchronized void stop() {
this.running = false;
}
@Override
public synchronized boolean isRunning() {
return this.running;
}
@Override
@@ -169,7 +190,9 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
long id = this.getIdForTweet(tweet);
if (id > this.lastEnqueuedId) {
this.tweets.add(tweet);
this.lastEnqueuedId = id;
synchronized (this.lastEnqueuedIdMonitor) {
this.lastEnqueuedId = id;
}
}
}
}
@@ -177,9 +200,11 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
private void refreshTweetQueueIfNecessary() {
try {
if (this.tweets.size() <= this.prefetchThreshold) {
List<T> tweets = pollForTweets(this.lastEnqueuedId);
if (!CollectionUtils.isEmpty(tweets)) {
enqueueAll(tweets);
synchronized (this.lastEnqueuedIdMonitor) {
List<T> tweets = pollForTweets(this.lastEnqueuedId);
if (!CollectionUtils.isEmpty(tweets)) {
enqueueAll(tweets);
}
}
}
}
@@ -222,7 +247,9 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
synchronized (this) {
this.metadataStore.remove(this.metadataKey);
this.lastProcessedId = -1L;
this.lastEnqueuedId = -1L;
synchronized (this.lastEnqueuedIdMonitor) {
this.lastEnqueuedId = -1L;
}
}
}