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

INT-1603 added logic to separate markedId vs lastProcessedId

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-11-11 15:10:52 -05:00
parent 6335dd44d7
commit a2adf47992
3 changed files with 32 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -64,6 +64,8 @@ public abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint
protected volatile int prefetchThreshold = 0;
protected volatile long markerId = -1;
protected volatile long processedId = -1;
protected final TwitterOperations twitter;
@@ -124,7 +126,7 @@ public abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint
String lastId = this.metadataStore.get(this.metadataKey);
// initialize the last status ID from the metadataStore
if (StringUtils.hasText(lastId)){
this.markLastStatusId(Long.parseLong(lastId));
this.markerId = Long.parseLong(lastId);
}
}
@@ -166,7 +168,7 @@ public abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint
public Message<?> receive() {
Tweet tweet = tweets.poll();
if (tweet != null){
this.markProcessedId(tweet.getId());
return MessageBuilder.withPayload(tweet).build();
}
return null;
@@ -176,22 +178,16 @@ public abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint
synchronized (this.markerGuard) {
long id = tweet.getId();
String lastId = this.metadataStore.get(this.metadataKey);
long lastTweetId = 0;
if (lastId != null) {
lastTweetId = Long.parseLong(lastId);
}
if (id > lastTweetId) {
markLastStatusId(tweet.getId());
if (id > this.markerId) {
this.markerId = id;
tweets.add(tweet);
}
}
}
protected void markLastStatusId(long statusId) {
this.markerId = statusId;
protected void markProcessedId(long statusId) {
this.processedId = statusId;
this.metadataStore.put(this.metadataKey, String.valueOf(statusId));
}
}

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

@@ -33,7 +33,9 @@ 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.Message;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.store.PropertiesPersistingMetadataStore;
import org.springframework.integration.test.util.TestUtils;
@@ -131,8 +133,8 @@ public class DirectMessageReceivingMessageSourceTests {
Tweet message = (Tweet) msg.poll();
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
verify(twitter, times(1)).getDirectMessages();
// based on the Mock, the Queue shoud now have 1 more messages third and fourth
assertTrue(((Queue)TestUtils.getPropertyValue(source, "tweets")).size() == 2);
source.stop();
}
@@ -165,8 +167,9 @@ public class DirectMessageReceivingMessageSourceTests {
Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
assertTrue(!CollectionUtils.isEmpty(msg));
assertEquals(1, msg.size());
Tweet message = (Tweet) msg.poll();
assertEquals(2000, message.getId());
Message message = source.receive();
Tweet tweet = (Tweet) message.getPayload();
assertEquals(2000, tweet.getId());
source.stop();
Thread.sleep(3000);
@@ -190,10 +193,12 @@ public class DirectMessageReceivingMessageSourceTests {
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());
message = source.receive();
tweet = (Tweet) message.getPayload();
assertEquals(3000, tweet.getId());
message = source.receive();
tweet = (Tweet) message.getPayload();
assertEquals(4000, tweet.getId());
file.delete();
}

18
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSourceTests.java
View File

@@ -34,6 +34,7 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.store.PropertiesPersistingMetadataStore;
import org.springframework.integration.test.util.TestUtils;
@@ -165,10 +166,11 @@ public class TimelineUpdateReceivingMessageSourceTests {
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());
Message message = (Message) source.receive();
Tweet tweet = (Tweet) message.getPayload();
assertEquals(2000, tweet.getId());
source.stop();
Thread.sleep(3000);
Thread.sleep(2000);
// Resuming
@@ -190,10 +192,12 @@ public class TimelineUpdateReceivingMessageSourceTests {
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());
message = (Message) source.receive();
tweet = (Tweet) message.getPayload();
assertEquals(3000, tweet.getId());
message = (Message) source.receive();
tweet = (Tweet) message.getPayload();
assertEquals(4000, tweet.getId());
file.delete();
}