diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterEndpointSupport.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterEndpointSupport.java index 4ee0f19a2c..be32e65469 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterEndpointSupport.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterEndpointSupport.java @@ -24,8 +24,11 @@ import org.apache.commons.lang.exception.ExceptionUtils; import org.springframework.context.Lifecycle; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; +import org.springframework.integration.context.metadata.MetadataPersister; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.endpoint.AbstractEndpoint; +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; @@ -35,18 +38,22 @@ import twitter4j.ResponseList; import twitter4j.Twitter; /** - * There are a lot of operations that are common to receiving the various types of messages when using the Twitter API, and this - * class abstracts most of them for you. Implementers must take note of {@link org.springframework.integration.twitter.AbstractInboundTwitterEndpointSupport#runAsAPIRateLimitsPermit(org.springframework.integration.twitter.AbstractInboundTwitterEndpointSupport.ApiCallback)} - * which will invoke the instance of {@link org.springframework.integration.twitter.AbstractInboundTwitterEndpointSupport.ApiCallback} when the rate-limit API - * deems that its OK to do so. This class handles keeping tabs on that and on spacing out requests as required. + * There are a lot of operations that are common to receiving the various types of messages when using the + * Twitter API, and this + * class abstracts most of them for you. Implementers must take note of + * {@link AbstractInboundTwitterEndpointSupport#runAsAPIRateLimitsPermit(AbstractInboundTwitterEndpointSupport.ApiCallback)} + * which will invoke the instance of {@link AbstractInboundTwitterEndpointSupport.ApiCallback} when the + * rate-limit API deems that its OK to do so. This class handles keeping tabs on that and on spacing out requests + * as required. *

- * Simialarly, this class handles keeping track on the latest inbound message its received and avoiding, where possible, redelivery of - * common messages. This functionality is enabled using the {@link org.springframework.integration.context.metadata.MetadataPersister} implementation + * Simialarly, this class handles keeping track on the latest inbound message its received and avoiding, where + * possible, redelivery of common messages. This functionality is enabled using the + * {@link MetadataPersister} implementation * * @author Josh Long * @since 2.0 */ -public abstract class AbstractInboundTwitterEndpointSupport extends AbstractEndpoint implements Lifecycle { +public abstract class AbstractInboundTwitterEndpointSupport extends AbstractEndpoint implements Lifecycle, TrackableComponent { protected volatile OAuthConfiguration configuration; protected final MessagingTemplate messagingTemplate = new MessagingTemplate(); private volatile MessageChannel requestChannel; @@ -54,33 +61,30 @@ public abstract class AbstractInboundTwitterEndpointSupport extends AbstractE protected Twitter twitter; private final Object markerGuard = new Object(); private final Object apiPermitGuard = new Object(); + + private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor(); - @SuppressWarnings("unused") public void setConfiguration(OAuthConfiguration configuration) { this.configuration = configuration; } - abstract protected void markLastStatusId(T statusId); - - abstract protected List sort(List rl); - - protected void forwardAll(ResponseList tResponses) { - List stats = new ArrayList(); - - for (T t : tResponses) - stats.add(t); - - for (T twitterResponse : sort(stats)) - forward(twitterResponse); - } - public long getMarkerId() { return markerId; } + + public String getComponentType() { + return "twitter:inbound-dm-channel-adapter"; + } + + public void setRequestChannel(MessageChannel requestChannel) { + this.messagingTemplate.setDefaultChannel(requestChannel); + this.requestChannel = requestChannel; + } @Override protected void doStart() { try { + this.historyWritingPostProcessor.setTrackableComponent(this); refresh(); } catch (Exception e) { throw new RuntimeException(e); @@ -90,10 +94,26 @@ public abstract class AbstractInboundTwitterEndpointSupport extends AbstractE protected void forward(T status) { synchronized (this.markerGuard) { Message twtMsg = MessageBuilder.withPayload(status).build(); - messagingTemplate.send(requestChannel, twtMsg); + messagingTemplate.convertAndSend(requestChannel, twtMsg, this.historyWritingPostProcessor); markLastStatusId(status); } } + + abstract protected List sort(List rl); + + abstract protected void markLastStatusId(T statusId); + + abstract protected void refresh() throws Exception; + + protected void forwardAll(ResponseList tResponses) { + List stats = new ArrayList(); + + for (T t : tResponses) + stats.add(t); + + for (T twitterResponse : sort(stats)) + forward(twitterResponse); + } @SuppressWarnings("unchecked") protected void runAsAPIRateLimitsPermit(ApiCallback cb) @@ -148,8 +168,6 @@ public abstract class AbstractInboundTwitterEndpointSupport extends AbstractE return markerId > -1; } - abstract protected void refresh() throws Exception; - @Override protected void onInit() throws Exception { messagingTemplate.afterPropertiesSet(); @@ -162,12 +180,6 @@ public abstract class AbstractInboundTwitterEndpointSupport extends AbstractE protected void doStop() { } - @SuppressWarnings("unused") - public void setRequestChannel(MessageChannel requestChannel) { - this.messagingTemplate.setDefaultChannel(requestChannel); - this.requestChannel = requestChannel; - } - /** * Hook for clients to run logic when the API rate limiting lets us *

@@ -178,4 +190,9 @@ public abstract class AbstractInboundTwitterEndpointSupport extends AbstractE public static interface ApiCallback { void run(C t, Twitter twitter) throws Exception; } + + @Override + public void setShouldTrack(boolean shouldTrack) { + this.historyWritingPostProcessor.setShouldTrack(shouldTrack); + } } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractOutboundTwitterEndpointSupport.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractOutboundTwitterEndpointSupport.java index 4b2de425ec..ebe9b2233e 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractOutboundTwitterEndpointSupport.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractOutboundTwitterEndpointSupport.java @@ -15,10 +15,10 @@ */ package org.springframework.integration.twitter; -import org.springframework.integration.core.MessageHandler; -import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.twitter.oauth.OAuthConfiguration; import org.springframework.util.Assert; + import twitter4j.Twitter; @@ -27,12 +27,11 @@ import twitter4j.Twitter; * * @author Josh Long */ -public abstract class AbstractOutboundTwitterEndpointSupport extends AbstractEndpoint implements MessageHandler { +public abstract class AbstractOutboundTwitterEndpointSupport extends AbstractMessageHandler { protected volatile OAuthConfiguration configuration; protected volatile Twitter twitter; - protected volatile StatusUpdateSupport statusUpdateSupport = new StatusUpdateSupport(); + protected final StatusUpdateOptboundMessageMapper statusUpdateSupport = new StatusUpdateOptboundMessageMapper(); - @SuppressWarnings("unused") public void setConfiguration(OAuthConfiguration configuration) { this.configuration = configuration; } @@ -44,13 +43,4 @@ public abstract class AbstractOutboundTwitterEndpointSupport extends AbstractEnd Assert.notNull(this.twitter, "'twitter' can't be null"); } - - @Override - protected void doStart() { - } - - @Override - protected void doStop() { - } - } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/OutboundDirectMessageStatusMessageHandler.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/OutboundDirectMessageStatusMessageHandler.java index 036ba12dfe..9fbfcc100a 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/OutboundDirectMessageStatusMessageHandler.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/OutboundDirectMessageStatusMessageHandler.java @@ -16,10 +16,8 @@ package org.springframework.integration.twitter; import org.springframework.integration.Message; -import org.springframework.integration.MessageDeliveryException; -import org.springframework.integration.MessageHandlingException; -import org.springframework.integration.MessageRejectedException; import org.springframework.util.Assert; + import twitter4j.TwitterException; @@ -31,7 +29,9 @@ import twitter4j.TwitterException; * @see twitter4j.Twitter */ public class OutboundDirectMessageStatusMessageHandler extends AbstractOutboundTwitterEndpointSupport { - public void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException { + + @Override + protected void handleMessageInternal(Message message) throws Exception { try { String txt = (String) message.getPayload(); Object toUser = diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/OutboundUpdatedStatusMessageHandler.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/OutboundUpdatedStatusMessageHandler.java index 21878574ac..cd2f5a6c3d 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/OutboundUpdatedStatusMessageHandler.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/OutboundUpdatedStatusMessageHandler.java @@ -16,10 +16,8 @@ package org.springframework.integration.twitter; import org.springframework.integration.Message; -import org.springframework.integration.MessageDeliveryException; -import org.springframework.integration.MessageHandlingException; -import org.springframework.integration.MessageRejectedException; import org.springframework.util.Assert; + import twitter4j.StatusUpdate; @@ -30,15 +28,11 @@ import twitter4j.StatusUpdate; * @since 2.0 */ public class OutboundUpdatedStatusMessageHandler extends AbstractOutboundTwitterEndpointSupport { - public void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException { - try { - StatusUpdate statusUpdate = this.statusUpdateSupport.fromMessage(message); - Assert.notNull(statusUpdate, "couldn't send message, unable to build a StatusUpdate instance correctly"); - this.twitter.updateStatus(statusUpdate); - } catch (Throwable e) { - this.logger.debug(e); - throw new RuntimeException(e); - } + @Override + protected void handleMessageInternal(Message message) throws Exception { + StatusUpdate statusUpdate = this.statusUpdateSupport.fromMessage(message); + Assert.notNull(statusUpdate, "couldn't send message, unable to build a StatusUpdate instance correctly"); + this.twitter.updateStatus(statusUpdate); } } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/StatusUpdateSupport.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/StatusUpdateOptboundMessageMapper.java similarity index 82% rename from spring-integration-twitter/src/main/java/org/springframework/integration/twitter/StatusUpdateSupport.java rename to spring-integration-twitter/src/main/java/org/springframework/integration/twitter/StatusUpdateOptboundMessageMapper.java index 5c2c000ff5..c0aaf45cc5 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/StatusUpdateSupport.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/StatusUpdateOptboundMessageMapper.java @@ -16,7 +16,10 @@ package org.springframework.integration.twitter; import org.springframework.integration.Message; +import org.springframework.integration.MessageHandlingException; +import org.springframework.integration.mapping.OutboundMessageMapper; import org.springframework.util.StringUtils; + import twitter4j.GeoLocation; import twitter4j.StatusUpdate; @@ -29,16 +32,15 @@ import twitter4j.StatusUpdate; * @see org.springframework.integration.twitter.TwitterHeaders * @since 2.0 */ -public class StatusUpdateSupport { + +public class StatusUpdateOptboundMessageMapper implements OutboundMessageMapper{ /** * {@link StatusUpdate} instances are used to drive status updates. * * @param message the inbound messages * @return a {@link StatusUpdate} that's been materialized from the inbound message - * @throws Throwable thrown if something goes wrong */ - public StatusUpdate fromMessage(Message message) - throws Throwable { + public StatusUpdate fromMessage(Message message) { Object payload = message.getPayload(); StatusUpdate statusUpdate = null; @@ -77,9 +79,12 @@ public class StatusUpdateSupport { } } } - - if (payload instanceof StatusUpdate) { + else if (payload instanceof StatusUpdate) { statusUpdate = (StatusUpdate) payload; + } + else { + throw new MessageHandlingException(message, "Failde to create StatusUpdate from the payload of type: " + message.getPayload().getClass() + + " Only java.lang.String or twitter4j.StatusUpdate is currently supported"); } return statusUpdate; diff --git a/spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/twitter_connection_using_ns.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/SimpleTwitterTestClient-context.xml similarity index 100% rename from spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/twitter_connection_using_ns.xml rename to spring-integration-twitter/src/test/java/org/springframework/integration/twitter/SimpleTwitterTestClient-context.xml diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/SimpleTwitterTestClient.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/SimpleTwitterTestClient.java index 7e251e7df3..09684b83a6 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/SimpleTwitterTestClient.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/SimpleTwitterTestClient.java @@ -18,7 +18,7 @@ import java.util.Collection; * * @author Josh Long */ -@ContextConfiguration(locations = "org/springframework/integration/twitter/twitter_connection_using_ns.xml") +@ContextConfiguration public class SimpleTwitterTestClient { private Twitter twitter; @Autowired diff --git a/spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/receiving_dms_using_ns.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace-context.xml similarity index 85% rename from spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/receiving_dms_using_ns.xml rename to spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace-context.xml index c661db1a0a..60a5166338 100644 --- a/spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/receiving_dms_using_ns.xml +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace-context.xml @@ -27,34 +27,35 @@ xmlns:lang="http://www.springframework.org/schema/lang" xmlns:twitter="http://www.springframework.org/schema/integration/twitter" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd - http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd + http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd http://www.springframework.org/schema/tool http://www.springframework.org/schema/tool/spring-tool-3.0.xsd http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-3.0.xsd http://www.springframework.org/schema/integration/twitter http://www.springframework.org/schema/integration/twitter/spring-integration-twitter.xsd"> - + + + - + - + consumer-secret="${twitter.oauth.consumerSecret}"/> - diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace.java index 8574f42364..3ce38bb37d 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace.java @@ -25,9 +25,7 @@ import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; /** * @author Josh Long */ -@ContextConfiguration(locations = { - "/org/springframework/integration/twitter/receiving_dms_using_ns.xml"} -) +@ContextConfiguration public class TestRecievingUsingNamespace extends AbstractJUnit4SpringContextTests { @Autowired private TwitterAnnouncer twitterAnnouncer; diff --git a/spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/sending_dms_using_ns.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingDMsUsingNamespace-context.xml similarity index 100% rename from spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/sending_dms_using_ns.xml rename to spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingDMsUsingNamespace-context.xml diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingDMsUsingNamespace.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingDMsUsingNamespace.java index 2dd6d8b7d5..fb254d6707 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingDMsUsingNamespace.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingDMsUsingNamespace.java @@ -33,9 +33,7 @@ import twitter4j.GeoLocation; /** * @author Josh Long */ -@ContextConfiguration(locations = { - "/org/springframework/integration/twitter/sending_dms_using_ns.xml"} -) +@ContextConfiguration public class TestSendingDMsUsingNamespace extends AbstractJUnit4SpringContextTests { private volatile MessagingTemplate messagingTemplate = new MessagingTemplate(); @Value("#{out}") diff --git a/spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/sending_updates_using_ns.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingUpdatesUsingNamespace-context.xml similarity index 97% rename from spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/sending_updates_using_ns.xml rename to spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingUpdatesUsingNamespace-context.xml index 360e2973c6..758fdb830d 100644 --- a/spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/sending_updates_using_ns.xml +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingUpdatesUsingNamespace-context.xml @@ -37,7 +37,7 @@ base-package="org.springframework.integration.twitter"/> mb = MessageBuilder.withPayload("'Hello world!', from the Spring Integration outbound Twitter adapter") + MessageBuilder mb = MessageBuilder.withPayload("test message 1") .setHeader(TwitterHeaders.TWITTER_IN_REPLY_TO_STATUS_ID, 21927437001L) .setHeader(TwitterHeaders.TWITTER_GEOLOCATION, new GeoLocation(-76.226823, 23.642465)) // antarctica .setHeader(TwitterHeaders.TWITTER_DISPLAY_COORDINATES, true); diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TwitterAnnouncer.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TwitterAnnouncer.java index 5bf6b01fc8..4a7c6fb06c 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TwitterAnnouncer.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TwitterAnnouncer.java @@ -1,6 +1,7 @@ package org.springframework.integration.twitter; import org.springframework.stereotype.Component; + import twitter4j.DirectMessage; import twitter4j.Status; diff --git a/spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/receiving_replies_using_ns.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/receiving_replies_using_ns.xml similarity index 100% rename from spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/receiving_replies_using_ns.xml rename to spring-integration-twitter/src/test/java/org/springframework/integration/twitter/receiving_replies_using_ns.xml diff --git a/spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/receiving_updates_using_ns.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/receiving_updates_using_ns.xml similarity index 100% rename from spring-integration-twitter/src/test/resources/org/springframework/integration/twitter/receiving_updates_using_ns.xml rename to spring-integration-twitter/src/test/java/org/springframework/integration/twitter/receiving_updates_using_ns.xml