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

formatting

Browse Source
This commit is contained in:
Mark Fisher
2010-10-26 15:13:30 -04:00
parent 6750130105
commit ce0e9f4e0a
1 changed files with 123 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.twitter.inbound;
import java.util.ArrayList;
@@ -20,13 +21,11 @@ import java.util.List;
import java.util.Properties;
import java.util.concurrent.ScheduledFuture;
import org.springframework.context.Lifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.context.metadata.FileBasedPropertiesStore;
import org.springframework.integration.context.metadata.MetadataStore;
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;
@@ -37,131 +36,142 @@ import twitter4j.Status;
import twitter4j.Twitter;
/**
* 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
*
* Abstract class that defines common operations for receiving various types of
* messages when using the Twitter API. This class also handles keeping track of
* the latest inbound message it has received and avoiding, where possible,
* redelivery of common messages. This functionality is enabled using the
* {@link org.springframework.integration.context.metadata.MetadataStore}
* strategy.
*
* @author Josh Long
* @author Oleg Zhurakousky
*
* @since 2.0
*/
public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessageProducerSupport implements Lifecycle, TrackableComponent{
private volatile MetadataStore metadataStore;
public abstract class AbstractInboundTwitterEndpointSupport<T> extends MessageProducerSupport {
private volatile MetadataStore metadataStore;
private volatile String metadataKey;
private volatile Properties lastPersistentEntry = new Properties();
protected volatile OAuthConfiguration configuration;
protected volatile long markerId = -1;
protected Twitter twitter;
private final Object markerGuard = new Object();
private volatile ScheduledFuture<?> twitterUpdatePollingTask;
private String persistentIdentifier;
protected volatile OAuthConfiguration configuration;
protected volatile long markerId = -1;
protected Twitter twitter;
private final Object markerGuard = new Object();
private volatile ScheduledFuture<?> twitterUpdatePollingTask;
private volatile String persistentIdentifier;
private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor();
abstract protected List<T> sort(List<T> rl);
abstract Runnable getApiCallback();
protected void markLastStatusId(long statusId){
lastPersistentEntry.put(metadataKey, String.valueOf(statusId));
}
public void setPersistentIdentifier(String persistentIdentifier) {
public void setPersistentIdentifier(String persistentIdentifier) {
this.persistentIdentifier = persistentIdentifier;
}
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() {
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");
metadataKey = this.getComponentType() + "@" + this.getComponentName() + "#" + this.configuration.getConsumerKey();
try {
if (StringUtils.hasText(this.persistentIdentifier)){
if (this.metadataStore == null){
logger.info("Creating FileBasedPropertiesStore");
metadataStore = new FileBasedPropertiesStore(this.persistentIdentifier);
((FileBasedPropertiesStore)metadataStore).afterPropertiesSet();
}
lastPersistentEntry = metadataStore.load();
}
} catch (Exception e) {
logger.warn("Failed to initailize initiaize and load from MetadataStore. Potential duplicates ppossible", e);
}
}
public void setConfiguration(OAuthConfiguration configuration) {
this.configuration = configuration;
}
protected void forwardAll(List<T> tResponses) {
List<T> stats = new ArrayList<T>();
public void setShouldTrack(boolean shouldTrack) {
this.historyWritingPostProcessor.setShouldTrack(shouldTrack);
}
for (T t : tResponses){
stats.add(t);
}
for (T twitterResponse : sort(stats)) {
forward(twitterResponse);
}
}
public long getMarkerId() {
return this.markerId;
}
@Override
protected void doStart() {
historyWritingPostProcessor.setTrackableComponent(this);
RateLimitStatusTrigger trigger = new RateLimitStatusTrigger(this.twitter);
Runnable apiCallback = this.getApiCallback();
twitterUpdatePollingTask = this.getTaskScheduler().schedule(apiCallback, trigger);
}
protected boolean hasMarkedStatus() {
return this.markerId > -1;
}
@Override
protected void doStop() {
twitterUpdatePollingTask.cancel(true);
}
protected void forward(T message) {
synchronized (this.markerGuard) {
Message<T> twtMsg = MessageBuilder.withPayload(message).build();
@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");
metadataKey = this.getComponentType() + "@" + this.getComponentName()
+ "#" + this.configuration.getConsumerKey();
try {
if (StringUtils.hasText(this.persistentIdentifier)) {
if (this.metadataStore == null) {
logger.info("Creating FileBasedPropertiesStore");
metadataStore = new FileBasedPropertiesStore(this.persistentIdentifier);
((FileBasedPropertiesStore) metadataStore).afterPropertiesSet();
}
lastPersistentEntry = metadataStore.load();
}
}
catch (Exception e) {
logger.warn("Failed to initialize and load from MetadataStore. Potential duplicates possible.", e);
}
}
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)) {
forward(twitterResponse);
}
}
abstract protected List<T> sort(List<T> rl);
abstract Runnable getApiCallback();
@Override
protected void doStart() {
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 message) {
synchronized (this.markerGuard) {
Message<T> twtMsg = MessageBuilder.withPayload(message).build();
long id = 0;
if (message instanceof DirectMessage) {
id = ((DirectMessage) message).getId();
}
else if (message instanceof Status) {
id = ((Status) message).getId();
}
else {
throw new IllegalArgumentException("Unsupported type of Twitter message: " + message.getClass());
}
String lastId = lastPersistentEntry.getProperty(this.metadataKey);
long lastTweetId = 0;
if (lastId != null) {
lastTweetId = Long.parseLong(lastId);
}
if (id > lastTweetId) {
sendMessage(twtMsg);
markLastStatusId(id);
if (metadataStore != null) {
metadataStore.write(this.lastPersistentEntry);
}
}
}
}
protected void markLastStatusId(long statusId) {
lastPersistentEntry.put(metadataKey, String.valueOf(statusId));
}
long id = 0;
if (message instanceof DirectMessage) {
id = ((DirectMessage)message).getId();
}
else if (message instanceof Status){
id = ((Status)message).getId();
}
else {
throw new IllegalArgumentException("Unsupported type of Twitter message: " + message.getClass());
}
String lastId = lastPersistentEntry.getProperty(this.metadataKey);
long lastTweetId = 0;
if (lastId != null){
lastTweetId = Long.parseLong(lastId);
}
if (id > lastTweetId){
sendMessage(twtMsg);
markLastStatusId(id);
if (metadataStore != null){
metadataStore.write(this.lastPersistentEntry);
}
}
}
}
protected boolean hasMarkedStatus() {
return markerId > -1;
}
}