You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2017/05/24 11:11:16 UTC
[5/7] camel git commit: CAMEL-11242 Split camel-twitter
component/endpoint into 4 (directmessage, search, streaming, timeline)
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointEvent.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointEvent.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointEvent.java
index c9edc40..54d707f 100644
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointEvent.java
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointEvent.java
@@ -19,26 +19,27 @@ package org.apache.camel.component.twitter;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
-import org.apache.camel.component.twitter.consumer.TwitterConsumer;
+import org.apache.camel.component.twitter.consumer.AbstractTwitterConsumerHandler;
import org.apache.camel.component.twitter.consumer.TwitterConsumerEvent;
import org.apache.camel.component.twitter.data.EndpointType;
import org.apache.camel.impl.DefaultEndpoint;
+@Deprecated
public class TwitterEndpointEvent extends DefaultEndpoint implements TwitterEndpoint {
- private final String remaining;
+ private final String kind;
// only TwitterEndpointPolling is annotated
private TwitterConfiguration properties;
public TwitterEndpointEvent(String uri, String remaining, TwitterComponent component, TwitterConfiguration properties) {
super(uri, component);
- this.remaining = remaining;
+ this.kind = remaining;
this.properties = properties;
}
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- TwitterConsumer twitter4jConsumer = TwitterHelper.createConsumer(this, getEndpointUri(), remaining);
+ AbstractTwitterConsumerHandler twitter4jConsumer = TwitterHelper.createConsumer(this, getEndpointUri(), kind);
return new TwitterConsumerEvent(this, processor, twitter4jConsumer);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointPolling.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointPolling.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointPolling.java
index 51b739e..008167a 100644
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointPolling.java
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterEndpointPolling.java
@@ -21,20 +21,26 @@ import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.api.management.ManagedAttribute;
import org.apache.camel.api.management.ManagedResource;
-import org.apache.camel.component.twitter.consumer.TwitterConsumer;
+import org.apache.camel.component.twitter.consumer.AbstractTwitterConsumerHandler;
import org.apache.camel.component.twitter.consumer.TwitterConsumerPolling;
import org.apache.camel.component.twitter.data.EndpointType;
import org.apache.camel.impl.DefaultPollingEndpoint;
+import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriPath;
/**
- * This component integrates with Twitter to send tweets or search for tweets and more.
+ * Use twitter-directmessage, twitter-search, twitter-streaming and twitter-timeline instead of this component.
+ * @deprecated
*/
+@Deprecated
@ManagedResource(description = "Managed Twitter Endpoint")
-@UriEndpoint(firstVersion = "2.10.0", scheme = "twitter", title = "Twitter", syntax = "twitter:kind", consumerClass = TwitterConsumer.class, label = "api,social")
+@UriEndpoint(firstVersion = "2.10.0", scheme = "twitter", title = "Twitter", syntax = "twitter:kind", consumerClass = AbstractTwitterConsumerHandler.class, label = "api,social")
public class TwitterEndpointPolling extends DefaultPollingEndpoint implements TwitterEndpoint {
- private final String remaining;
+ @UriPath(description = "The kind of endpoint", enums = "directmessage,search,streaming/filter,streaming/sample,streaming/user"
+ + ",timeline/home,timeline/mentions,timeline/retweetsofme,timeline/user") @Metadata(required = "true")
+ private final String kind;
@UriParam(optionalPrefix = "consumer.", defaultValue = "" + TwitterConsumerPolling.DEFAULT_CONSUMER_DELAY, label = "consumer,scheduler",
description = "Milliseconds before the next poll.")
@@ -45,7 +51,7 @@ public class TwitterEndpointPolling extends DefaultPollingEndpoint implements Tw
public TwitterEndpointPolling(String uri, String remaining, TwitterComponent component, TwitterConfiguration properties) {
super(uri, component);
- this.remaining = remaining;
+ this.kind = remaining;
this.properties = properties;
setDelay(delay); // reconfigure the default delay
@@ -53,7 +59,7 @@ public class TwitterEndpointPolling extends DefaultPollingEndpoint implements Tw
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- TwitterConsumer twitter4jConsumer = TwitterHelper.createConsumer(this, getEndpointUri(), remaining);
+ AbstractTwitterConsumerHandler twitter4jConsumer = TwitterHelper.createConsumer(this, getEndpointUri(), kind);
// update the pulling lastID with sinceId
twitter4jConsumer.setLastId(properties.getSinceId());
TwitterConsumerPolling tc = new TwitterConsumerPolling(this, processor, twitter4jConsumer);
@@ -63,7 +69,7 @@ public class TwitterEndpointPolling extends DefaultPollingEndpoint implements Tw
@Override
public Producer createProducer() throws Exception {
- return TwitterHelper.createProducer(this, getEndpointUri(), remaining);
+ return TwitterHelper.createProducer(this, getEndpointUri(), kind);
}
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterHelper.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterHelper.java
index 81fefa9..e51e08f 100644
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterHelper.java
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/TwitterHelper.java
@@ -16,25 +16,32 @@
*/
package org.apache.camel.component.twitter;
+import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
-import org.apache.camel.component.twitter.consumer.TwitterConsumer;
-import org.apache.camel.component.twitter.consumer.directmessage.DirectMessageConsumer;
-import org.apache.camel.component.twitter.consumer.search.SearchConsumer;
-import org.apache.camel.component.twitter.consumer.streaming.FilterStreamingConsumer;
-import org.apache.camel.component.twitter.consumer.streaming.SampleStreamingConsumer;
-import org.apache.camel.component.twitter.consumer.streaming.UserStreamingConsumer;
-import org.apache.camel.component.twitter.consumer.timeline.HomeConsumer;
-import org.apache.camel.component.twitter.consumer.timeline.MentionsConsumer;
-import org.apache.camel.component.twitter.consumer.timeline.RetweetsConsumer;
-import org.apache.camel.component.twitter.consumer.timeline.UserConsumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.twitter.consumer.AbstractTwitterConsumerHandler;
+import org.apache.camel.component.twitter.consumer.DefaultTwitterConsumer;
+import org.apache.camel.component.twitter.consumer.TwitterConsumerDirect;
+import org.apache.camel.component.twitter.consumer.TwitterConsumerEvent;
+import org.apache.camel.component.twitter.consumer.TwitterConsumerPolling;
import org.apache.camel.component.twitter.data.ConsumerType;
import org.apache.camel.component.twitter.data.StreamingType;
import org.apache.camel.component.twitter.data.TimelineType;
-import org.apache.camel.component.twitter.producer.DirectMessageProducer;
-import org.apache.camel.component.twitter.producer.SearchProducer;
-import org.apache.camel.component.twitter.producer.TwitterProducer;
-import org.apache.camel.component.twitter.producer.UserProducer;
+import org.apache.camel.component.twitter.directmessage.DirectMessageConsumerHandler;
+import org.apache.camel.component.twitter.directmessage.DirectMessageProducer;
+import org.apache.camel.component.twitter.search.SearchConsumerHandler;
+import org.apache.camel.component.twitter.search.SearchProducer;
+import org.apache.camel.component.twitter.streaming.FilterStreamingConsumerHandler;
+import org.apache.camel.component.twitter.streaming.SampleStreamingConsumerHandler;
+import org.apache.camel.component.twitter.streaming.UserStreamingConsumerHandler;
+import org.apache.camel.component.twitter.timeline.HomeConsumerHandler;
+import org.apache.camel.component.twitter.timeline.MentionsConsumerHandler;
+import org.apache.camel.component.twitter.timeline.RetweetsConsumerHandler;
+import org.apache.camel.component.twitter.timeline.UserConsumerHandler;
+import org.apache.camel.component.twitter.timeline.UserProducer;
+
import twitter4j.User;
public final class TwitterHelper {
@@ -58,30 +65,31 @@ public final class TwitterHelper {
message.setHeader(TwitterConstants.TWITTER_USER_ROLE + index, role);
}
- public static TwitterConsumer createConsumer(TwitterEndpoint te, String uri, String remaining) throws IllegalArgumentException {
+ @Deprecated
+ public static AbstractTwitterConsumerHandler createConsumer(TwitterEndpoint te, String uri, String remaining) throws IllegalArgumentException {
String[] tokens = remaining.split("/");
-
+
if (tokens.length > 0) {
switch (ConsumerType.fromString(tokens[0])) {
case DIRECTMESSAGE:
- return new DirectMessageConsumer(te);
+ return new DirectMessageConsumerHandler(te);
case SEARCH:
boolean hasNoKeywords = te.getProperties().getKeywords() == null
|| te.getProperties().getKeywords().trim().isEmpty();
if (hasNoKeywords) {
throw new IllegalArgumentException("Type set to SEARCH but no keywords were provided.");
} else {
- return new SearchConsumer(te);
+ return new SearchConsumerHandler(te);
}
case STREAMING:
if (tokens.length > 1) {
switch (StreamingType.fromString(tokens[1])) {
case SAMPLE:
- return new SampleStreamingConsumer(te);
+ return new SampleStreamingConsumerHandler(te);
case FILTER:
- return new FilterStreamingConsumer(te);
+ return new FilterStreamingConsumerHandler(te);
case USER:
- return new UserStreamingConsumer(te);
+ return new UserStreamingConsumerHandler(te);
default:
break;
}
@@ -91,16 +99,16 @@ public final class TwitterHelper {
if (tokens.length > 1) {
switch (TimelineType.fromString(tokens[1])) {
case HOME:
- return new HomeConsumer(te);
+ return new HomeConsumerHandler(te);
case MENTIONS:
- return new MentionsConsumer(te);
+ return new MentionsConsumerHandler(te);
case RETWEETSOFME:
- return new RetweetsConsumer(te);
+ return new RetweetsConsumerHandler(te);
case USER:
if (te.getProperties().getUser() == null || te.getProperties().getUser().trim().isEmpty()) {
throw new IllegalArgumentException("Fetch type set to USER TIMELINE but no user was set.");
} else {
- return new UserConsumer(te);
+ return new UserConsumerHandler(te);
}
default:
break;
@@ -116,7 +124,24 @@ public final class TwitterHelper {
+ ". A consumer type was not provided (or an incorrect pairing was used).");
}
- public static TwitterProducer createProducer(TwitterEndpoint te, String uri, String remaining) throws IllegalArgumentException {
+ public static Consumer createConsumer(Processor processor, AbstractTwitterEndpoint endpoint, AbstractTwitterConsumerHandler handler) throws Exception {
+ Consumer answer = new DefaultTwitterConsumer(endpoint, processor, handler);
+ switch (endpoint.getEndpointType()) {
+ case POLLING:
+ handler.setLastId(endpoint.getProperties().getSinceId());
+ endpoint.configureConsumer(answer);
+ break;
+ case DIRECT:
+ endpoint.configureConsumer(answer);
+ break;
+ default:
+ break;
+ }
+ return answer;
+ }
+
+ @Deprecated
+ public static Producer createProducer(TwitterEndpoint te, String uri, String remaining) throws IllegalArgumentException {
String[] tokens = remaining.split("/");
if (tokens.length > 0) {
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/AbstractTwitterConsumerHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/AbstractTwitterConsumerHandler.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/AbstractTwitterConsumerHandler.java
new file mode 100644
index 0000000..b0ec1f0
--- /dev/null
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/AbstractTwitterConsumerHandler.java
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer;
+
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.twitter.TwitterEndpoint;
+import twitter4j.Paging;
+import twitter4j.Twitter;
+import twitter4j.TwitterException;
+
+
+public abstract class AbstractTwitterConsumerHandler {
+
+ /**
+ * Instance of TwitterEndpoint.
+ */
+ protected final TwitterEndpoint endpoint;
+
+ /**
+ * The last tweet ID received.
+ */
+ private long lastId;
+
+ protected AbstractTwitterConsumerHandler(TwitterEndpoint endpoint) {
+ this.endpoint = endpoint;
+ this.lastId = -1;
+ }
+
+ /**
+ * Called by polling consumers during each poll. It needs to be separate
+ * from directConsume() since, as an example, streaming API polling allows
+ * tweets to build up between polls.
+ */
+ public abstract List<Exchange> pollConsume() throws TwitterException;
+
+ /**
+ * Called by direct consumers.
+ */
+ public abstract List<Exchange> directConsume() throws TwitterException;
+
+ /**
+ * Can't assume that the end of the list will be the most recent ID.
+ * The Twitter API sometimes returns them slightly out of order.
+ */
+ protected void setLastIdIfGreater(long newId) {
+ if (newId > lastId) {
+ lastId = newId;
+ }
+ }
+
+ /**
+ * Support to update the Consumer's lastId when starting the consumer
+ */
+ public void setLastId(long sinceId) {
+ lastId = sinceId;
+ }
+
+ protected Twitter getTwitter() {
+ return endpoint.getProperties().getTwitter();
+ }
+
+ protected long getLastId() {
+ return lastId;
+ }
+
+ protected Paging getLastIdPaging() {
+ return new Paging(lastId);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/DefaultTwitterConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/DefaultTwitterConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/DefaultTwitterConsumer.java
new file mode 100644
index 0000000..7b2b669
--- /dev/null
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/DefaultTwitterConsumer.java
@@ -0,0 +1,121 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer;
+
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.twitter.AbstractTwitterEndpoint;
+import org.apache.camel.component.twitter.streaming.AbstractStreamingConsumerHandler;
+import org.apache.camel.impl.ScheduledPollConsumer;
+
+/**
+ * Provides a scheduled polling consumer as well as event based consumer for streaming.
+ */
+public class DefaultTwitterConsumer extends ScheduledPollConsumer implements TwitterEventListener {
+
+ public static final long DEFAULT_CONSUMER_DELAY = 30 * 1000L;
+ private final AbstractTwitterEndpoint endpoint;
+ private final AbstractTwitterConsumerHandler handler;
+
+ public DefaultTwitterConsumer(AbstractTwitterEndpoint endpoint, Processor processor, AbstractTwitterConsumerHandler handler) {
+ super(endpoint, processor);
+ setDelay(DEFAULT_CONSUMER_DELAY);
+ this.endpoint = endpoint;
+ this.handler = handler;
+ }
+
+ @Override
+ public AbstractTwitterEndpoint getEndpoint() {
+ return (AbstractTwitterEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ switch (endpoint.getEndpointType()) {
+ case POLLING:
+ if (handler instanceof AbstractStreamingConsumerHandler) {
+ ((AbstractStreamingConsumerHandler) handler).start();
+ }
+ break;
+ case EVENT:
+ if (handler instanceof AbstractStreamingConsumerHandler) {
+ ((AbstractStreamingConsumerHandler) handler).setEventListener(this);
+ ((AbstractStreamingConsumerHandler) handler).start();
+ }
+ break;
+ default:
+ List<Exchange> exchanges = handler.directConsume();
+ for (int i = 0; i < exchanges.size(); i++) {
+ getProcessor().process(exchanges.get(i));
+ }
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ switch (endpoint.getEndpointType()) {
+ case POLLING:
+ if (handler instanceof AbstractStreamingConsumerHandler) {
+ ((AbstractStreamingConsumerHandler) handler).stop();
+ }
+ break;
+ case EVENT:
+ if (handler instanceof AbstractStreamingConsumerHandler) {
+ ((AbstractStreamingConsumerHandler) handler).removeEventListener(this);
+ ((AbstractStreamingConsumerHandler) handler).stop();
+ }
+ break;
+ default:
+ break;
+ }
+
+ super.doStop();
+ }
+
+ @Override
+ protected int poll() throws Exception {
+ List<Exchange> exchanges = handler.pollConsume();
+
+ int index = 0;
+ for (; index < exchanges.size(); index++) {
+ getProcessor().process(exchanges.get(index));
+ }
+
+ return index;
+ }
+
+ @Override
+ public void onEvent(Exchange exchange) {
+ if (!isRunAllowed()) {
+ return;
+ }
+
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing exchange on status update", exchange, exchange.getException());
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumer.java
deleted file mode 100644
index 3379105..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumer.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer;
-
-import java.util.List;
-
-import org.apache.camel.Exchange;
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import twitter4j.Paging;
-import twitter4j.Twitter;
-import twitter4j.TwitterException;
-
-
-public abstract class TwitterConsumer {
-
- /**
- * Instance of TwitterEndpoint.
- */
- protected final TwitterEndpoint endpoint;
-
- /**
- * The last tweet ID received.
- */
- private long lastId;
-
- protected TwitterConsumer(TwitterEndpoint endpoint) {
- this.endpoint = endpoint;
- this.lastId = -1;
- }
-
- /**
- * Called by polling consumers during each poll. It needs to be separate
- * from directConsume() since, as an example, streaming API polling allows
- * tweets to build up between polls.
- */
- public abstract List<Exchange> pollConsume() throws TwitterException;
-
- /**
- * Called by direct consumers.
- */
- public abstract List<Exchange> directConsume() throws TwitterException;
-
- /**
- * Can't assume that the end of the list will be the most recent ID.
- * The Twitter API sometimes returns them slightly out of order.
- */
- protected void setLastIdIfGreater(long newId) {
- if (newId > lastId) {
- lastId = newId;
- }
- }
-
- /**
- * Support to update the Consumer's lastId when starting the consumer
- */
- public void setLastId(long sinceId) {
- lastId = sinceId;
- }
-
- protected Twitter getTwitter() {
- return endpoint.getProperties().getTwitter();
- }
-
- protected long getLastId() {
- return lastId;
- }
-
- protected Paging getLastIdPaging() {
- return new Paging(lastId);
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerDirect.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerDirect.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerDirect.java
index 9c47259..3ae4ba6 100644
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerDirect.java
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerDirect.java
@@ -26,11 +26,12 @@ import org.apache.camel.impl.DefaultConsumer;
/**
* Camel DirectConsumer implementation.
*/
+@Deprecated
public class TwitterConsumerDirect extends DefaultConsumer {
- private final TwitterConsumer twitter4jConsumer;
+ private final AbstractTwitterConsumerHandler twitter4jConsumer;
- public TwitterConsumerDirect(TwitterEndpoint endpoint, Processor processor, TwitterConsumer twitter4jConsumer) {
+ public TwitterConsumerDirect(TwitterEndpoint endpoint, Processor processor, AbstractTwitterConsumerHandler twitter4jConsumer) {
super(endpoint, processor);
this.twitter4jConsumer = twitter4jConsumer;
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerEvent.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerEvent.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerEvent.java
index f5b94a0..6c3f87c 100644
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerEvent.java
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerEvent.java
@@ -19,13 +19,14 @@ package org.apache.camel.component.twitter.consumer;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.component.twitter.TwitterEndpoint;
-import org.apache.camel.component.twitter.consumer.streaming.AbstractStreamingConsumer;
+import org.apache.camel.component.twitter.streaming.AbstractStreamingConsumerHandler;
import org.apache.camel.impl.DefaultConsumer;
+@Deprecated
public class TwitterConsumerEvent extends DefaultConsumer implements TwitterEventListener {
- private final TwitterConsumer twitter4jConsumer;
+ private final AbstractTwitterConsumerHandler twitter4jConsumer;
- public TwitterConsumerEvent(TwitterEndpoint endpoint, Processor processor, TwitterConsumer twitter4jConsumer) {
+ public TwitterConsumerEvent(TwitterEndpoint endpoint, Processor processor, AbstractTwitterConsumerHandler twitter4jConsumer) {
super(endpoint, processor);
this.twitter4jConsumer = twitter4jConsumer;
}
@@ -34,17 +35,17 @@ public class TwitterConsumerEvent extends DefaultConsumer implements TwitterEven
protected void doStart() throws Exception {
super.doStart();
- if (twitter4jConsumer instanceof AbstractStreamingConsumer) {
- ((AbstractStreamingConsumer) twitter4jConsumer).setEventListener(this);
- ((AbstractStreamingConsumer) twitter4jConsumer).start();
+ if (twitter4jConsumer instanceof AbstractStreamingConsumerHandler) {
+ ((AbstractStreamingConsumerHandler) twitter4jConsumer).setEventListener(this);
+ ((AbstractStreamingConsumerHandler) twitter4jConsumer).start();
}
}
@Override
protected void doStop() throws Exception {
- if (twitter4jConsumer instanceof AbstractStreamingConsumer) {
- ((AbstractStreamingConsumer) twitter4jConsumer).removeEventListener(this);
- ((AbstractStreamingConsumer) twitter4jConsumer).stop();
+ if (twitter4jConsumer instanceof AbstractStreamingConsumerHandler) {
+ ((AbstractStreamingConsumerHandler) twitter4jConsumer).removeEventListener(this);
+ ((AbstractStreamingConsumerHandler) twitter4jConsumer).stop();
}
super.doStop();
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerPolling.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerPolling.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerPolling.java
index 368d6ef..0adbb3c 100644
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerPolling.java
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/TwitterConsumerPolling.java
@@ -20,20 +20,21 @@ import java.util.List;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.camel.component.twitter.TwitterEndpoint;
import org.apache.camel.component.twitter.TwitterEndpointPolling;
-import org.apache.camel.component.twitter.consumer.streaming.AbstractStreamingConsumer;
-import org.apache.camel.component.twitter.util.TwitterSorter;
+import org.apache.camel.component.twitter.streaming.AbstractStreamingConsumerHandler;
import org.apache.camel.impl.ScheduledPollConsumer;
/**
* Provides a scheduled polling consumer
*/
+@Deprecated
public class TwitterConsumerPolling extends ScheduledPollConsumer {
public static final long DEFAULT_CONSUMER_DELAY = 30 * 1000L;
- private final TwitterConsumer twitter4jConsumer;
+ private final AbstractTwitterConsumerHandler twitter4jConsumer;
- public TwitterConsumerPolling(TwitterEndpointPolling endpoint, Processor processor, TwitterConsumer twitter4jConsumer) {
+ public TwitterConsumerPolling(TwitterEndpoint endpoint, Processor processor, AbstractTwitterConsumerHandler twitter4jConsumer) {
super(endpoint, processor);
setDelay(DEFAULT_CONSUMER_DELAY);
this.twitter4jConsumer = twitter4jConsumer;
@@ -47,15 +48,15 @@ public class TwitterConsumerPolling extends ScheduledPollConsumer {
@Override
protected void doStart() throws Exception {
super.doStart();
- if (twitter4jConsumer instanceof AbstractStreamingConsumer) {
- ((AbstractStreamingConsumer) twitter4jConsumer).start();
+ if (twitter4jConsumer instanceof AbstractStreamingConsumerHandler) {
+ ((AbstractStreamingConsumerHandler) twitter4jConsumer).start();
}
}
@Override
protected void doStop() throws Exception {
- if (twitter4jConsumer instanceof AbstractStreamingConsumer) {
- ((AbstractStreamingConsumer) twitter4jConsumer).stop();
+ if (twitter4jConsumer instanceof AbstractStreamingConsumerHandler) {
+ ((AbstractStreamingConsumerHandler) twitter4jConsumer).stop();
}
super.doStop();
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/directmessage/DirectMessageConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/directmessage/DirectMessageConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/directmessage/DirectMessageConsumer.java
deleted file mode 100644
index d0b89de..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/directmessage/DirectMessageConsumer.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.directmessage;
-
-import java.util.List;
-
-import org.apache.camel.Exchange;
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import org.apache.camel.component.twitter.consumer.TwitterConsumer;
-import org.apache.camel.component.twitter.consumer.TwitterEventType;
-import twitter4j.DirectMessage;
-import twitter4j.TwitterException;
-
-/**
- * Consumes a user's direct messages
- */
-public class DirectMessageConsumer extends TwitterConsumer {
-
- public DirectMessageConsumer(TwitterEndpoint te) {
- super(te);
- }
-
- @Override
- public List<Exchange> pollConsume() throws TwitterException {
- List<DirectMessage> directMessages = getTwitter().getDirectMessages(getLastIdPaging());
- for (int i = 0; i < directMessages.size(); i++) {
- setLastIdIfGreater(directMessages.get(i).getId());
- }
-
- return TwitterEventType.DIRECT_MESSAGE.createExchangeList(endpoint, directMessages);
- }
-
- @Override
- public List<Exchange> directConsume() throws TwitterException {
- return TwitterEventType.DIRECT_MESSAGE.createExchangeList(
- endpoint,
- getTwitter().getDirectMessages()
- );
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/search/SearchConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/search/SearchConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/search/SearchConsumer.java
deleted file mode 100644
index 1c3a087..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/search/SearchConsumer.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.search;
-
-import java.util.Collections;
-import java.util.List;
-
-import org.apache.camel.Exchange;
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import org.apache.camel.component.twitter.consumer.TwitterConsumer;
-import org.apache.camel.component.twitter.consumer.TwitterEventType;
-import org.apache.camel.util.ObjectHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import twitter4j.GeoLocation;
-import twitter4j.Query;
-import twitter4j.Query.Unit;
-import twitter4j.QueryResult;
-import twitter4j.Status;
-import twitter4j.Twitter;
-import twitter4j.TwitterException;
-
-/**
- * Consumes search requests
- */
-public class SearchConsumer extends TwitterConsumer {
-
- private static final Logger LOG = LoggerFactory.getLogger(SearchConsumer.class);
-
- public SearchConsumer(TwitterEndpoint te) {
- super(te);
- }
-
- public List<Exchange> pollConsume() throws TwitterException {
- String keywords = endpoint.getProperties().getKeywords();
-
- Query query;
-
- if (keywords != null && keywords.trim().length() > 0) {
- query = new Query(keywords);
- LOG.debug("Searching twitter with keywords: {}", keywords);
- } else {
- query = new Query();
- LOG.debug("Searching twitter without keywords.");
- }
-
- if (endpoint.getProperties().isFilterOld()) {
- query.setSinceId(getLastId());
- }
-
- return search(query);
- }
-
- public List<Exchange> directConsume() throws TwitterException {
- String keywords = endpoint.getProperties().getKeywords();
- if (keywords == null || keywords.trim().length() == 0) {
- return Collections.emptyList();
- }
- Query query = new Query(keywords);
-
- LOG.debug("Searching twitter with keywords: {}", keywords);
- return search(query);
- }
-
- private List<Exchange> search(Query query) throws TwitterException {
- Integer numberOfPages = 1;
-
- if (ObjectHelper.isNotEmpty(endpoint.getProperties().getLang())) {
- query.setLang(endpoint.getProperties().getLang());
- }
-
- if (ObjectHelper.isNotEmpty(endpoint.getProperties().getCount())) {
- query.setCount(endpoint.getProperties().getCount());
- }
-
- if (ObjectHelper.isNotEmpty(endpoint.getProperties().getNumberOfPages())) {
- numberOfPages = endpoint.getProperties().getNumberOfPages();
- }
-
- if (ObjectHelper.isNotEmpty(endpoint.getProperties().getLatitude())
- && ObjectHelper.isNotEmpty(endpoint.getProperties().getLongitude())
- && ObjectHelper.isNotEmpty(endpoint.getProperties().getRadius())) {
- GeoLocation location = new GeoLocation(endpoint.getProperties().getLatitude(), endpoint.getProperties().getLongitude());
- query.setGeoCode(location, endpoint.getProperties().getRadius(), Unit.valueOf(endpoint.getProperties().getDistanceMetric()));
-
- LOG.debug("Searching with additional geolocation parameters.");
- }
-
- LOG.debug("Searching with {} pages.", numberOfPages);
-
- Twitter twitter = getTwitter();
- QueryResult qr = twitter.search(query);
- List<Status> tweets = qr.getTweets();
-
- for (int i = 1; i < numberOfPages; i++) {
- if (!qr.hasNext()) {
- break;
- }
-
- qr = twitter.search(qr.nextQuery());
- tweets.addAll(qr.getTweets());
- }
-
- if (endpoint.getProperties().isFilterOld()) {
- for (int i = 0; i < tweets.size(); i++) {
- setLastIdIfGreater(tweets.get(i).getId());
- }
- }
-
- return TwitterEventType.STATUS.createExchangeList(endpoint, tweets);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/AbstractStreamingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/AbstractStreamingConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/AbstractStreamingConsumer.java
deleted file mode 100644
index 7a7bfcb..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/AbstractStreamingConsumer.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.streaming;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.apache.camel.Exchange;
-import org.apache.camel.Service;
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import org.apache.camel.component.twitter.consumer.TwitterConsumer;
-import org.apache.camel.component.twitter.consumer.TwitterEventListener;
-import org.apache.camel.component.twitter.consumer.TwitterEventType;
-import twitter4j.Status;
-import twitter4j.StatusDeletionNotice;
-import twitter4j.StatusListener;
-import twitter4j.TwitterException;
-import twitter4j.TwitterStream;
-
-/**
- * Super class providing consuming capabilities for the streaming API.
- */
-public abstract class AbstractStreamingConsumer extends TwitterConsumer implements StatusListener, Service {
- private final TwitterStream twitterStream;
- private final List<Exchange> receivedStatuses;
- private final AtomicReference<TwitterEventListener> twitterEventListener;
- private boolean clear;
-
- public AbstractStreamingConsumer(TwitterEndpoint te) {
- super(te);
- this.receivedStatuses = new ArrayList<>();
- this.twitterStream = te.getProperties().createTwitterStream();
- this.twitterStream.addListener(this);
- this.twitterEventListener = new AtomicReference<>();
- this.clear = true;
- }
-
- @Override
- public List<Exchange> pollConsume() throws TwitterException {
- List<Exchange> result;
-
- synchronized (receivedStatuses) {
- clear = true;
- result = Collections.unmodifiableList(new ArrayList<>(receivedStatuses));
- }
-
- return result;
- }
-
- @Override
- public List<Exchange> directConsume() throws TwitterException {
- return Collections.emptyList();
- }
-
- @Override
- public void onException(Exception ex) {
- }
-
- @Override
- public void onStatus(Status status) {
- onEvent(TwitterEventType.STATUS.createExchange(endpoint, status));
- }
-
- @Override
- public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {
- // noop
- }
-
- @Override
- public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
- // noop
- }
-
- @Override
- public void onScrubGeo(long userId, long upToStatusId) {
- // noop
- }
-
- public void setEventListener(TwitterEventListener tweeterStatusListener) {
- twitterEventListener.set(tweeterStatusListener);
- }
-
- public void removeEventListener(TwitterEventListener tweeterStatusListener) {
- twitterEventListener.compareAndSet(tweeterStatusListener, null);
- }
-
- @Override
- public void stop() {
- twitterStream.removeListener(this);
- twitterStream.shutdown();
- twitterStream.cleanUp();
- }
-
- protected TwitterStream getTwitterStream() {
- return twitterStream;
- }
-
- protected void onEvent(Exchange exchange) {
- TwitterEventListener listener = twitterEventListener.get();
- if (listener != null) {
- listener.onEvent(exchange);
- } else {
- synchronized (receivedStatuses) {
- if (clear) {
- receivedStatuses.clear();
- clear = false;
- }
- receivedStatuses.add(exchange);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/FilterStreamingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/FilterStreamingConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/FilterStreamingConsumer.java
deleted file mode 100644
index 62460cc..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/FilterStreamingConsumer.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.streaming;
-
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import twitter4j.FilterQuery;
-import twitter4j.StallWarning;
-
-/**
- * Consumes the filter stream
- */
-public class FilterStreamingConsumer extends AbstractStreamingConsumer {
-
- public FilterStreamingConsumer(TwitterEndpoint endpoint) {
- super(endpoint);
- }
-
- @Override
- public void start() {
- getTwitterStream().filter(createFilter());
- }
-
- @Override
- public void onStallWarning(StallWarning stallWarning) {
- // noop
- }
-
- private FilterQuery createFilter() {
- FilterQuery filterQuery = new FilterQuery();
- String allLocationsString = endpoint.getProperties().getLocations();
- if (allLocationsString != null) {
- String[] locationStrings = allLocationsString.split(";");
- double[][] locations = new double[locationStrings.length][2];
- for (int i = 0; i < locationStrings.length; i++) {
- String[] coords = locationStrings[i].split(",");
- locations[i][0] = Double.valueOf(coords[0]);
- locations[i][1] = Double.valueOf(coords[1]);
- }
- filterQuery.locations(locations);
- }
-
- String keywords = endpoint.getProperties().getKeywords();
- if (keywords != null && keywords.length() > 0) {
- filterQuery.track(keywords.split(","));
- }
-
- String userIds = endpoint.getProperties().getUserIds();
- if (userIds != null) {
- String[] stringUserIds = userIds.split(",");
- long[] longUserIds = new long[stringUserIds.length];
- for (int i = 0; i < stringUserIds.length; i++) {
- longUserIds[i] = Long.valueOf(stringUserIds[i]);
- }
- filterQuery.follow(longUserIds);
- }
-
- if (allLocationsString == null && keywords == null && userIds == null) {
- throw new IllegalArgumentException("At least one filter parameter is required");
- }
-
- return filterQuery;
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/SampleStreamingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/SampleStreamingConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/SampleStreamingConsumer.java
deleted file mode 100644
index e0a5b27..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/SampleStreamingConsumer.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.streaming;
-
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import twitter4j.StallWarning;
-
-/**
- * Consumes the sample stream
- */
-public class SampleStreamingConsumer extends AbstractStreamingConsumer {
-
- public SampleStreamingConsumer(TwitterEndpoint endpoint) {
- super(endpoint);
- }
-
- @Override
- public void start() {
- getTwitterStream().sample();
- }
-
- @Override
- public void onStallWarning(StallWarning stallWarning) {
- // noop
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/UserStreamingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/UserStreamingConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/UserStreamingConsumer.java
deleted file mode 100644
index e04ba31..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/streaming/UserStreamingConsumer.java
+++ /dev/null
@@ -1,226 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.streaming;
-
-import org.apache.camel.Exchange;
-import org.apache.camel.component.twitter.TwitterConstants;
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import org.apache.camel.component.twitter.TwitterHelper;
-import org.apache.camel.component.twitter.consumer.TwitterEventType;
-import twitter4j.DirectMessage;
-import twitter4j.StallWarning;
-import twitter4j.Status;
-import twitter4j.User;
-import twitter4j.UserList;
-import twitter4j.UserStreamListener;
-
-public class UserStreamingConsumer extends AbstractStreamingConsumer implements UserStreamListener {
-
- public UserStreamingConsumer(TwitterEndpoint endpoint) {
- super(endpoint);
- }
-
- @Override
- public void start() {
- getTwitterStream().user();
- }
-
- @Override
- public void onDeletionNotice(long directMessageId, long userId) {
- // noop
- }
-
- @Override
- public void onFriendList(long[] friendIds) {
- // noop
- }
-
- @Override
- public void onFavorite(User source, User target, Status favoritedStatus) {
- Exchange exchange = TwitterEventType.FAVORITE.createExchange(endpoint, favoritedStatus);
- TwitterHelper.setUserHeader(exchange, 1, source, "source");
- TwitterHelper.setUserHeader(exchange, 2, target, "target");
-
- onEvent(exchange);
- }
-
- @Override
- public void onUnfavorite(User source, User target, Status unfavoritedStatus) {
- Exchange exchange = TwitterEventType.UNFAVORITE.createExchange(endpoint, unfavoritedStatus);
- TwitterHelper.setUserHeader(exchange, 1, source, "source");
- TwitterHelper.setUserHeader(exchange, 2, target, "target");
-
- onEvent(exchange);
- }
-
- @Override
- public void onFollow(User source, User followedUser) {
- Exchange exchange = TwitterEventType.FOLLOW.createExchange(endpoint);
- TwitterHelper.setUserHeader(exchange, 1, source, "source");
- TwitterHelper.setUserHeader(exchange, 2, followedUser, "followed");
-
- onEvent(exchange);
- }
-
- @Override
- public void onUnfollow(User source, User unfollowedUser) {
- Exchange exchange = TwitterEventType.UNFOLLOW.createExchange(endpoint);
- TwitterHelper.setUserHeader(exchange, 1, source, "source");
- TwitterHelper.setUserHeader(exchange, 2, unfollowedUser, "unfollowed");
-
- onEvent(exchange);
- }
-
- @Override
- public void onDirectMessage(DirectMessage directMessage) {
- onEvent(TwitterEventType.DIRECT_MESSAGE.createExchange(endpoint, directMessage));
- }
-
- @Override
- public void onUserListMemberAddition(User addedMember, User listOwner, UserList list) {
- Exchange exchange = TwitterEventType.USERLIST_MEMBER_ADDITION.createExchange(endpoint, list);
- TwitterHelper.setUserHeader(exchange, 1, addedMember, "addedMember");
- TwitterHelper.setUserHeader(exchange, 2, listOwner, "listOwner");
-
- onEvent(exchange);
- }
-
- @Override
- public void onUserListMemberDeletion(User deletedMember, User listOwner, UserList list) {
- Exchange exchange = TwitterEventType.USERLIST_MEMBER_DELETION.createExchange(endpoint, list);
- TwitterHelper.setUserHeader(exchange, 1, deletedMember, "deletedMember");
- TwitterHelper.setUserHeader(exchange, 2, listOwner, "listOwner");
-
- onEvent(exchange);
- }
-
- @Override
- public void onUserListSubscription(User subscriber, User listOwner, UserList list) {
- Exchange exchange = TwitterEventType.USERLIST_SUBSCRIPTION.createExchange(endpoint, list);
- TwitterHelper.setUserHeader(exchange, 1, subscriber, "subscriber");
- TwitterHelper.setUserHeader(exchange, 2, listOwner, "listOwner");
-
- onEvent(exchange);
- }
-
- @Override
- public void onUserListUnsubscription(User subscriber, User listOwner, UserList list) {
- Exchange exchange = TwitterEventType.USERLIST_UNSUBSCRIPTION.createExchange(endpoint, list);
- TwitterHelper.setUserHeader(exchange, 1, subscriber, "subscriber");
- TwitterHelper.setUserHeader(exchange, 2, listOwner, "listOwner");
-
- onEvent(exchange);
- }
-
- @Override
- public void onUserListCreation(User user, UserList userList) {
- Exchange exchange = TwitterEventType.USERLIST_CREATION.createExchange(endpoint, userList);
- TwitterHelper.setUserHeader(exchange, user);
-
- onEvent(exchange);
- }
-
- @Override
- public void onUserListUpdate(User user, UserList userList) {
- Exchange exchange = TwitterEventType.USERLIST_UPDATE.createExchange(endpoint, userList);
- TwitterHelper.setUserHeader(exchange, user);
-
- onEvent(exchange);
- }
-
- @Override
- public void onUserListDeletion(User user, UserList userList) {
- Exchange exchange = TwitterEventType.USERLIST_DELETETION.createExchange(endpoint, userList);
- TwitterHelper.setUserHeader(exchange, user);
-
- onEvent(exchange);
- }
-
- @Override
- public void onUserProfileUpdate(User user) {
- Exchange exchange = TwitterEventType.USER_PROFILE_UPDATE.createExchange(endpoint);
- TwitterHelper.setUserHeader(exchange, user);
-
- onEvent(exchange);
- }
-
- @Override
- public void onUserSuspension(long suspendedUser) {
- Exchange exchange = TwitterEventType.USER_SUSPENSION.createExchange(endpoint);
- exchange.getIn().setHeader(TwitterConstants.TWITTER_USER, suspendedUser);
-
- onEvent(exchange);
- }
-
- @Override
- public void onUserDeletion(long deletedUser) {
- Exchange exchange = TwitterEventType.USER_DELETION.createExchange(endpoint);
- exchange.getIn().setHeader(TwitterConstants.TWITTER_USER, deletedUser);
-
- onEvent(exchange);
- }
-
- @Override
- public void onBlock(User source, User blockedUser) {
- Exchange exchange = TwitterEventType.BLOCK.createExchange(endpoint);
- TwitterHelper.setUserHeader(exchange, 1, source, "source");
- TwitterHelper.setUserHeader(exchange, 2, blockedUser, "blocked");
-
- onEvent(exchange);
- }
-
- @Override
- public void onUnblock(User source, User unblockedUser) {
- Exchange exchange = TwitterEventType.UNBLOCK.createExchange(endpoint);
- TwitterHelper.setUserHeader(exchange, 1, source, "source");
- TwitterHelper.setUserHeader(exchange, 2, unblockedUser, "unblocked");
-
- onEvent(exchange);
- }
-
- @Override
- public void onStallWarning(StallWarning stallWarning) {
- // noop
- }
-
- @Override
- public void onRetweetedRetweet(User source, User target, Status retweetedStatus) {
- Exchange exchange = TwitterEventType.RETWEETED_RETWEET.createExchange(endpoint, retweetedStatus);
- TwitterHelper.setUserHeader(exchange, 1, source, "source");
- TwitterHelper.setUserHeader(exchange, 2, target, "target");
-
- onEvent(exchange);
- }
-
- @Override
- public void onFavoritedRetweet(User source, User target, Status favoritedRetweeet) {
- Exchange exchange = TwitterEventType.FAVORITED_RETWEET.createExchange(endpoint, favoritedRetweeet);
- TwitterHelper.setUserHeader(exchange, 1, source, "source");
- TwitterHelper.setUserHeader(exchange, 2, target, "target");
-
- onEvent(exchange);
- }
-
- @Override
- public void onQuotedTweet(User source, User target, Status quotingTweet) {
- Exchange exchange = TwitterEventType.QUOTED_TWEET.createExchange(endpoint, quotingTweet);
- TwitterHelper.setUserHeader(exchange, 1, source, "source");
- TwitterHelper.setUserHeader(exchange, 2, target, "target");
-
- onEvent(exchange);
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/AbstractStatusConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/AbstractStatusConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/AbstractStatusConsumer.java
deleted file mode 100644
index 18f0aee..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/AbstractStatusConsumer.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.timeline;
-
-import java.util.List;
-
-import org.apache.camel.Exchange;
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import org.apache.camel.component.twitter.consumer.TwitterConsumer;
-import org.apache.camel.component.twitter.consumer.TwitterEventType;
-import twitter4j.Status;
-import twitter4j.TwitterException;
-
-/**
- * Consumes the user's home timeline.
- */
-abstract class AbstractStatusConsumer extends TwitterConsumer {
-
- AbstractStatusConsumer(TwitterEndpoint endpoint) {
- super(endpoint);
- }
-
- @Override
- public List<Exchange> pollConsume() throws TwitterException {
- List<Status> statusList = doPoll();
- for (int i = 0; i < statusList.size(); i++) {
- setLastIdIfGreater(statusList.get(i).getId());
- }
-
- return TwitterEventType.STATUS.createExchangeList(endpoint, statusList);
- }
-
- @Override
- public List<Exchange> directConsume() throws TwitterException {
- return TwitterEventType.STATUS.createExchangeList(endpoint, doDirect());
- }
-
- protected abstract List<Status> doPoll() throws TwitterException;
-
- protected abstract List<Status> doDirect() throws TwitterException;
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/HomeConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/HomeConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/HomeConsumer.java
deleted file mode 100644
index 252bacc..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/HomeConsumer.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.timeline;
-
-import java.util.List;
-
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import twitter4j.Status;
-import twitter4j.TwitterException;
-
-/**
- * Consumes the user's home timeline.
- */
-public class HomeConsumer extends AbstractStatusConsumer {
-
- public HomeConsumer(TwitterEndpoint endpoint) {
- super(endpoint);
- }
-
- @Override
- protected List<Status> doPoll() throws TwitterException {
- return getTwitter().getHomeTimeline(getLastIdPaging());
- }
-
- @Override
- protected List<Status> doDirect() throws TwitterException {
- return getTwitter().getHomeTimeline();
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/MentionsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/MentionsConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/MentionsConsumer.java
deleted file mode 100644
index c3671c8..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/MentionsConsumer.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.timeline;
-
-import java.util.List;
-
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import twitter4j.Status;
-import twitter4j.TwitterException;
-
-/**
- * Consumes tweets in which the user has been mentioned.
- */
-public class MentionsConsumer extends AbstractStatusConsumer {
-
- public MentionsConsumer(TwitterEndpoint endpoint) {
- super(endpoint);
- }
-
- @Override
- protected List<Status> doPoll() throws TwitterException {
- return getTwitter().getMentionsTimeline(getLastIdPaging());
- }
-
- @Override
- protected List<Status> doDirect() throws TwitterException {
- return getTwitter().getMentionsTimeline();
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/RetweetsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/RetweetsConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/RetweetsConsumer.java
deleted file mode 100644
index dfd0815..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/RetweetsConsumer.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.timeline;
-
-import java.util.List;
-
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import twitter4j.Status;
-import twitter4j.TwitterException;
-
-/**
- * Consumes a user's tweets that have been retweeted
- */
-public class RetweetsConsumer extends AbstractStatusConsumer {
-
- public RetweetsConsumer(TwitterEndpoint endpoint) {
- super(endpoint);
- }
-
- @Override
- protected List<Status> doPoll() throws TwitterException {
- return getTwitter().getRetweetsOfMe(getLastIdPaging());
- }
-
- @Override
- protected List<Status> doDirect() throws TwitterException {
- return getTwitter().getRetweetsOfMe();
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/UserConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/UserConsumer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/UserConsumer.java
deleted file mode 100644
index 51ef02a..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/consumer/timeline/UserConsumer.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.consumer.timeline;
-
-import java.util.List;
-
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import twitter4j.Status;
-import twitter4j.TwitterException;
-
-/**
- * Consumes the timeline of a given user.
- */
-public class UserConsumer extends AbstractStatusConsumer {
-
- public UserConsumer(TwitterEndpoint endpoint) {
- super(endpoint);
- }
-
- @Override
- protected List<Status> doPoll() throws TwitterException {
- return getTwitter().getUserTimeline(endpoint.getProperties().getUser(), getLastIdPaging());
- }
-
- @Override
- protected List<Status> doDirect() throws TwitterException {
- return getTwitter().getUserTimeline(endpoint.getProperties().getUser());
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/data/TrendsType.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/data/TrendsType.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/data/TrendsType.java
index 2c4f49a..8e481d4 100644
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/data/TrendsType.java
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/data/TrendsType.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.twitter.data;
import org.apache.camel.component.twitter.TwitterHelper;
+@Deprecated
public enum TrendsType {
DAILY, WEEKLY, UNKNOWN;
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageConsumerHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageConsumerHandler.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageConsumerHandler.java
new file mode 100644
index 0000000..2fd125e
--- /dev/null
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageConsumerHandler.java
@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.twitter.directmessage;
+
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.twitter.TwitterEndpoint;
+import org.apache.camel.component.twitter.consumer.AbstractTwitterConsumerHandler;
+import org.apache.camel.component.twitter.consumer.TwitterEventType;
+import twitter4j.DirectMessage;
+import twitter4j.TwitterException;
+
+/**
+ * Consumes a user's direct messages
+ */
+public class DirectMessageConsumerHandler extends AbstractTwitterConsumerHandler {
+
+ public DirectMessageConsumerHandler(TwitterEndpoint te) {
+ super(te);
+ }
+
+ @Override
+ public List<Exchange> pollConsume() throws TwitterException {
+ List<DirectMessage> directMessages = getTwitter().getDirectMessages(getLastIdPaging());
+ for (int i = 0; i < directMessages.size(); i++) {
+ setLastIdIfGreater(directMessages.get(i).getId());
+ }
+
+ return TwitterEventType.DIRECT_MESSAGE.createExchangeList(endpoint, directMessages);
+ }
+
+ @Override
+ public List<Exchange> directConsume() throws TwitterException {
+ return TwitterEventType.DIRECT_MESSAGE.createExchangeList(
+ endpoint,
+ getTwitter().getDirectMessages()
+ );
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageProducer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageProducer.java
new file mode 100644
index 0000000..078c6cd
--- /dev/null
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageProducer.java
@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.twitter.directmessage;
+
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.twitter.TwitterConstants;
+import org.apache.camel.component.twitter.TwitterEndpoint;
+import org.apache.camel.impl.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Produces text as a direct message.
+ */
+public class DirectMessageProducer extends DefaultProducer {
+
+ private TwitterEndpoint endpoint;
+
+ public DirectMessageProducer(TwitterEndpoint endpoint) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ }
+
+ public void process(Exchange exchange) throws Exception {
+ // send direct message
+ String toUsername = endpoint.getProperties().getUser();
+ if (ObjectHelper.isNotEmpty(exchange.getIn().getHeader(TwitterConstants.TWITTER_USER, String.class))) {
+ toUsername = exchange.getIn().getHeader(TwitterConstants.TWITTER_USER, String.class);
+ }
+ String text = exchange.getIn().getBody(String.class);
+
+ if (toUsername.isEmpty()) {
+ throw new CamelExchangeException("Username not configured on TwitterEndpoint", exchange);
+ } else {
+ log.debug("Sending to: {} message: {}", toUsername, text);
+ endpoint.getProperties().getTwitter().sendDirectMessage(toUsername, text);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/TwitterDirectMessageComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/TwitterDirectMessageComponent.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/TwitterDirectMessageComponent.java
new file mode 100644
index 0000000..516bb31
--- /dev/null
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/TwitterDirectMessageComponent.java
@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.twitter.directmessage;
+
+import java.util.Map;
+
+import org.apache.camel.ComponentVerifier;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.twitter.AbstractTwitterComponent;
+import org.apache.camel.component.twitter.DefaultTwitterComponentVerifier;
+import org.apache.camel.component.twitter.TwitterConfiguration;
+import org.apache.camel.spi.Metadata;
+
+/**
+ * Twitter direct message component.
+ */
+@Metadata(label = "verifiers", enums = "parameters,connectivity")
+public class TwitterDirectMessageComponent extends AbstractTwitterComponent {
+
+ protected Endpoint doCreateEndpoint(TwitterConfiguration properties, String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ return new TwitterDirectMessageEndpoint(uri, remaining, this, properties);
+ }
+
+ /**
+ * Get a verifier for the twitter directmessage component.
+ */
+ public ComponentVerifier getVerifier() {
+ return new DefaultTwitterComponentVerifier(this, "twitter-directmessage");
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/TwitterDirectMessageEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/TwitterDirectMessageEndpoint.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/TwitterDirectMessageEndpoint.java
new file mode 100644
index 0000000..86ede93
--- /dev/null
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/TwitterDirectMessageEndpoint.java
@@ -0,0 +1,44 @@
+package org.apache.camel.component.twitter.directmessage;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.twitter.AbstractTwitterEndpoint;
+import org.apache.camel.component.twitter.TwitterConfiguration;
+import org.apache.camel.component.twitter.TwitterHelper;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriPath;
+
+/**
+ * The Twitter Direct Message Component consumes/produces user's direct messages.
+ */
+@UriEndpoint(firstVersion = "2.10.0", scheme = "twitter-directmessage", title = "Twitter Direct Message", syntax = "twitter-directmessage:endpointId", consumerClass = DirectMessageConsumerHandler.class, label = "api,social")
+public class TwitterDirectMessageEndpoint extends AbstractTwitterEndpoint {
+
+ @UriPath(description = "The endpoint ID (not used).")
+ @Metadata(required = "true")
+ private String endpointId;
+
+ public TwitterDirectMessageEndpoint(String uri, String remaining, TwitterDirectMessageComponent component, TwitterConfiguration properties) {
+ super(uri, component, properties);
+ this.endpointId = remaining;
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ if (getProperties().getUser() == null || getProperties().getUser().trim().isEmpty()) {
+ throw new IllegalArgumentException(
+ "Producer type set to DIRECT MESSAGE but no recipient user was set.");
+ } else {
+ return new DirectMessageProducer(this);
+ }
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ Consumer answer = TwitterHelper.createConsumer(processor, this, new DirectMessageConsumerHandler(this));
+ return answer;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bb64dca0/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/producer/DirectMessageProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/producer/DirectMessageProducer.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/producer/DirectMessageProducer.java
deleted file mode 100644
index 3f22526..0000000
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/producer/DirectMessageProducer.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.camel.component.twitter.producer;
-
-import org.apache.camel.CamelExchangeException;
-import org.apache.camel.Exchange;
-import org.apache.camel.component.twitter.TwitterConstants;
-import org.apache.camel.component.twitter.TwitterEndpoint;
-import org.apache.camel.util.ObjectHelper;
-
-/**
- * Produces text as a direct message.
- */
-public class DirectMessageProducer extends TwitterProducer {
-
- public DirectMessageProducer(TwitterEndpoint endpoint) {
- super(endpoint);
- }
-
- public void process(Exchange exchange) throws Exception {
- // send direct message
- String toUsername = endpoint.getProperties().getUser();
- if (ObjectHelper.isNotEmpty(exchange.getIn().getHeader(TwitterConstants.TWITTER_USER, String.class))) {
- toUsername = exchange.getIn().getHeader(TwitterConstants.TWITTER_USER, String.class);
- }
- String text = exchange.getIn().getBody(String.class);
-
- if (toUsername.isEmpty()) {
- throw new CamelExchangeException("Username not configured on TwitterEndpoint", exchange);
- } else {
- log.debug("Sending to: {} message: {}", toUsername, text);
- endpoint.getProperties().getTwitter().sendDirectMessage(toUsername, text);
- }
- }
-
-}