INT-1471, polishing Twitter modules, added MessageHistory

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.
  * <p/>
- * 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<T> extends AbstractEndpoint implements Lifecycle {
+public abstract class AbstractInboundTwitterEndpointSupport<T> 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<T> 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<T> sort(List<T> rl);
-
-	protected void forwardAll(ResponseList<T> tResponses) {
-		List<T> stats = new ArrayList<T>();
-
-		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<T> extends AbstractE
 	protected void forward(T status) {
 		synchronized (this.markerGuard) {
 			Message<T> twtMsg = MessageBuilder.withPayload(status).build();
-			messagingTemplate.send(requestChannel, twtMsg);
+			messagingTemplate.convertAndSend(requestChannel, twtMsg, this.historyWritingPostProcessor);
 			markLastStatusId(status);
 		}
 	}
+	
+	abstract protected List<T> sort(List<T> rl);
+	
+	abstract protected void markLastStatusId(T statusId);
+	
+	abstract protected void refresh() throws Exception;
+
+	protected void forwardAll(ResponseList<T> tResponses) {
+		List<T> stats = new ArrayList<T>();
+
+		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<T> 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<T> 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
 	 * <p/>
@@ -178,4 +190,9 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends AbstractE
 	public static interface ApiCallback<C> {
 		void run(C t, Twitter twitter) throws Exception;
 	}
+
+	@Override
+	public void setShouldTrack(boolean shouldTrack) {
+		this.historyWritingPostProcessor.setShouldTrack(shouldTrack);
+	}
 }

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() {
-	}
-
 }

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 =

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);
 	}
 
 }

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<StatusUpdate>{
 	/**
 	 * {@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;

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

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">
-
+	
+	<message-history/>
+	
 	<context:component-scan
 			base-package="org.springframework.integration.twitter"/>
 
 	<context:property-placeholder
-			location="file://${user.home}/Desktop/twitter.properties"
+			location="classpath:twitter.properties"
 			ignore-unresolvable="true"/>
 
 	<channel id="inbound_dm"/>
 
-	<twitter:inbound-dm-channel-adapter
-			channel="inbound_dm" twitter-connection="tc"
-			/>
+	<twitter:inbound-dm-channel-adapter id="twitterInbound"
+										channel="inbound_dm" 
+										twitter-connection="tc"/>
 
-	<twitter:twitter-connection
-			id="tc"
+	<twitter:twitter-connection id="tc"
 			access-token="${twitter.oauth.accessToken}"
 			access-token-secret="${twitter.oauth.accessTokenSecret}"
 			consumer-key="${twitter.oauth.consumerKey}"
-			consumer-secret="${twitter.oauth.consumerSecret}"
-			/>
+			consumer-secret="${twitter.oauth.consumerSecret}"/>
 
-	<service-activator input-channel="inbound_dm"
+	<service-activator id="twitterService"
+					   input-channel="inbound_dm"
 					   ref="twitterAnnouncer"
 					   method="dm"/>
 

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;

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}")

spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingUpdatesUsingNamespace-context.xml

@@ -37,7 +37,7 @@
 			base-package="org.springframework.integration.twitter"/>
 
 	<context:property-placeholder
-			location="file://${user.home}/Desktop/twitter.properties"
+			location="classpath:twitter.properties"
 			ignore-unresolvable="true"/>
         
 	<twitter:twitter-connection

spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingUpdatesUsingNamespace.java

@@ -32,9 +32,7 @@ import twitter4j.GeoLocation;
 /**
  * @author Josh Long
  */
-@ContextConfiguration(locations = {
-        "/org/springframework/integration/twitter/sending_updates_using_ns.xml"}
-)
+@ContextConfiguration
 public class TestSendingUpdatesUsingNamespace extends AbstractJUnit4SpringContextTests {
 	private MessagingTemplate messagingTemplate = new MessagingTemplate();
 	@Value("#{out}")
@@ -43,7 +41,7 @@ public class TestSendingUpdatesUsingNamespace extends AbstractJUnit4SpringContex
 	@Test
 	@Ignore
 	public void testSendingATweet() throws Throwable {
-		MessageBuilder<String> mb = MessageBuilder.withPayload("'Hello world!', from the Spring Integration outbound Twitter adapter")
+		MessageBuilder<String> 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);

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;