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 fc4368a83c..92462ed722 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 @@ -13,14 +13,12 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.integration.twitter; -import java.util.ArrayList; -import java.util.List; - 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.core.MessagingTemplate; @@ -29,190 +27,190 @@ import org.springframework.integration.history.HistoryWritingMessagePostProcesso 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 twitter4j.RateLimitStatus; import twitter4j.ResponseList; import twitter4j.Twitter; +import java.util.ArrayList; +import java.util.List; + + /** - * There are a lot of operations that are common to receiving the various types of messages when using the + * 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 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 + * 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 + * 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.MetadataStore} implementation * * @author Josh Long * @since 2.0 */ public abstract class AbstractInboundTwitterEndpointSupport extends AbstractEndpoint implements Lifecycle, TrackableComponent { + protected volatile OAuthConfiguration configuration; + protected final MessagingTemplate messagingTemplate = new MessagingTemplate(); + private volatile MessageChannel requestChannel; + protected volatile long markerId = -1; + protected Twitter twitter; + private final Object markerGuard = new Object(); + private final Object apiPermitGuard = new Object(); + private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = + new HistoryWritingMessagePostProcessor(); - protected volatile OAuthConfiguration configuration; + public void setConfiguration(OAuthConfiguration configuration) { + this.configuration = configuration; + } - protected final MessagingTemplate messagingTemplate = new MessagingTemplate(); + abstract protected void markLastStatusId(T statusId); - private volatile MessageChannel requestChannel; + abstract protected List sort(List rl); - protected volatile long markerId = -1; + protected void forwardAll(List tResponses) { + List stats = new ArrayList(); - protected Twitter twitter; + for (T t : tResponses) + stats.add(t); - private final Object markerGuard = new Object(); + for (T twitterResponse : sort(stats)) + forward(twitterResponse); + } - private final Object apiPermitGuard = new Object(); - - private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor(); + public long getMarkerId() { + return markerId; + } - public void setConfiguration(OAuthConfiguration configuration) { - this.configuration = configuration; - } + public String getComponentType() { + return "twitter:inbound-dm-channel-adapter"; + } + public void setRequestChannel(MessageChannel requestChannel) { + this.messagingTemplate.setDefaultChannel(requestChannel); + this.requestChannel = requestChannel; + } - abstract protected void markLastStatusId(T statusId); + @Override + protected void doStart() { + try { + this.historyWritingPostProcessor.setTrackableComponent(this); + refresh(); + } catch (Exception e) { + throw new RuntimeException(e); + } + } - abstract protected List sort( List rl); + protected void forward(T status) { + synchronized (this.markerGuard) { + Message twtMsg = MessageBuilder.withPayload(status).build(); + messagingTemplate.convertAndSend(requestChannel, twtMsg, + this.historyWritingPostProcessor); + markLastStatusId(status); + } + } - protected void forwardAll( List tResponses) { - List stats = new ArrayList(); + abstract protected void refresh() throws Exception; - for (T t : tResponses) - stats.add(t); + protected void forwardAll(ResponseList tResponses) { + List stats = new ArrayList(); - for (T twitterResponse : sort(stats)) - forward(twitterResponse); - } + for (T t : tResponses) + stats.add(t); - 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; - } + for (T twitterResponse : sort(stats)) + forward(twitterResponse); + } - @Override - protected void doStart() { - try { - this.historyWritingPostProcessor.setTrackableComponent(this); - refresh(); - } catch (Exception e) { - throw new RuntimeException(e); - } - } + @SuppressWarnings("unchecked") + protected void runAsAPIRateLimitsPermit(ApiCallback cb) + throws Exception { + synchronized (this.apiPermitGuard) { + while (waitUntilPullAvailable()) { + if (logger.isDebugEnabled()) { + logger.debug("have room to make an API request now"); + } - protected void forward(T status) { - synchronized (this.markerGuard) { - Message twtMsg = MessageBuilder.withPayload(status).build(); - 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; + cb.run(this, twitter); + } + } + } - protected void forwardAll(ResponseList tResponses) { - List stats = new ArrayList(); + protected boolean handleReceivingRateLimitStatus( + RateLimitStatus rateLimitStatus) { + try { + int secondsUntilReset = rateLimitStatus.getSecondsUntilReset(); + int remainingHits = rateLimitStatus.getRemainingHits(); - for (T t : tResponses) - stats.add(t); + if (remainingHits == 0) { + logger.debug( + "rate status limit service returned 0 for the remaining hits value"); - for (T twitterResponse : sort(stats)) - forward(twitterResponse); - } + return false; + } - @SuppressWarnings("unchecked") - protected void runAsAPIRateLimitsPermit(ApiCallback cb) - throws Exception { - synchronized (this.apiPermitGuard) { - while (waitUntilPullAvailable()) { - if (logger.isDebugEnabled()) { - logger.debug("have room to make an API request now"); - } + if (secondsUntilReset == 0) { + logger.debug( + "rate status limit service returned 0 for the seconds until reset period value"); - cb.run(this, twitter); - } - } - } + return false; + } - protected boolean handleReceivingRateLimitStatus(RateLimitStatus rateLimitStatus) { - try { - int secondsUntilReset = rateLimitStatus.getSecondsUntilReset(); - int remainingHits = rateLimitStatus.getRemainingHits(); + int secondsUntilWeCanPullAgain = secondsUntilReset / remainingHits; + long msUntilWeCanPullAgain = secondsUntilWeCanPullAgain * 1000; - if (remainingHits == 0) { - logger.debug("rate status limit service returned 0 for the remaining hits value"); + logger.debug("need to Thread.sleep() " + + secondsUntilWeCanPullAgain + + " seconds until the next timeline pull. Have " + remainingHits + + " remaining pull this rate period. The period ends in " + + secondsUntilReset); - return false; - } + Thread.sleep(msUntilWeCanPullAgain); + } catch (Throwable throwable) { + logger.debug("encountered an error when" + + " trying to refresh the timeline: " + + ExceptionUtils.getFullStackTrace(throwable)); + } - if (secondsUntilReset == 0) { - logger.debug("rate status limit service returned 0 for the seconds until reset period value"); + return true; + } - return false; - } + protected boolean waitUntilPullAvailable() throws Exception { + return this.handleReceivingRateLimitStatus(this.twitter.getRateLimitStatus()); + } - int secondsUntilWeCanPullAgain = secondsUntilReset / remainingHits; - long msUntilWeCanPullAgain = secondsUntilWeCanPullAgain * 1000; + protected boolean hasMarkedStatus() { + return markerId > -1; + } - logger.debug("need to Thread.sleep() " + secondsUntilWeCanPullAgain + - " seconds until the next timeline pull. Have " + remainingHits + - " remaining pull this rate period. The period ends in " + secondsUntilReset); + @Override + protected void onInit() throws Exception { + messagingTemplate.afterPropertiesSet(); + Assert.notNull(this.configuration, "'configuration' can't be null"); + this.twitter = this.configuration.getTwitter(); + Assert.notNull(this.twitter, "'twitter' instance can't be null"); + } - Thread.sleep(msUntilWeCanPullAgain); - } catch (Throwable throwable) { - logger.debug("encountered an error when" + " trying to refresh the timeline: " + ExceptionUtils.getFullStackTrace(throwable)); - } + @Override + protected void doStop() { + } - return true; - } + public void setShouldTrack(boolean shouldTrack) { + this.historyWritingPostProcessor.setShouldTrack(shouldTrack); + } - protected boolean waitUntilPullAvailable() throws Exception { - return this.handleReceivingRateLimitStatus(this.twitter.getRateLimitStatus()); - } - - protected boolean hasMarkedStatus() { - return markerId > -1; - } - - @Override - protected void onInit() throws Exception { - messagingTemplate.afterPropertiesSet(); - Assert.notNull(this.configuration, "'configuration' can't be null"); - this.twitter = this.configuration.getTwitter(); - Assert.notNull(this.twitter, "'twitter' instance can't be null"); - } - - @Override - protected void doStop() { - } - - /** - * Hook for clients to run logic when the API rate limiting lets us - *

- * Simply register your callback using #runAsAPIRateLimitsPermit - * - * @param - */ - public static interface ApiCallback { - void run(C t, Twitter twitter) throws Exception; - } - - public void setShouldTrack(boolean shouldTrack) { - this.historyWritingPostProcessor.setShouldTrack(shouldTrack); - } + /** + * Hook for clients to run logic when the API rate limiting lets us + *

+ * Simply register your callback using #runAsAPIRateLimitsPermit + * + * @param + */ + public static interface ApiCallback { + void run(C t, Twitter twitter) throws Exception; + } } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterStatusEndpointSupport.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterStatusEndpointSupport.java index 3c111ab766..352a16317f 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterStatusEndpointSupport.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/AbstractInboundTwitterStatusEndpointSupport.java @@ -36,6 +36,7 @@ import java.util.List; */ abstract public class AbstractInboundTwitterStatusEndpointSupport extends AbstractInboundTwitterEndpointSupport { + private Comparator statusComparator = new Comparator() { public int compare(Status status, Status status1) { return status.getCreatedAt().compareTo(status1.getCreatedAt()); diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/InboundDirectMessageStatusEndpoint.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/InboundDirectMessageStatusEndpoint.java index c51fb8f667..04277ce880 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/InboundDirectMessageStatusEndpoint.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/InboundDirectMessageStatusEndpoint.java @@ -35,6 +35,7 @@ import java.util.List; * @author Josh Long */ public class InboundDirectMessageStatusEndpoint extends AbstractInboundTwitterEndpointSupport { + private Comparator dmComparator = new Comparator() { public int compare(DirectMessage directMessage, DirectMessage directMessage1) { return directMessage.getCreatedAt().compareTo(directMessage1.getCreatedAt()); diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace-context.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace-context.xml index 60a5166338..98fdeb1d28 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace-context.xml +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace-context.xml @@ -39,14 +39,12 @@ base-package="org.springframework.integration.twitter"/> - - + + - + + + + + + + + + + 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 3ce38bb37d..e1d5e4617c 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 @@ -15,7 +15,6 @@ */ package org.springframework.integration.twitter; -import org.junit.Ignore; import org.junit.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.ContextConfiguration; @@ -23,15 +22,17 @@ import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; /** + * provides a simple example of a command-line client to receive DMs, messages, timeline updates. + * * @author Josh Long */ @ContextConfiguration public class TestRecievingUsingNamespace extends AbstractJUnit4SpringContextTests { - @Autowired - private TwitterAnnouncer twitterAnnouncer; + + @Autowired private TwitterAnnouncer twitterAnnouncer; @Test - @Ignore + //@Ignore public void testIt() throws Throwable { long ctr = 0; long s = 1000; diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingUpdatesUsingNamespace.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingUpdatesUsingNamespace.java index 0ec7144e02..e721f4c549 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingUpdatesUsingNamespace.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestSendingUpdatesUsingNamespace.java @@ -23,6 +23,7 @@ import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.twitter.model.Twitter4jGeoLocationImpl; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; @@ -32,7 +33,7 @@ import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; @ContextConfiguration public class TestSendingUpdatesUsingNamespace extends AbstractJUnit4SpringContextTests { - private StatusUpdateSupport statusUpdateSupport = new StatusUpdateSupport(); + private MessagingTemplate messagingTemplate = new MessagingTemplate(); @@ -44,7 +45,7 @@ public class TestSendingUpdatesUsingNamespace extends AbstractJUnit4SpringContex MessageBuilder mb = MessageBuilder.withPayload("test message 1") .setHeader(TwitterHeaders.TWITTER_IN_REPLY_TO_STATUS_ID, 21927437001L) .setHeader(TwitterHeaders.TWITTER_GEOLOCATION, - this.statusUpdateSupport.fromLatitudeLongitudePair(-76.226823, 23.642465)) // antarctica + new Twitter4jGeoLocationImpl(-76.226823, 23.642465)) // antarctica .setHeader(TwitterHeaders.TWITTER_DISPLAY_COORDINATES, true); Message m = mb.build(); this.messagingTemplate.send(this.channel, m); 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 22ae6ff576..79be70167f 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 @@ -4,12 +4,6 @@ import org.springframework.integration.twitter.model.DirectMessage; import org.springframework.integration.twitter.model.Status; import org.springframework.stereotype.Component; -<<<<<<< HEAD -======= -import twitter4j.DirectMessage; -import twitter4j.Status; ->>>>>>> 64fec64d4095a6e793739607816d63d600e1ed0c - @Component public class TwitterAnnouncer { @@ -22,7 +16,7 @@ public class TwitterAnnouncer { System.out.println("A tweet mentioning (or replying) to " + "you was received having text " + s.getText() + " from " + s.getSource()); } - public void friendsTimelineUpdated(Status t) { - System.out.println("Received " + t.getText() + " from " + t.getSource()); + public void updates(Status t) { + System.out.println("Received timeline update: " + t.getText() + " from " + t.getSource()); } }