Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1471, Twitter polishing after making it Polling consumer

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-10-27 17:55:17 -04:00
parent 1f352ac8fd
commit bef427e6b3
8 changed files with 47 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/UpdateEndpointParser.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.twitter.config;
import static org.springframework.integration.twitter.config.TwitterNamespaceHandler.BASE_PACKAGE;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
@@ -51,8 +52,7 @@ public class UpdateEndpointParser extends AbstractPollingInboundChannelAdapterPa
}
BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(className);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "twitter-connection", "configuration");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "id", "persistentIdentifier");
String name = BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
return builder.getBeanDefinition();
return new RuntimeBeanReference(name);
}
}

43
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java
View File

@@ -16,18 +16,19 @@
package org.springframework.integration.twitter.inbound;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.history.HistoryWritingMessagePostProcessor;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.store.MetadataStore;
@@ -54,8 +55,8 @@ import twitter4j.Twitter;
* @since 2.0
*/
@SuppressWarnings("rawtypes")
public abstract class AbstractInboundTwitterEndpointSupport<T> extends IntegrationObjectSupport
implements MessageSource, Lifecycle, TrackableComponent {
public abstract class AbstractInboundTwitterEndpointSupport<T> extends AbstractEndpoint
implements MessageSource, SmartLifecycle, TrackableComponent {
private volatile MetadataStore metadataStore;
@@ -72,8 +73,6 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends Integrati
protected Twitter twitter;
private final Object markerGuard = new Object();
private volatile boolean isRunning;
private volatile ScheduledFuture<?> twitterUpdatePollingTask;
@@ -120,38 +119,35 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends Integrati
+ "." + this.configuration.getConsumerKey();
}
@SuppressWarnings("unchecked")
protected void forwardAll(List<T> tResponses) {
List<T> stats = new ArrayList<T>();
for (T t : tResponses) {
stats.add(t);
}
for (T twitterResponse : this.sort(stats)) {
Collections.sort(tResponses, this.getComparator());
for (T twitterResponse : tResponses) {
forward(twitterResponse);
}
}
abstract protected List<T> sort(List<T> rl);
abstract Runnable getApiCallback();
protected Comparator getComparator() {
return new Comparator<Status>() {
public int compare(Status status, Status status1) {
return status.getCreatedAt().compareTo(status1.getCreatedAt());
}
};
}
@Override
public void start() {
protected void doStart(){
historyWritingPostProcessor.setTrackableComponent(this);
RateLimitStatusTrigger trigger = new RateLimitStatusTrigger(this.twitter);
Runnable apiCallback = this.getApiCallback();
twitterUpdatePollingTask = this.getTaskScheduler().schedule(apiCallback, trigger);
this.isRunning = true;
}
@Override
public void stop() {
protected void doStop(){
twitterUpdatePollingTask.cancel(true);
this.isRunning = false;
}
@Override
public boolean isRunning() {
return this.isRunning;
}
@Override
@@ -192,5 +188,4 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends Integrati
protected void markLastStatusId(long statusId) {
this.metadataStore.put(this.metadataKey, String.valueOf(statusId));
}
}

47
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterStatusEndpointSupport.java
View File

@@ -1,47 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.twitter.inbound;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import twitter4j.Status;
/**
* Simple base class for the reply and timeline cases (as well as any other {@link twitter4j.Status} implementations of
* {@link twitter4j.TwitterResponse}.
*
* @author Josh Long
* @author Oleg ZHurakousky
*/
abstract public class AbstractInboundTwitterStatusEndpointSupport extends AbstractInboundTwitterEndpointSupport<Status> {
private Comparator<Status> statusComparator = new Comparator<Status>() {
public int compare(Status status, Status status1) {
return status.getCreatedAt().compareTo(status1.getCreatedAt());
}
};
@Override
protected List<Status> sort(List<Status> rl) {
List<Status> statusArrayList = new ArrayList<Status>();
statusArrayList.addAll(rl);
Collections.sort(statusArrayList, statusComparator);
return statusArrayList;
}
}

29
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundDirectMessageEndpoint.java
View File

@@ -15,8 +15,6 @@
*/
package org.springframework.integration.twitter.inbound;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
@@ -34,24 +32,6 @@ import twitter4j.Paging;
*/
public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpointSupport<DirectMessage> {
private Comparator<DirectMessage> dmComparator = new Comparator<DirectMessage>() {
public int compare(DirectMessage directMessage, DirectMessage directMessage1) {
return directMessage.getCreatedAt().compareTo(directMessage1.getCreatedAt());
}
};
@Override
protected List<DirectMessage> sort(List<DirectMessage> rl) {
List<DirectMessage> dms = new ArrayList<DirectMessage>();
dms.addAll(rl);
Collections.sort(dms, dmComparator);
return dms;
}
@Override
public String getComponentType() {
return "twitter:inbound-dm-channel-adapter";
@@ -84,4 +64,13 @@ public class InboundDirectMessageEndpoint extends AbstractInboundTwitterEndpoint
};
return apiCallback;
}
@SuppressWarnings("rawtypes")
protected Comparator getComparator() {
return new Comparator<DirectMessage>() {
public int compare(DirectMessage directMessage, DirectMessage directMessage1) {
return directMessage.getCreatedAt().compareTo(directMessage1.getCreatedAt());
}
};
}
}

4
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundMentionEndpoint.java
View File

@@ -20,6 +20,7 @@ import java.util.List;
import org.springframework.integration.MessagingException;
import twitter4j.Paging;
import twitter4j.Status;
/**
* Handles forwarding all new {@link twitter4j.Status} that are 'replies' or 'mentions' to some other tweet.
@@ -27,7 +28,7 @@ import twitter4j.Paging;
* @author Josh Long
* @author Oleg Zhurakousky
*/
public class InboundMentionEndpoint extends AbstractInboundTwitterStatusEndpointSupport {
public class InboundMentionEndpoint extends AbstractInboundTwitterEndpointSupport<Status> {
@Override
public String getComponentType() {
@@ -57,5 +58,4 @@ public class InboundMentionEndpoint extends AbstractInboundTwitterStatusEndpoint
};
return apiCallback;
}
}

3
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/InboundTimelineUpdateEndpoint.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.twitter.inbound;
import org.springframework.integration.MessagingException;
import twitter4j.Paging;
import twitter4j.Status;
/**
@@ -28,7 +29,7 @@ import twitter4j.Paging;
* @author Oleg Zhurakousky
* @since 2.0
*/
public class InboundTimelineUpdateEndpoint extends AbstractInboundTwitterStatusEndpointSupport {
public class InboundTimelineUpdateEndpoint extends AbstractInboundTwitterEndpointSupport<Status> {
@Override
public String getComponentType() {

20
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingUsingNamespace-context.xml
View File

@@ -33,16 +33,20 @@
consumer-secret="${twitter.oauth.consumerSecret}"/>
<!-- <twitter:inbound-mention-channel-adapter twitter-connection="tc" channel="inbound_mentions"/>-->
<!-- <service-activator input-channel="inbound_mentions" ref="twitterAnnouncer" method="mention"/>-->
<twitter:inbound-dm-channel-adapter twitter-connection="tc" channel="inbound_dm">
<twitter:inbound-mention-channel-adapter twitter-connection="tc" channel="inbound_mentions">
<poller fixed-rate="5000" max-messages-per-poll="3"/>
</twitter:inbound-dm-channel-adapter>
<service-activator input-channel="inbound_dm" ref="twitterAnnouncer" method="dm"/>
</twitter:inbound-mention-channel-adapter>
<service-activator input-channel="inbound_mentions" ref="twitterAnnouncer" method="mention"/>
<!-- <twitter:inbound-update-channel-adapter id="twitterInbound" twitter-connection="tc" channel="inbound_updates"/>-->
<!-- <service-activator input-channel="inbound_updates" ref="twitterAnnouncer" method="updates"/>-->
<!-- <twitter:inbound-dm-channel-adapter twitter-connection="tc" channel="inbound_dm">-->
<!-- <poller fixed-rate="5000" max-messages-per-poll="3"/>-->
<!-- </twitter:inbound-dm-channel-adapter>-->
<!-- <service-activator input-channel="inbound_dm" ref="twitterAnnouncer" method="dm"/>-->
<twitter:inbound-update-channel-adapter id="twitterInbound" twitter-connection="tc" channel="inbound_updates">
<poller fixed-rate="5000" max-messages-per-poll="3"/>
</twitter:inbound-update-channel-adapter>
<service-activator input-channel="inbound_updates" ref="twitterAnnouncer" method="updates"/>
<beans:bean id="twitterAnnouncer" class="org.springframework.integration.twitter.config.TwitterAnnouncer"/>
</beans:beans>

3
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingUsingNamespace.java
View File

@@ -18,7 +18,6 @@ package org.springframework.integration.twitter.config;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -29,7 +28,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
public class TestReceivingUsingNamespace {
@Test
@Ignore
// @Ignore
/*
* In order to run this test you need to provide values to the twitter.properties file
*/