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

updating tests.\

Browse Source
This commit is contained in:
Josh Long
2010-10-23 14:56:16 -07:00
parent 13966b4f75
commit d1c05b17ff
9 changed files with 129 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,9 +20,7 @@ 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;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.history.HistoryWritingMessagePostProcessor;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.support.MessageBuilder;
@@ -36,6 +34,9 @@ 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;
/**
@@ -53,16 +54,35 @@ import java.util.List;
* @author Josh Long
* @since 2.0
*/
public abstract class AbstractInboundTwitterEndpointSupport<T> extends AbstractEndpoint implements Lifecycle, TrackableComponent {
public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessageProducerSupport implements Lifecycle, TrackableComponent, Runnable {
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();
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;
@@ -72,6 +92,14 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends AbstractE
abstract protected List<T> sort(List<T> rl);
@Override
protected void onInit() {
super.onInit();
Assert.notNull(this.configuration, "'configuration' can't be null");
this.twitter = this.configuration.getTwitter();
Assert.notNull(this.twitter, "'twitter' instance can't be null");
}
protected void forwardAll(List<T> tResponses) {
List<T> stats = new ArrayList<T>();
@@ -86,35 +114,39 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends AbstractE
return markerId;
}
public String getComponentType() {
return "twitter:inbound-dm-channel-adapter";
}
public void setRequestChannel(MessageChannel requestChannel) {
this.messagingTemplate.setDefaultChannel(requestChannel);
this.requestChannel = requestChannel;
}
abstract public String getComponentType();
@Override
protected void doStart() {
try {
this.historyWritingPostProcessor.setTrackableComponent(this);
refresh();
} catch (Exception e) {
throw new RuntimeException(e);
}
historyWritingPostProcessor.setTrackableComponent(this);
checkTaskExecutor(getClass().getName() + "-taskExecutor");
taskExecutor.execute(this);
}
protected void forward(T status) {
synchronized (this.markerGuard) {
Message<T> twtMsg = MessageBuilder.withPayload(status).build();
messagingTemplate.convertAndSend(requestChannel, twtMsg,
this.historyWritingPostProcessor);
sendMessage(twtMsg);
markLastStatusId(status);
}
}
abstract protected void refresh() throws Exception;
/**
* 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<T> tResponses) {
List<T> stats = new ArrayList<T>();
@@ -127,7 +159,7 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends AbstractE
}
@SuppressWarnings("unchecked")
protected void runAsAPIRateLimitsPermit(ApiCallback cb)
protected void runAsAPIRateLimitsPermit(ApiCallback apiCallback)
throws Exception {
synchronized (this.apiPermitGuard) {
while (waitUntilPullAvailable()) {
@@ -135,7 +167,7 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends AbstractE
logger.debug("have room to make an API request now");
}
cb.run(this, twitter);
apiCallback.run(this, twitter);
}
}
}
@@ -171,8 +203,8 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends AbstractE
Thread.sleep(msUntilWeCanPullAgain);
} catch (Throwable throwable) {
logger.debug("encountered an error when" +
" trying to refresh the timeline: " +
logger.debug(
"encountered an error when trying to refresh the timeline: " +
ExceptionUtils.getFullStackTrace(throwable));
}
@@ -187,14 +219,6 @@ public abstract class AbstractInboundTwitterEndpointSupport<T> extends AbstractE
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() {
}

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

@@ -19,6 +19,7 @@ package org.springframework.integration.twitter;
//import twitter4j.Status;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.twitter.model.Status;
import org.springframework.integration.twitter.model.Twitter4jStatusImpl;
@@ -26,6 +27,9 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
/**
@@ -34,8 +38,9 @@ import java.util.List;
*
* @author Josh Long
*/
abstract public class AbstractInboundTwitterStatusEndpointSupport
extends AbstractInboundTwitterEndpointSupport<Status> {
abstract public class AbstractInboundTwitterStatusEndpointSupport extends AbstractInboundTwitterEndpointSupport<Status> {
private Comparator<Status> statusComparator = new Comparator<Status>() {
public int compare(Status status, Status status1) {

12
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/InboundDirectMessageStatusEndpoint.java
View File

@@ -27,6 +27,9 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
/**
@@ -60,7 +63,13 @@ public class InboundDirectMessageStatusEndpoint extends AbstractInboundTwitterEn
}
@Override
protected void refresh() throws Exception {
public String getComponentType() {
return null;
}
@Override
protected void beginPolling() throws Exception {
this.runAsAPIRateLimitsPermit(new ApiCallback<InboundDirectMessageStatusEndpoint>() {
public void run(InboundDirectMessageStatusEndpoint t, Twitter twitter)
throws Exception {
@@ -77,5 +86,4 @@ public class InboundDirectMessageStatusEndpoint extends AbstractInboundTwitterEn
});
}
}

22
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/InboundMentionStatusEndpoint.java
View File

@@ -26,22 +26,26 @@ import java.util.List;
*
* @author Josh Long
*/
public class InboundMentionStatusEndpoint
extends AbstractInboundTwitterStatusEndpointSupport {
public class InboundMentionStatusEndpoint extends AbstractInboundTwitterStatusEndpointSupport {
@Override
protected void refresh() throws Exception {
this.runAsAPIRateLimitsPermit(new ApiCallback<InboundMentionStatusEndpoint>() {
public void run(InboundMentionStatusEndpoint ctx,
Twitter twitter) throws Exception {
public String getComponentType() {
return null;
}
@Override
protected void beginPolling() throws Exception {
this.runAsAPIRateLimitsPermit(new ApiCallback<InboundMentionStatusEndpoint>() {
public void run(InboundMentionStatusEndpoint ctx, Twitter twitter) throws Exception {
List<twitter4j.Status> stats = (!hasMarkedStatus())
? twitter.getMentions()
: twitter.getMentions(new Paging(ctx.getMarkerId()));
forwardAll( fromTwitter4jStatuses( stats));
}
});
}
}

11
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/InboundUpdatedStatusEndpoint.java
View File

@@ -27,8 +27,14 @@ import twitter4j.Twitter;
* @since 2.0
*/
public class InboundUpdatedStatusEndpoint extends AbstractInboundTwitterStatusEndpointSupport {
@Override
protected void refresh() throws Exception {
public String getComponentType() {
return null;
}
@Override
protected void beginPolling() throws Exception {
this.runAsAPIRateLimitsPermit(new ApiCallback<InboundUpdatedStatusEndpoint>() {
public void run(InboundUpdatedStatusEndpoint t, Twitter twitter)
throws Exception {
@@ -36,6 +42,5 @@ public class InboundUpdatedStatusEndpoint extends AbstractInboundTwitterStatusEn
? twitter.getFriendsTimeline() :
twitter.getFriendsTimeline(new Paging(t.getMarkerId()))));
}
});
}
}); }
}

2
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/DirectMessageInboundEndpointParser.java
View File

@@ -21,7 +21,7 @@ public class DirectMessageInboundEndpointParser extends AbstractSingleBeanDefini
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "requestChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "twitter-connection", "configuration");
}
}

2
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/MentionInboundEndpointParser.java
View File

@@ -21,7 +21,7 @@ public class MentionInboundEndpointParser extends AbstractSingleBeanDefinitionPa
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "requestChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "twitter-connection", "configuration");
}
}

2
spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/UpdatedStatusInboundEndpointParser.java
View File

@@ -22,7 +22,7 @@ public class UpdatedStatusInboundEndpointParser extends AbstractSingleBeanDefini
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
"channel", "requestChannel");
"channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
"twitter-connection", "configuration");
}

55
spring-integration-twitter/src/test/java/org/springframework/integration/twitter/TestRecievingUsingNamespace-context.xml
View File

@@ -17,51 +17,50 @@
-->
<beans:beans
xmlns="http://www.springframework.org/schema/integration"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:tool="http://www.springframework.org/schema/tool"
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
xmlns="http://www.springframework.org/schema/integration"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:tool="http://www.springframework.org/schema/tool"
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-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"
ignore-unresolvable="true"/>
<message-history/>
<channel id="inbound_dm"/>
<context:component-scan
base-package="org.springframework.integration.twitter"/>
<context:property-placeholder
location="file://${user.home}/Desktop/twitter.properties"
ignore-unresolvable="true"/>
<channel id="inbound_dm"/>
<channel id="inbound_mentions"/>
<channel id="inbound_updates"/>
<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}"/>
<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}"/>
<twitter:inbound-update-channel-adapter twitter-connection="tc" channel="inbound_updates"/>
<service-activator input-channel="inbound_updates" ref="twitterAnnouncer" method="updates"/>
<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-dm-channel-adapter twitter-connection="tc" channel="inbound_dm"/>
<service-activator input-channel="inbound_dm" ref="twitterAnnouncer" method="dm"/>
<twitter:inbound-update-channel-adapter twitter-connection="tc" channel="inbound_updates"/>
<service-activator input-channel="inbound_updates" ref="twitterAnnouncer" method="updates"/>
</beans:beans>