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

INT-1471 consolidated Inbound parsers, added initial support for MetadataStore to handle duplicates

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-10-26 14:59:33 -04:00
committed by Chris Beams
parent 0a49a46af9
commit 601726b6d3
9 changed files with 93 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundDirectMessageEndpointParser.java
View File

@@ -1,48 +0,0 @@
/*
* 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.config;
import static org.springframework.integration.twitter.config.TwitterNamespaceHandler.BASE_PACKAGE;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.w3c.dom.Element;
/**
* A parser for a 'inbound-dm-channel' element
*
* @author Josh Long
* @since 2.0
*/
public class InboundDirectMessageEndpointParser extends AbstractSingleBeanDefinitionParser {
@Override
protected String getBeanClassName(Element element) {
return BASE_PACKAGE + ".inbound.InboundDirectMessageEndpoint";
}
@Override
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "twitter-connection", "configuration");
}
}

48
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundMentionEndpointParser.java
View File

@@ -1,48 +0,0 @@
/*
* Copyright 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.config;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.w3c.dom.Element;
import static org.springframework.integration.twitter.config.TwitterNamespaceHandler.BASE_PACKAGE;
/**
* Parser for 'inbound-mention-channel-adapter' element
*
* @author Josh Long
* @since 2.0
*/
public class InboundMentionEndpointParser extends AbstractSingleBeanDefinitionParser {
@Override
protected String getBeanClassName(Element element) {
return BASE_PACKAGE + ".inbound.InboundMentionEndpoint";
}
@Override
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "twitter-connection", "configuration");
}
}

6
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterNamespaceHandler.java
View File

@@ -36,9 +36,9 @@ public class TwitterNamespaceHandler extends org.springframework.beans.factory.x
registerBeanDefinitionParser("twitter-connection", new ConnectionParser());
// inbound
registerBeanDefinitionParser("inbound-update-channel-adapter", new InboundTimelineUpdateEndpointParser());
registerBeanDefinitionParser("inbound-dm-channel-adapter", new InboundDirectMessageEndpointParser());
registerBeanDefinitionParser("inbound-mention-channel-adapter", new InboundMentionEndpointParser());
registerBeanDefinitionParser("inbound-update-channel-adapter", new UpdateEndpointParser());
registerBeanDefinitionParser("inbound-dm-channel-adapter", new UpdateEndpointParser());
registerBeanDefinitionParser("inbound-mention-channel-adapter", new UpdateEndpointParser());
// outbound
registerBeanDefinitionParser("outbound-update-channel-adapter", new OutboundTimelineUpdateMessageHandlerParser());

26
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundTimelineUpdateEndpointParser.java → spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/UpdateEndpointParser.java
View File

@@ -26,23 +26,35 @@ import static org.springframework.integration.twitter.config.TwitterNamespaceHan
/**
* A parser for InboundTimelineUpdateEndpoint endpoint.
*
* @author Josh Long
* @author Oleg Zhurakousky
* @since 2.0
*/
public class InboundTimelineUpdateEndpointParser extends AbstractSingleBeanDefinitionParser {
public class UpdateEndpointParser extends AbstractSingleBeanDefinitionParser {
@Override
protected String getBeanClassName(Element element) {
return BASE_PACKAGE +".inbound.InboundTimelineUpdateEndpoint" ;
String elementName = element.getLocalName().trim();
if ("inbound-update-channel-adapter".equals(elementName)){
return BASE_PACKAGE +".inbound.InboundTimelineUpdateEndpoint" ;
}
else if ("inbound-dm-channel-adapter".equals(elementName)){
return BASE_PACKAGE + ".inbound.InboundDirectMessageEndpoint";
}
else if ("inbound-mention-channel-adapter".equals(elementName)){
return BASE_PACKAGE + ".inbound.InboundMentionEndpoint";
}
else {
throw new IllegalArgumentException("Element '" + elementName + "' is not supported by this parser");
}
}
@Override
protected boolean shouldGenerateIdAsFallback() {
return true;
}
protected boolean shouldGenerateId() {
return true;
}
@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");
}
}

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

@@ -17,20 +17,25 @@ package org.springframework.integration.twitter.inbound;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ScheduledFuture;
import org.springframework.context.Lifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.context.metadata.FileBasedPropertiesStore;
import org.springframework.integration.context.metadata.MetadataStore;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.history.HistoryWritingMessagePostProcessor;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.twitter.oauth.OAuthConfiguration;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import twitter4j.DirectMessage;
import twitter4j.Status;
import twitter4j.Twitter;
/**
* Abstract class that defines common operations for receiving various types of messages when using the
* Twitter API.
@@ -44,21 +49,30 @@ import twitter4j.Twitter;
* @since 2.0
*/
public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessageProducerSupport implements Lifecycle, TrackableComponent{
private volatile MetadataStore metadataStore;
private volatile String metadataKey;
private volatile Properties lastPersistentEntry = new Properties();
protected volatile OAuthConfiguration configuration;
protected volatile long markerId = -1;
protected Twitter twitter;
private final Object markerGuard = new Object();
private volatile ScheduledFuture<?> twitterUpdatePollingTask;
private String persistentIdentifier;
private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor();
private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor();
abstract protected void markLastStatusId(T statusId);
abstract protected List<T> sort(List<T> rl);
abstract protected List<T> sort(List<T> rl);
abstract Runnable getApiCallback();
protected void markLastStatusId(long statusId){
lastPersistentEntry.put(metadataKey, String.valueOf(statusId));
}
public void setPersistentIdentifier(String persistentIdentifier) {
this.persistentIdentifier = persistentIdentifier;
}
public void setConfiguration(OAuthConfiguration configuration) {
this.configuration = configuration;
}
@@ -72,11 +86,24 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessagePr
}
@Override
protected void onInit() {
protected void onInit() {
super.onInit();
Assert.notNull(this.configuration, "'configuration' can't be null");
this.twitter = this.configuration.getTwitter();
Assert.notNull(this.twitter, "'twitter' instance can't be null");
metadataKey = this.getComponentType() + "@" + this.getComponentName() + "#" + this.configuration.getConsumerKey();
try {
if (StringUtils.hasText(this.persistentIdentifier)){
if (this.metadataStore == null){
logger.info("Creating FileBasedPropertiesStore");
metadataStore = new FileBasedPropertiesStore(this.persistentIdentifier);
((FileBasedPropertiesStore)metadataStore).afterPropertiesSet();
}
lastPersistentEntry = metadataStore.load();
}
} catch (Exception e) {
logger.warn("Failed to initailize initiaize and load from MetadataStore. Potential duplicates ppossible", e);
}
}
protected void forwardAll(List<T> tResponses) {
@@ -104,13 +131,34 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessagePr
twitterUpdatePollingTask.cancel(true);
}
protected void forward(T status) {
protected void forward(T message) {
synchronized (this.markerGuard) {
Message<T> twtMsg = MessageBuilder.withPayload(status).build();
Message<T> twtMsg = MessageBuilder.withPayload(message).build();
sendMessage(twtMsg);
markLastStatusId(status);
long id = 0;
if (message instanceof DirectMessage) {
id = ((DirectMessage)message).getId();
}
else if (message instanceof Status){
id = ((Status)message).getId();
}
else {
throw new IllegalArgumentException("Unsupported type of Twitter message: " + message.getClass());
}
String lastId = lastPersistentEntry.getProperty(this.metadataKey);
long lastTweetId = 0;
if (lastId != null){
lastTweetId = Long.parseLong(lastId);
}
if (id > lastTweetId){
sendMessage(twtMsg);
markLastStatusId(id);
if (metadataStore != null){
metadataStore.write(this.lastPersistentEntry);
}
}
}
}
protected boolean hasMarkedStatus() {

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

@@ -20,7 +20,6 @@ import java.util.Comparator;
import java.util.List;
import twitter4j.Status;
import twitter4j.TwitterFactory;
/**
* Simple base class for the reply and timeline cases (as well as any other {@link twitter4j.Status} implementations of
@@ -37,19 +36,6 @@ abstract public class AbstractInboundTwitterStatusEndpointSupport extends Abstra
}
};
// protected List<Status> fromTwitter4jStatuses(List<twitter4j.Status> stats) {
// List<Status> fwd = new ArrayList<Status>();
// for (twitter4j.Status s : stats) {
// fwd.add((Status) TwitterFactory.formTwitter4jMessage(s));
// }
// return fwd;
// }
@Override
protected void markLastStatusId(Status statusId) {
this.markerId = statusId.getId();
}
@Override
protected List<Status> sort(List<Status> rl) {
List<Status> statusArrayList = new ArrayList<Status>();

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

@@ -40,11 +40,6 @@ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpoint
}
};
@Override
protected void markLastStatusId(DirectMessage dm) {
this.markerId = dm.getId();
}
@Override
protected List<DirectMessage> sort(List<DirectMessage> rl) {
@@ -73,11 +68,6 @@ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpoint
? twitter.getDirectMessages()
: twitter.getDirectMessages(new Paging(sinceId));
// List<DirectMessage> dmsToFwd = new ArrayList<DirectMessage>();
//
// for( twitter4j.DirectMessage dm : dms) {
// dmsToFwd.add((DirectMessage) TwitterFactory.formTwitter4jMessage(dm));
// }
forwardAll(dms);
} catch (Exception e) {
e.printStackTrace();
@@ -92,5 +82,4 @@ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpoint
};
return apiCallback;
}
}

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

@@ -33,13 +33,13 @@
consumer-secret="${twitter.oauth.consumerSecret}"/>
<twitter:inbound-mention-channel-adapter twitter-connection="tc" channel="inbound_mentions"/>
<service-activator input-channel="inbound_mentions" ref="twitterAnnouncer" method="mention"/>
<!-- <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-dm-channel-adapter twitter-connection="tc" channel="inbound_dm"/>
<service-activator input-channel="inbound_dm" ref="twitterAnnouncer" method="dm"/>
<twitter:inbound-update-channel-adapter twitter-connection="tc" channel="inbound_updates"/>
<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"/>

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

@@ -17,6 +17,7 @@ package org.springframework.integration.twitter.inbound;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNotNull;
import static junit.framework.Assert.assertNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -62,11 +63,10 @@ public class InboundDirectMessageStatusEndpointTests {
endpoint.start();
Message message1 = channel.receive(3000);
assertNotNull(message1);
System.out.println();
// should be second message since its timestamp is newer
assertEquals(secondMessage.getId(), ((DirectMessage)message1.getPayload()).getId());
Message message2 = channel.receive(3000);
assertNotNull(message2);
assertEquals(firstMessage.getId(), ((DirectMessage)message2.getPayload()).getId());
Message message2 = channel.receive(100);
assertNull(message2); // should be null, since
}
@Before