From 601726b6d3359aea083c0146920c3058bd3f149b Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 26 Oct 2010 14:59:33 -0400 Subject: [PATCH] INT-1471 consolidated Inbound parsers, added initial support for MetadataStore to handle duplicates --- .../InboundDirectMessageEndpointParser.java | 48 ------------ .../config/InboundMentionEndpointParser.java | 48 ------------ .../config/TwitterNamespaceHandler.java | 6 +- ...tParser.java => UpdateEndpointParser.java} | 26 +++++-- ...AbstractInboundTwitterEndpointSupport.java | 74 +++++++++++++++---- ...ctInboundTwitterStatusEndpointSupport.java | 14 ---- .../inbound/InboundDirectMessageEndpoint.java | 11 --- .../TestReceivingUsingNamespace-context.xml | 12 +-- ...boundDirectMessageStatusEndpointTests.java | 8 +- 9 files changed, 93 insertions(+), 154 deletions(-) delete mode 100644 spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundDirectMessageEndpointParser.java delete mode 100644 spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundMentionEndpointParser.java rename spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/{InboundTimelineUpdateEndpointParser.java => UpdateEndpointParser.java} (65%) diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundDirectMessageEndpointParser.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundDirectMessageEndpointParser.java deleted file mode 100644 index dbd11dac39..0000000000 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundDirectMessageEndpointParser.java +++ /dev/null @@ -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"); - } -} diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundMentionEndpointParser.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundMentionEndpointParser.java deleted file mode 100644 index 5b7d0b096b..0000000000 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundMentionEndpointParser.java +++ /dev/null @@ -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"); - } -} diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterNamespaceHandler.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterNamespaceHandler.java index a7cebabaf0..1c44a26e54 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterNamespaceHandler.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterNamespaceHandler.java @@ -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()); diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundTimelineUpdateEndpointParser.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/UpdateEndpointParser.java similarity index 65% rename from spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundTimelineUpdateEndpointParser.java rename to spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/UpdateEndpointParser.java index cd2ae1ed43..ffe8f4a8ba 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/InboundTimelineUpdateEndpointParser.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/UpdateEndpointParser.java @@ -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"); } } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java index 550504d6d4..3877add1f9 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java @@ -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 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 sort(List rl); + abstract protected List sort(List 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 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 tResponses) { @@ -104,13 +131,34 @@ public abstract class AbstractInboundTwitterEndpointSupport extends MessagePr twitterUpdatePollingTask.cancel(true); } - protected void forward(T status) { + protected void forward(T message) { synchronized (this.markerGuard) { - Message twtMsg = MessageBuilder.withPayload(status).build(); + Message 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() { diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterStatusEndpointSupport.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterStatusEndpointSupport.java index 6bf5048846..f26d4f711d 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterStatusEndpointSupport.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterStatusEndpointSupport.java @@ -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 fromTwitter4jStatuses(List stats) { -// List fwd = new ArrayList(); -// 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 sort(List rl) { List statusArrayList = new ArrayList(); diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundDirectMessageEndpoint.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundDirectMessageEndpoint.java index 1a4e4670fc..b18f9c82fc 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundDirectMessageEndpoint.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundDirectMessageEndpoint.java @@ -40,11 +40,6 @@ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpoint } }; - @Override - protected void markLastStatusId(DirectMessage dm) { - this.markerId = dm.getId(); - } - @Override protected List sort(List rl) { @@ -73,11 +68,6 @@ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpoint ? twitter.getDirectMessages() : twitter.getDirectMessages(new Paging(sinceId)); -// List dmsToFwd = new ArrayList(); -// -// 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; } - } diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingUsingNamespace-context.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingUsingNamespace-context.xml index c406470708..dc594acaf1 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingUsingNamespace-context.xml +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingUsingNamespace-context.xml @@ -33,13 +33,13 @@ consumer-secret="${twitter.oauth.consumerSecret}"/> - - + + + + + - - - - + diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/InboundDirectMessageStatusEndpointTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/InboundDirectMessageStatusEndpointTests.java index e9f1ca5226..9c43b5859f 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/InboundDirectMessageStatusEndpointTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/InboundDirectMessageStatusEndpointTests.java @@ -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