diff --git a/spring-integration-twitter/pom.xml b/spring-integration-twitter/pom.xml index 18aa36ed62..cb2b66c141 100644 --- a/spring-integration-twitter/pom.xml +++ b/spring-integration-twitter/pom.xml @@ -81,5 +81,9 @@ commons-io 1.4 + + org.mockito + mockito-all + diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/twitter/Twitter4jDirectMessage.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/twitter/Twitter4jDirectMessage.java index 985db37bee..f6b0322f09 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/twitter/Twitter4jDirectMessage.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/core/twitter/Twitter4jDirectMessage.java @@ -1,37 +1,39 @@ /* - * Copyright 2010 the original author or authors + * 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 + * 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 + * 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. + * 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.core.twitter; import org.springframework.integration.twitter.core.User; +import org.springframework.util.Assert; + import twitter4j.DirectMessage; import java.util.Date; - /** - * implementation of the {@link org.springframework.integration.twitter.core.DirectMessage} interface - * that wraps, and works with, a {@link twitter4j.DirectMessage} instance. + * {@link twitter4j.DirectMessage} wrapper for {@link org.springframework.integration.twitter.core.DirectMessage}. * * @author Josh Long + * @author Oleg ZHurakousky * @since 2.0 */ public class Twitter4jDirectMessage implements org.springframework.integration.twitter.core.DirectMessage { private DirectMessage directMessage; public Twitter4jDirectMessage(DirectMessage directMessage) { + Assert.notNull(directMessage, "'directMessage' must not be null"); this.directMessage = directMessage; } 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 eff55d2954..550504d6d4 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 @@ -1,98 +1,75 @@ /* - * Copyright 2010 the original author or authors + * 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 + * 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 + * 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. + * 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.inbound; -import org.apache.commons.lang.exception.ExceptionUtils; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ScheduledFuture; import org.springframework.context.Lifecycle; - import org.springframework.integration.Message; 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 twitter4j.RateLimitStatus; -import twitter4j.ResponseList; import twitter4j.Twitter; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.Executor; -import java.util.concurrent.Executors; -import java.util.concurrent.ThreadFactory; - /** - * 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 + * Abstract class that defines common operations for receiving various types of messages when using the + * Twitter API. + * This class also 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 + * @author Oleg Zhurakousky + * * @since 2.0 */ -public abstract class AbstractInboundTwitterEndpointSupport extends MessageProducerSupport implements Lifecycle, TrackableComponent, Runnable { - +public abstract class AbstractInboundTwitterEndpointSupport extends MessageProducerSupport implements Lifecycle, TrackableComponent{ + protected volatile OAuthConfiguration configuration; protected volatile long markerId = -1; protected Twitter twitter; private final Object markerGuard = new Object(); - private final Object apiPermitGuard = new Object(); + private volatile ScheduledFuture twitterUpdatePollingTask; private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor(); - protected Executor taskExecutor; - protected int poolSize = 1; - - public void setPoolSize(int poolSize) { - this.poolSize = poolSize; - } - - protected void checkTaskExecutor(final String threadName) { - if (this.taskExecutor == null) { - this.taskExecutor = Executors.newFixedThreadPool(this.poolSize, - new ThreadFactory() { - public Thread newThread(Runnable runner) { - Thread thread = new Thread(runner); - thread.setName(threadName); - thread.setDaemon(true); - - return thread; - } - }); - } - } - - public void setConfiguration(OAuthConfiguration configuration) { - this.configuration = configuration; - } - abstract protected void markLastStatusId(T statusId); abstract protected List sort(List rl); + + abstract Runnable getApiCallback(); + + public void setConfiguration(OAuthConfiguration configuration) { + this.configuration = configuration; + } + + public void setShouldTrack(boolean shouldTrack) { + this.historyWritingPostProcessor.setShouldTrack(shouldTrack); + } + + public long getMarkerId() { + return markerId; + } @Override protected void onInit() { @@ -105,28 +82,28 @@ public abstract class AbstractInboundTwitterEndpointSupport extends MessagePr protected void forwardAll(List tResponses) { List stats = new ArrayList(); - for (T t : tResponses) - stats.add(t); - - for (T twitterResponse : sort(stats)) - forward(twitterResponse); + for (T t : tResponses){ + stats.add(t); + } + + for (T twitterResponse : sort(stats)) { + forward(twitterResponse); + } } - public long getMarkerId() { - return markerId; - } - - abstract public String getComponentType(); - @Override protected void doStart() { - historyWritingPostProcessor.setTrackableComponent(this); - - checkTaskExecutor(getClass().getName() + "-taskExecutor"); - - taskExecutor.execute(this); + historyWritingPostProcessor.setTrackableComponent(this); + RateLimitStatusTrigger trigger = new RateLimitStatusTrigger(this.twitter); + Runnable apiCallback = this.getApiCallback(); + twitterUpdatePollingTask = this.getTaskScheduler().schedule(apiCallback, trigger); } + @Override + protected void doStop() { + twitterUpdatePollingTask.cancel(true); + } + protected void forward(T status) { synchronized (this.markerGuard) { Message twtMsg = MessageBuilder.withPayload(status).build(); @@ -136,107 +113,7 @@ public abstract class AbstractInboundTwitterEndpointSupport extends MessagePr markLastStatusId(status); } } - - /** - * this is execu - */ - public void run() { - try { - beginPolling(); - } catch (Exception e) { - throw new RuntimeException(e); - } - } - - abstract protected void beginPolling() 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 apiCallback) - throws Exception { - synchronized (this.apiPermitGuard) { - while (waitUntilPullAvailable()) { - if (logger.isDebugEnabled()) { - logger.debug("have room to make an API request now"); - } - - apiCallback.run(this, twitter); - } - } - } - - protected boolean handleReceivingRateLimitStatus( - RateLimitStatus rateLimitStatus) { - try { - int secondsUntilReset = rateLimitStatus.getSecondsUntilReset(); - int remainingHits = rateLimitStatus.getRemainingHits(); - - if (remainingHits == 0) { - logger.debug( - "rate status limit service returned 0 for the remaining hits value"); - - return false; - } - - if (secondsUntilReset == 0) { - logger.debug( - "rate status limit service returned 0 for the seconds until reset period value"); - - return false; - } - - int secondsUntilWeCanPullAgain = secondsUntilReset / remainingHits; - long msUntilWeCanPullAgain = secondsUntilWeCanPullAgain * 1000; - - 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); - - Thread.sleep(msUntilWeCanPullAgain); - } catch (Throwable throwable) { - logger.debug( - "encountered an error when trying to refresh the timeline: " + - ExceptionUtils.getFullStackTrace(throwable)); - } - - return true; - } - - protected boolean waitUntilPullAvailable() throws Exception { - return this.handleReceivingRateLimitStatus(this.twitter.getRateLimitStatus()); - } - protected boolean hasMarkedStatus() { return markerId > -1; } - - @Override - protected void doStop() { - } - - 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/inbound/AbstractInboundTwitterStatusEndpointSupport.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterStatusEndpointSupport.java index 3d34694675..2be4967d30 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 @@ -1,17 +1,17 @@ /* - * Copyright 2010 the original author or authors + * 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 + * 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 + * 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. + * 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.inbound; import org.springframework.integration.twitter.core.Status; @@ -31,8 +31,6 @@ 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()); @@ -41,8 +39,9 @@ abstract public class AbstractInboundTwitterStatusEndpointSupport extends Abstra protected List fromTwitter4jStatuses(List stats) { List fwd = new ArrayList(); - for (twitter4j.Status s : stats) + for (twitter4j.Status s : stats) { fwd.add(new Twitter4jStatus(s)); + } return fwd; } 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 9a7e2eb81c..248a38f428 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 @@ -1,35 +1,37 @@ /* - * Copyright 2010 the original author or authors + * 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 + * 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 + * 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. + * 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.inbound; -import org.springframework.integration.twitter.core.DirectMessage; -import org.springframework.integration.twitter.core.twitter.Twitter4jDirectMessage; -import twitter4j.Paging; -import twitter4j.Twitter; - import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; import java.util.List; +import org.springframework.integration.MessagingException; +import org.springframework.integration.twitter.core.DirectMessage; +import org.springframework.integration.twitter.core.twitter.Twitter4jDirectMessage; + +import twitter4j.Paging; /** * This class handles support for receiving DMs (direct messages) using Twitter. * * @author Josh Long + * @author Oleg Zhurakousky + * @since 2.0 */ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpointSupport { @@ -62,22 +64,35 @@ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpoint } @Override - protected void beginPolling() throws Exception { + Runnable getApiCallback() { + Runnable apiCallback = new Runnable() { + @Override + public void run() { + try { + long sinceId = getMarkerId(); + + List dms = !hasMarkedStatus() + ? twitter.getDirectMessages() + : twitter.getDirectMessages(new Paging(sinceId)); - this.runAsAPIRateLimitsPermit(new ApiCallback() { - public void run(InboundDirectMessageEndpoint t, Twitter twitter) - throws Exception { - - List dms = !hasMarkedStatus() ? t.twitter.getDirectMessages() : t.twitter.getDirectMessages(new Paging(t.getMarkerId())); - - List dmsToFwd = new ArrayList(); - - for( twitter4j.DirectMessage dm : dms) - dmsToFwd.add( new Twitter4jDirectMessage( dm)); - - forwardAll( dmsToFwd ); + List dmsToFwd = new ArrayList(); + + for( twitter4j.DirectMessage dm : dms) { + dmsToFwd.add( new Twitter4jDirectMessage( dm)); + } + forwardAll(dmsToFwd); + } catch (Exception e) { + e.printStackTrace(); + if (e instanceof RuntimeException){ + throw (RuntimeException)e; + } + else { + throw new MessagingException("Failed to poll for Twitter mentions updates", e); + } + } } - }); + }; + return apiCallback; } } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundMentionEndpoint.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundMentionEndpoint.java index e269dc097b..ef4de53c0d 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundMentionEndpoint.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundMentionEndpoint.java @@ -1,30 +1,31 @@ /* - * Copyright 2010 the original author or authors + * 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 + * 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 + * 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. + * 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.inbound; -import twitter4j.Paging; -import twitter4j.Twitter; - import java.util.List; +import org.springframework.integration.MessagingException; + +import twitter4j.Paging; /** * Handles forwarding all new {@link twitter4j.Status} that are 'replies' or 'mentions' to some other tweet. * * @author Josh Long + * @author Oleg Zhurakousky */ public class InboundMentionEndpoint extends AbstractInboundTwitterStatusEndpointSupport { @@ -32,20 +33,29 @@ public class InboundMentionEndpoint extends AbstractInboundTwitterStatusEndpoint public String getComponentType() { return "twitter:inbound-mention-channel-adapter"; } - @Override - protected void beginPolling() throws Exception { - this.runAsAPIRateLimitsPermit(new ApiCallback() { - - public void run(InboundMentionEndpoint ctx, Twitter twitter) throws Exception { - List stats = (!hasMarkedStatus()) - ? twitter.getMentions() - : twitter.getMentions(new Paging(ctx.getMarkerId())); - forwardAll( fromTwitter4jStatuses( stats)); + Runnable getApiCallback() { + Runnable apiCallback = new Runnable() { + @Override + public void run() { + try { + long sinceId = getMarkerId(); + List stats = (!hasMarkedStatus()) + ? twitter.getMentions() + : twitter.getMentions(new Paging(sinceId)); + System.out.println("Polling. . . ."); + forwardAll( fromTwitter4jStatuses( stats)); + } catch (Exception e) { + if (e instanceof RuntimeException){ + throw (RuntimeException)e; + } + else { + throw new MessagingException("Failed to poll for Twitter mentions updates", e); + } + } } - - }); + }; + return apiCallback; } - } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundTimelineUpdateEndpoint.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundTimelineUpdateEndpoint.java index 65d3ce2a64..11fa68509e 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundTimelineUpdateEndpoint.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundTimelineUpdateEndpoint.java @@ -1,22 +1,23 @@ /* - * Copyright 2010 the original author or authors + * 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 + * 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 + * 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. + * 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.inbound; +import org.springframework.integration.MessagingException; + import twitter4j.Paging; -import twitter4j.Twitter; /** @@ -24,6 +25,7 @@ import twitter4j.Twitter; * as messages. It has support for dynamic throttling of API requests. * * @author Josh Long + * @author Oleg Zhurakousky * @since 2.0 */ public class InboundTimelineUpdateEndpoint extends AbstractInboundTwitterStatusEndpointSupport { @@ -34,14 +36,25 @@ public class InboundTimelineUpdateEndpoint extends AbstractInboundTwitterStatusE } @Override - protected void beginPolling() throws Exception { - this.runAsAPIRateLimitsPermit(new ApiCallback() { - public void run(InboundTimelineUpdateEndpoint t, Twitter twitter) - throws Exception { - forwardAll( fromTwitter4jStatuses(!t.hasMarkedStatus() - ? twitter.getFriendsTimeline() : - twitter.getFriendsTimeline(new Paging(t.getMarkerId())))); + Runnable getApiCallback() { + Runnable apiCallback = new Runnable() { + @Override + public void run() { + try { + long sinceId = getMarkerId(); + forwardAll( fromTwitter4jStatuses(!hasMarkedStatus() + ? twitter.getFriendsTimeline() : + twitter.getFriendsTimeline(new Paging(sinceId)))); + } catch (Exception e) { + if (e instanceof RuntimeException){ + throw (RuntimeException)e; + } + else { + throw new MessagingException("Failed to poll for Twitter mentions updates", e); + } + } } - }); + }; + return apiCallback; } } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/RateLimitStatusTrigger.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/RateLimitStatusTrigger.java new file mode 100644 index 0000000000..0654a0c47f --- /dev/null +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/RateLimitStatusTrigger.java @@ -0,0 +1,77 @@ +/* + * 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.inbound; + +import java.util.Date; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.scheduling.SchedulingException; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; +import org.springframework.util.Assert; + +import twitter4j.RateLimitStatus; +import twitter4j.Twitter; +import twitter4j.TwitterException; + +/** + * @author Oleg Zhurakousky + * @since 2.0 + */ +class RateLimitStatusTrigger implements Trigger { + protected final Log logger = LogFactory.getLog(getClass()); + private Twitter twitter; + + public RateLimitStatusTrigger(Twitter twitter){ + Assert.notNull(twitter, "'twitter' must not be null"); + this.twitter = twitter; + } + + /* (non-Javadoc) + * @see org.springframework.scheduling.Trigger#nextExecutionTime(org.springframework.scheduling.TriggerContext) + */ + @Override + public Date nextExecutionTime(TriggerContext triggerContext) { + if (triggerContext.lastCompletionTime() == null){ + return new Date(System.currentTimeMillis()); + } + try { + RateLimitStatus rateLimitStatus = twitter.getRateLimitStatus(); + int secondsUntilReset = rateLimitStatus.getSecondsUntilReset(); + int remainingHits = rateLimitStatus.getRemainingHits(); + if (remainingHits == 0) { + logger.debug( + "rate status limit service returned 0 for the remaining hits value"); + return null; + } + if (secondsUntilReset == 0) { + logger.debug( + "rate status limit service returned 0 for the seconds until reset period value"); + return null; + } + int secondsUntilWeCanPullAgain = secondsUntilReset / remainingHits; + long msUntilWeCanPullAgain = secondsUntilWeCanPullAgain * 1000; + 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 new Date(System.currentTimeMillis() + msUntilWeCanPullAgain); + } catch (TwitterException e) { + throw new SchedulingException("Failed to schedule the next Twitter update", e); + } + } +} diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/InboundDirectMessageStatusEndpointTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/InboundDirectMessageStatusEndpointTests.java new file mode 100644 index 0000000000..cae707e9d1 --- /dev/null +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/InboundDirectMessageStatusEndpointTests.java @@ -0,0 +1,121 @@ +/* + * 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; + +import static junit.framework.Assert.assertEquals; +import static junit.framework.Assert.assertNotNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.Message; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.twitter.core.twitter.Twitter4jDirectMessage; +import org.springframework.integration.twitter.inbound.InboundDirectMessageEndpoint; +import org.springframework.integration.twitter.oauth.OAuthConfiguration; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +import twitter4j.DirectMessage; +import twitter4j.Paging; +import twitter4j.RateLimitStatus; +import twitter4j.ResponseList; +import twitter4j.Twitter; + +/** + * @author Oleg Zhurakousky + * + */ +public class InboundDirectMessageStatusEndpointTests { + private DirectMessage firstMessage; + private DirectMessage secondMessage; + private Twitter twitter; + + @Test + @Ignore + /* + * In order to run this test you need to provide values to the twitter.properties file + */ + public void testUpdatesWithRealTwitter() throws Exception{ + CountDownLatch latch = new CountDownLatch(1); + new ClassPathXmlApplicationContext("TestRecievingUsingNamespace-context.xml", this.getClass()); + latch.await(10000, TimeUnit.SECONDS); + } + + @Test + public void testTwitterMockedUpdates() throws Exception{ + QueueChannel channel = new QueueChannel(); + + InboundDirectMessageEndpoint endpoint = new InboundDirectMessageEndpoint(); + endpoint.setOutputChannel(channel); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + endpoint.setTaskScheduler(scheduler); + + endpoint.setConfiguration(this.getTestConfigurationForDirectMessages()); + endpoint.afterPropertiesSet(); + endpoint.start(); + Message message1 = channel.receive(3000); + assertNotNull(message1); + assertEquals(secondMessage, ((Twitter4jDirectMessage)message1.getPayload()).getDirectMessage()); + Message message2 = channel.receive(3000); + assertNotNull(message2); + assertEquals(firstMessage, ((Twitter4jDirectMessage)message2.getPayload()).getDirectMessage()); + } + + @Before + public void prepare(){ + twitter = mock(Twitter.class); + firstMessage = mock(DirectMessage.class); + when(firstMessage.getCreatedAt()).thenReturn(new Date(5555555555L)); + when(firstMessage.getId()).thenReturn(200); + secondMessage = mock(DirectMessage.class); + when(secondMessage.getCreatedAt()).thenReturn(new Date(2222222222L)); + when(secondMessage.getId()).thenReturn(2000); + } + + @SuppressWarnings("unchecked") + private OAuthConfiguration getTestConfigurationForDirectMessages() throws Exception{ + OAuthConfiguration configuration = mock(OAuthConfiguration.class); + RateLimitStatus rateLimitStatus = mock(RateLimitStatus.class); + when(twitter.getRateLimitStatus()).thenReturn(rateLimitStatus); + when(configuration.getTwitter()).thenReturn(twitter); + + when(rateLimitStatus.getSecondsUntilReset()).thenReturn(2464); + when(rateLimitStatus.getRemainingHits()).thenReturn(250); + + ResponseList responses = mock(ResponseList.class); + List testMessages = new ArrayList(); + + testMessages.add(firstMessage); + testMessages.add(secondMessage); + + when(responses.iterator()).thenReturn(testMessages.iterator()); + when(twitter.getDirectMessages()).thenReturn(responses); + + when(twitter.getDirectMessages(Mockito.any(Paging.class))).thenReturn(responses); + return configuration; + } +} 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 deleted file mode 100644 index 09684b83a6..0000000000 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/SimpleTwitterTestClient.java +++ /dev/null @@ -1,40 +0,0 @@ -package org.springframework.integration.twitter; - -import org.junit.Assert; -import org.junit.Before; -import org.junit.Ignore; -import org.junit.Test; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.integration.twitter.oauth.OAuthConfiguration; -import org.springframework.test.context.ContextConfiguration; -import twitter4j.Status; -import twitter4j.Twitter; - -import java.util.Collection; - - -/** - * This class is used to simply demonstrating correctly factory-ing a {@link twitter4j.Twitter} instance - * - * @author Josh Long - */ -@ContextConfiguration -public class SimpleTwitterTestClient { - private Twitter twitter; - @Autowired - private volatile OAuthConfiguration oAuthConfiguration; - - @Before - public void begin() throws Exception { - this.twitter = oAuthConfiguration.getTwitter(); - } - - @Test - @Ignore - public void testConnectivity() throws Throwable { - Collection responses; - Assert.assertNotNull(this.twitter); - Assert.assertNotNull(responses = this.twitter.getFriendsTimeline()); - Assert.assertTrue(responses.size() > 0); - } -} 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 7e5f7ddc68..ea3d775a5c 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,7 +39,7 @@ base-package="org.springframework.integration.twitter"/> 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 1a56a5a95b..8c41578ad6 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,3 +1,18 @@ +/* + * 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; import org.springframework.integration.twitter.core.DirectMessage; diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/RateLimitStatusTriggerTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/RateLimitStatusTriggerTests.java new file mode 100644 index 0000000000..6b9506c9e9 --- /dev/null +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/RateLimitStatusTriggerTests.java @@ -0,0 +1,56 @@ +/* + * 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.inbound; + +import static junit.framework.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.Date; + +import org.junit.Test; +import org.springframework.scheduling.TriggerContext; + +import twitter4j.RateLimitStatus; +import twitter4j.Twitter; + +/** + * @author Oleg Zhurakousky + * + */ +public class RateLimitStatusTriggerTests { + + @Test + public void testTriggerImediateAndSubsequentExecutionTime() throws Exception{ + Twitter twitter = mock(Twitter.class); + RateLimitStatusTrigger trigger = new RateLimitStatusTrigger(twitter); + TriggerContext context = mock(TriggerContext.class); + Date currentDate = new Date(System.currentTimeMillis()); + Date nextDate = trigger.nextExecutionTime(context); + // as long as its within 1 msec we can consider it right away for the purpose of testing + assertTrue(nextDate.getTime() - currentDate.getTime() < 100); + + + RateLimitStatus rateLimitStatis = mock(RateLimitStatus.class); + when(twitter.getRateLimitStatus()).thenReturn(rateLimitStatis); + when(rateLimitStatis.getRemainingHits()).thenReturn(2000); + when(rateLimitStatis.getSecondsUntilReset()).thenReturn(4000); + + when(context.lastCompletionTime()).thenReturn(nextDate); + // based on the above values the next execution time should be at least 2000 msec + assertTrue(trigger.nextExecutionTime(context).getTime() - nextDate.getTime() > 2000); + } +} diff --git a/spring-integration-twitter/src/test/java/twitter.properties b/spring-integration-twitter/src/test/java/twitter.properties new file mode 100644 index 0000000000..d21e6292f9 --- /dev/null +++ b/spring-integration-twitter/src/test/java/twitter.properties @@ -0,0 +1,5 @@ +twitter.oauth.consumerKey= +twitter.oauth.consumerSecret= +twitter.oauth.pin= +twitter.oauth.accessToken= +twitter.oauth.accessTokenSecret= \ No newline at end of file