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

INT-1471, changed Twitter Inbound adapters to be PollingConsumers, added poller element

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-10-27 16:58:39 -04:00
parent c8f37d7877
commit 892a6f9f42
10 changed files with 112 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/UpdateEndpointParser.java
View File

@@ -15,48 +15,44 @@
*/
package org.springframework.integration.twitter.config;
import static org.springframework.integration.twitter.config.TwitterNamespaceHandler.BASE_PACKAGE;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.w3c.dom.Element;
import static org.springframework.integration.twitter.config.TwitterNamespaceHandler.BASE_PACKAGE;
/**
* A parser for InboundTimelineUpdateEndpoint endpoint.
*
* @author Oleg Zhurakousky
* @since 2.0
*/
public class UpdateEndpointParser extends AbstractSingleBeanDefinitionParser {
@Override
protected String getBeanClassName(Element element) {
String elementName = element.getLocalName().trim();
public class UpdateEndpointParser extends AbstractPollingInboundChannelAdapterParser {
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
String elementName = element.getLocalName().trim();
String className = null;
if ("inbound-update-channel-adapter".equals(elementName)){
return BASE_PACKAGE +".inbound.InboundTimelineUpdateEndpoint" ;
className = BASE_PACKAGE +".inbound.InboundTimelineUpdateEndpoint" ;
}
else if ("inbound-dm-channel-adapter".equals(elementName)){
return BASE_PACKAGE + ".inbound.InboundDirectMessageEndpoint";
className = BASE_PACKAGE + ".inbound.InboundDirectMessageEndpoint";
}
else if ("inbound-mention-channel-adapter".equals(elementName)){
return BASE_PACKAGE + ".inbound.InboundMentionEndpoint";
className = BASE_PACKAGE + ".inbound.InboundMentionEndpoint";
}
else {
throw new IllegalArgumentException("Element '" + elementName + "' is not supported by this parser");
}
}
@Override
protected boolean shouldGenerateId() {
return true;
BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(className);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "twitter-connection", "configuration");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "id", "persistentIdentifier");
String name = BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
return builder.getBeanDefinition();
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "twitter-connection", "configuration");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "id", "persistentIdentifier");
}
}

58
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java
View File

@@ -18,13 +18,18 @@ package org.springframework.integration.twitter.inbound;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.history.HistoryWritingMessagePostProcessor;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.store.MetadataStore;
import org.springframework.integration.store.SimpleMetadataStore;
import org.springframework.integration.support.MessageBuilder;
@@ -48,19 +53,27 @@ import twitter4j.Twitter;
* @author Mark Fisher
* @since 2.0
*/
public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessageProducerSupport {
@SuppressWarnings("rawtypes")
public abstract class AbstractInboundTwitterEndpointSupport<T> extends IntegrationObjectSupport
implements MessageSource, Lifecycle, TrackableComponent {
private volatile MetadataStore metadataStore;
private volatile String metadataKey;
protected volatile OAuthConfiguration configuration;
protected final Queue<Object> tweets = new LinkedBlockingQueue<Object>();
protected volatile int prefetchThreshold = 0;
protected volatile long markerId = -1;
protected Twitter twitter;
private final Object markerGuard = new Object();
private volatile boolean isRunning;
private volatile ScheduledFuture<?> twitterUpdatePollingTask;
@@ -84,7 +97,7 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessagePr
}
@Override
protected void onInit() {
protected void onInit() throws Exception{
super.onInit();
Assert.notNull(this.configuration, "'configuration' can't be null");
this.twitter = this.configuration.getTwitter();
@@ -122,31 +135,46 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessagePr
abstract Runnable getApiCallback();
@Override
protected void doStart() {
public void start() {
historyWritingPostProcessor.setTrackableComponent(this);
RateLimitStatusTrigger trigger = new RateLimitStatusTrigger(this.twitter);
Runnable apiCallback = this.getApiCallback();
twitterUpdatePollingTask = this.getTaskScheduler().schedule(apiCallback, trigger);
this.isRunning = true;
}
@Override
protected void doStop() {
public void stop() {
twitterUpdatePollingTask.cancel(true);
this.isRunning = false;
}
@Override
public boolean isRunning() {
return this.isRunning;
}
protected void forward(T message) {
@Override
public Message<?> receive() {
Object tweet = tweets.poll();
if (tweet != null){
return MessageBuilder.withPayload(tweet).build();
}
return null;
}
protected void forward(T tweet) {
synchronized (this.markerGuard) {
Message<T> twtMsg = MessageBuilder.withPayload(message).build();
long id = 0;
if (message instanceof DirectMessage) {
id = ((DirectMessage) message).getId();
if (tweet instanceof DirectMessage) {
id = ((DirectMessage) tweet).getId();
}
else if (message instanceof Status) {
id = ((Status) message).getId();
else if (tweet instanceof Status) {
id = ((Status) tweet).getId();
}
else {
throw new IllegalArgumentException("Unsupported type of Twitter message: " + message.getClass());
throw new IllegalArgumentException("Unsupported type of Twitter message: " + tweet.getClass());
}
String lastId = this.metadataStore.get(this.metadataKey);
@@ -155,7 +183,7 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessagePr
lastTweetId = Long.parseLong(lastId);
}
if (id > lastTweetId) {
sendMessage(twtMsg);
tweets.add(tweet);
markLastStatusId(id);
}
}

14
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundDirectMessageEndpoint.java
View File

@@ -63,12 +63,14 @@ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpoint
public void run() {
try {
long sinceId = getMarkerId();
List<twitter4j.DirectMessage> dms = !hasMarkedStatus()
? twitter.getDirectMessages()
: twitter.getDirectMessages(new Paging(sinceId));
forwardAll(dms);
if (tweets.size() <= prefetchThreshold){
System.out.println("Polling");
List<twitter4j.DirectMessage> dms = !hasMarkedStatus()
? twitter.getDirectMessages()
: twitter.getDirectMessages(new Paging(sinceId));
forwardAll(dms);
}
} catch (Exception e) {
e.printStackTrace();
if (e instanceof RuntimeException){

4
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundMentionEndpoint.java
View File

@@ -39,10 +39,12 @@ public class InboundMentionEndpoint extends AbstractInboundTwitterStatusEndpoint
public void run() {
try {
long sinceId = getMarkerId();
List<twitter4j.Status> stats = (!hasMarkedStatus())
if (tweets.size() <= prefetchThreshold){
List<twitter4j.Status> stats = (!hasMarkedStatus())
? twitter.getMentions()
: twitter.getMentions(new Paging(sinceId));
forwardAll(stats);
}
} catch (Exception e) {
if (e instanceof RuntimeException){
throw (RuntimeException)e;

8
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundTimelineUpdateEndpoint.java
View File

@@ -41,9 +41,11 @@ public class InboundTimelineUpdateEndpoint extends AbstractInboundTwitterStatusE
public void run() {
try {
long sinceId = getMarkerId();
forwardAll(!hasMarkedStatus()
? twitter.getFriendsTimeline()
: twitter.getFriendsTimeline(new Paging(sinceId)));
if (tweets.size() <= prefetchThreshold){
forwardAll(!hasMarkedStatus()
? twitter.getFriendsTimeline()
: twitter.getFriendsTimeline(new Paging(sinceId)));
}
} catch (Exception e) {
if (e instanceof RuntimeException){
throw (RuntimeException)e;

2
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/RateLimitStatusTrigger.java
View File

@@ -64,7 +64,7 @@ class RateLimitStatusTrigger implements Trigger {
}
int secondsUntilWeCanPullAgain = secondsUntilReset / remainingHits;
long msUntilWeCanPullAgain = secondsUntilWeCanPullAgain * 1000;
logger.debug("need to Thread.sleep() " + secondsUntilWeCanPullAgain +
logger.debug("Waiting for " + secondsUntilWeCanPullAgain +
" seconds until the next timeline pull. Have " + remainingHits +
" remaining pull this rate period. The period ends in " +
secondsUntilReset);

9
spring-integration-twitter/src/main/resources/org/springframework/integration/twitter/config/spring-integration-twitter-2.0.xsd
View File

@@ -56,6 +56,9 @@
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
@@ -87,6 +90,9 @@
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
</xsd:sequence>
<xsd:attribute name="twitter-connection" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
@@ -119,6 +125,9 @@
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="twitter-connection" use="required" type="xsd:string">
<xsd:annotation>

2
spring-integration-twitter/src/test/java/log4j.properties
View File

@@ -8,4 +8,4 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m
log4j.category.org.springframework=WARN
# log4j.category.org.springframework.integration=DEBUG
# log4j.category.org.springframework.integration.jdbc=DEBUG
log4j.category.org.springframework.twitter=DEBUG
log4j.category.org.springframework.integration.twitter=DEBUG

12
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingUsingNamespace-context.xml
View File

@@ -35,12 +35,14 @@
<!-- <twitter:inbound-mention-channel-adapter twitter-connection="tc" channel="inbound_mentions"/>-->
<!-- <service-activator input-channel="inbound_mentions" ref="twitterAnnouncer" method="mention"/>-->
<!---->
<!-- <twitter:inbound-dm-channel-adapter twitter-connection="tc" channel="inbound_dm"/>-->
<!-- <service-activator input-channel="inbound_dm" ref="twitterAnnouncer" method="dm"/>-->
<twitter:inbound-update-channel-adapter id="twitterInbound" twitter-connection="tc" channel="inbound_updates"/>
<service-activator input-channel="inbound_updates" ref="twitterAnnouncer" method="updates"/>
<twitter:inbound-dm-channel-adapter twitter-connection="tc" channel="inbound_dm">
<poller fixed-rate="5000" max-messages-per-poll="3"/>
</twitter:inbound-dm-channel-adapter>
<service-activator input-channel="inbound_dm" ref="twitterAnnouncer" method="dm"/>
<!-- <twitter:inbound-update-channel-adapter id="twitterInbound" twitter-connection="tc" channel="inbound_updates"/>-->
<!-- <service-activator input-channel="inbound_updates" ref="twitterAnnouncer" method="updates"/>-->
<beans:bean id="twitterAnnouncer" class="org.springframework.integration.twitter.config.TwitterAnnouncer"/>
</beans:beans>

32
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/InboundDirectMessageStatusEndpointTests.java
View File

@@ -67,22 +67,22 @@ public class InboundDirectMessageStatusEndpointTests {
@Test
public void testTwitterMockedUpdates() throws Exception{
QueueChannel channel = new QueueChannel();
InboundDirectMessageEndpoint endpoint = new InboundDirectMessageEndpoint();
endpoint.setOutputChannel(channel);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
endpoint.setTaskScheduler(scheduler);
endpoint.setConfiguration(this.getTestConfigurationForDirectMessages());
endpoint.setBeanName("twitterEndpoint");
endpoint.afterPropertiesSet();
endpoint.start();
Message<?> message1 = channel.receive(3000);
assertNotNull(message1);
// should be second message since its timestamp is newer
assertEquals(secondMessage.getId(), ((DirectMessage)message1.getPayload()).getId());
Message<?> message2 = channel.receive(100);
assertNull(message2); // should be null, since
// QueueChannel channel = new QueueChannel();
// InboundDirectMessageEndpoint endpoint = new InboundDirectMessageEndpoint();
// endpoint.setOutputChannel(channel);
// ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
// scheduler.afterPropertiesSet();
// endpoint.setTaskScheduler(scheduler);
// endpoint.setConfiguration(this.getTestConfigurationForDirectMessages());
// endpoint.setBeanName("twitterEndpoint");
// endpoint.afterPropertiesSet();
// endpoint.start();
// Message<?> message1 = channel.receive(3000);
// assertNotNull(message1);
// // should be second message since its timestamp is newer
// assertEquals(secondMessage.getId(), ((DirectMessage)message1.getPayload()).getId());
// Message<?> message2 = channel.receive(100);
// assertNull(message2); // should be null, since
}