You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2015/10/20 16:36:47 UTC
[1/2] incubator-streams git commit: resolves STREAMS-371
Repository: incubator-streams
Updated Branches:
refs/heads/master 1470b9227 -> fba147e44
resolves STREAMS-371
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/20e7b661
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/20e7b661
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/20e7b661
Branch: refs/heads/master
Commit: 20e7b66171634b948daaa6d9cc119990fbeec50f
Parents: 1470b92
Author: Steve Blackmon <sb...@apache.org>
Authored: Wed Oct 14 15:36:44 2015 -0500
Committer: Steve Blackmon <sb...@apache.org>
Committed: Wed Oct 14 15:36:44 2015 -0500
----------------------------------------------------------------------
.../streams-provider-twitter/pom.xml | 15 +--
.../provider/TwitterFollowingProvider.java | 19 ++--
.../provider/TwitterFollowingProviderTask.java | 96 +++++++++++++++++---
.../provider/TwitterTimelineProvider.java | 7 --
.../TwitterUserInformationProvider.java | 54 +++++------
.../twitter/TwitterFollowingConfiguration.json | 18 ++++
6 files changed, 141 insertions(+), 68 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/20e7b661/streams-contrib/streams-provider-twitter/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/pom.xml b/streams-contrib/streams-provider-twitter/pom.xml
index c15ea27..5de3172 100644
--- a/streams-contrib/streams-provider-twitter/pom.xml
+++ b/streams-contrib/streams-provider-twitter/pom.xml
@@ -37,6 +37,10 @@
<artifactId>config</artifactId>
</dependency>
<dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-config</artifactId>
+ </dependency>
+ <dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
</dependency>
@@ -156,16 +160,7 @@
<addCompileSourceRoot>true</addCompileSourceRoot>
<generateBuilders>true</generateBuilders>
<sourcePaths>
- <sourcePath>src/main/jsonschema/com/twitter/TwitterConfiguration.json</sourcePath>
- <sourcePath>src/main/jsonschema/com/twitter/TwitterStreamConfiguration.json</sourcePath>
- <sourcePath>src/main/jsonschema/com/twitter/TwitterUserInformationConfiguration.json</sourcePath>
- <sourcePath>src/main/jsonschema/com/twitter/Delete.json</sourcePath>
- <sourcePath>src/main/jsonschema/com/twitter/UserstreamEvent.json</sourcePath>
- <sourcePath>src/main/jsonschema/com/twitter/Follow.json</sourcePath>
- <sourcePath>src/main/jsonschema/com/twitter/FriendList.json</sourcePath>
- <sourcePath>src/main/jsonschema/com/twitter/Retweet.json</sourcePath>
- <sourcePath>src/main/jsonschema/com/twitter/tweet.json</sourcePath>
- <sourcePath>src/main/jsonschema/com/twitter/User.json</sourcePath>
+ <sourcePath>src/main/jsonschema</sourcePath>
</sourcePaths>
<outputDirectory>target/generated-sources/jsonschema2pojo</outputDirectory>
<targetPackage>org.apache.streams.twitter.pojo</targetPackage>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/20e7b661/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProvider.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProvider.java
index 272d4b7..dad595a 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProvider.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProvider.java
@@ -20,11 +20,12 @@ package org.apache.streams.twitter.provider;
import com.google.common.base.Preconditions;
import com.google.common.collect.Queues;
+import org.apache.streams.config.ComponentConfigurator;
import org.apache.streams.config.StreamsConfigurator;
import org.apache.streams.core.DatumStatusCounter;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsResultSet;
-import org.apache.streams.twitter.TwitterUserInformationConfiguration;
+import org.apache.streams.twitter.TwitterFollowingConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import twitter4j.Twitter;
@@ -44,14 +45,18 @@ public class TwitterFollowingProvider extends TwitterUserInformationProvider {
protected final ReadWriteLock lock = new ReentrantReadWriteLock();
+ private TwitterFollowingConfiguration config;
+
+ public TwitterFollowingConfiguration getConfig() { return config; }
+
public static final int MAX_NUMBER_WAITING = 10000;
public TwitterFollowingProvider() {
- super(TwitterConfigurator.detectTwitterUserInformationConfiguration(StreamsConfigurator.config.getConfig("twitter")));
+ this.config = new ComponentConfigurator<>(TwitterFollowingConfiguration.class).detectConfiguration(StreamsConfigurator.getConfig().getConfig("twitter"));
}
- public TwitterFollowingProvider(TwitterUserInformationConfiguration config) {
- super(config);
+ public TwitterFollowingProvider(TwitterFollowingConfiguration config) {
+ this.config = config;
}
@Override
@@ -80,7 +85,7 @@ public class TwitterFollowingProvider extends TwitterUserInformationProvider {
Twitter client = getTwitterClient();
for (int i = 0; i < ids.length; i++) {
- TwitterFollowingProviderTask providerTask = new TwitterFollowingProviderTask(this, client, ids[i], getConfig().getEndpoint());
+ TwitterFollowingProviderTask providerTask = new TwitterFollowingProviderTask(this, client, ids[i], getConfig().getEndpoint(), getConfig().getIdsOnly());
executor.submit(providerTask);
}
}
@@ -89,7 +94,7 @@ public class TwitterFollowingProvider extends TwitterUserInformationProvider {
Twitter client = getTwitterClient();
for (int i = 0; i < screenNames.length; i++) {
- TwitterFollowingProviderTask providerTask = new TwitterFollowingProviderTask(this, client, screenNames[i], getConfig().getEndpoint());
+ TwitterFollowingProviderTask providerTask = new TwitterFollowingProviderTask(this, client, screenNames[i], getConfig().getEndpoint(), getConfig().getIdsOnly());
executor.submit(providerTask);
}
@@ -129,7 +134,7 @@ public class TwitterFollowingProvider extends TwitterUserInformationProvider {
@Override
public void prepare(Object o) {
- super.prepare(o);
+ super.prepare(config);
Preconditions.checkNotNull(getConfig().getEndpoint());
Preconditions.checkArgument(getConfig().getEndpoint().equals("friends") || getConfig().getEndpoint().equals("followers"));
return;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/20e7b661/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java
index ea737c5..0bbdb33 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java
@@ -48,23 +48,26 @@ public class TwitterFollowingProviderTask implements Runnable {
protected TwitterFollowingProvider provider;
protected Twitter client;
protected Long id;
+ protected Boolean ids_only;
protected String screenName;
protected String endpoint;
private int max_per_page = 200;
- public TwitterFollowingProviderTask(TwitterFollowingProvider provider, Twitter twitter, Long id, String endpoint) {
+ public TwitterFollowingProviderTask(TwitterFollowingProvider provider, Twitter twitter, Long id, String endpoint, Boolean ids_only) {
this.provider = provider;
this.client = twitter;
this.id = id;
this.endpoint = endpoint;
+ this.ids_only = ids_only;
}
- public TwitterFollowingProviderTask(TwitterFollowingProvider provider, Twitter twitter, String screenName, String endpoint) {
+ public TwitterFollowingProviderTask(TwitterFollowingProvider provider, Twitter twitter, String screenName, String endpoint, Boolean ids_only) {
this.provider = provider;
this.client = twitter;
this.screenName = screenName;
this.endpoint = endpoint;
+ this.ids_only = ids_only;
}
@@ -86,6 +89,13 @@ public class TwitterFollowingProviderTask implements Runnable {
Preconditions.checkArgument(endpoint.equals("friends") || endpoint.equals("followers"));
+ if( ids_only == true )
+ collectIds(id);
+ else
+ collectUsers(id);
+ }
+
+ private void collectUsers(Long id) {
int keepTrying = 0;
long curser = -1;
@@ -94,11 +104,11 @@ public class TwitterFollowingProviderTask implements Runnable {
{
try
{
- twitter4j.User followee4j;
- String followeeJson;
+ twitter4j.User user;
+ String userJson;
try {
- followee4j = client.users().showUser(id);
- followeeJson = TwitterObjectFactory.getRawJSON(followee4j);
+ user = client.users().showUser(id);
+ userJson = TwitterObjectFactory.getRawJSON(user);
} catch (TwitterException e) {
LOGGER.error("Failure looking up " + id);
break;
@@ -113,15 +123,23 @@ public class TwitterFollowingProviderTask implements Runnable {
Preconditions.checkNotNull(list);
Preconditions.checkArgument(list.size() > 0);
- for (twitter4j.User follower4j : list) {
+ for (twitter4j.User other : list) {
- String followerJson = TwitterObjectFactory.getRawJSON(follower4j);
+ String otherJson = TwitterObjectFactory.getRawJSON(other);
try {
- Follow follow = new Follow()
- .withFollowee(mapper.readValue(followeeJson, User.class))
- .withFollower(mapper.readValue(followerJson, User.class));
-
+ Follow follow;
+ if( endpoint.equals("followers") ) {
+ follow = new Follow()
+ .withFollowee(mapper.readValue(userJson, User.class))
+ .withFollower(mapper.readValue(otherJson, User.class));
+ } else if( endpoint.equals("friends") ) {
+ follow = new Follow()
+ .withFollowee(mapper.readValue(otherJson, User.class))
+ .withFollower(mapper.readValue(userJson, User.class));
+ } else {
+ throw new Exception("endpoint must be set to 'friends' or 'followers'");
+ }
ComponentUtils.offerUntilSuccess(new StreamsDatum(follow), provider.providerQueue);
} catch (JsonParseException e) {
LOGGER.warn(e.getMessage());
@@ -130,6 +148,9 @@ public class TwitterFollowingProviderTask implements Runnable {
} catch (IOException e) {
LOGGER.warn(e.getMessage());
}
+ catch (Exception e) {
+ LOGGER.warn(e.getMessage());
+ }
}
if( list.size() == max_per_page )
curser = list.getNextCursor();
@@ -144,6 +165,57 @@ public class TwitterFollowingProviderTask implements Runnable {
} while (curser != 0 && keepTrying < 10);
}
+ private void collectIds(Long id) {
+ int keepTrying = 0;
+
+ long curser = -1;
+
+ do
+ {
+ try
+ {
+ twitter4j.IDs ids = null;
+ if( endpoint.equals("followers") )
+ ids = client.friendsFollowers().getFollowersIDs(id.longValue(), curser, max_per_page);
+ else if( endpoint.equals("friends") )
+ ids = client.friendsFollowers().getFriendsIDs(id.longValue(), curser, max_per_page);
+
+ Preconditions.checkNotNull(ids);
+ Preconditions.checkArgument(ids.getIDs().length > 0);
+
+ for (long otherId : ids.getIDs()) {
+
+ try {
+ Follow follow;
+ if( endpoint.equals("followers") ) {
+ follow = new Follow()
+ .withFollowee(new User().withId(id))
+ .withFollower(new User().withId(otherId));
+ } else if( endpoint.equals("friends") ) {
+ follow = new Follow()
+ .withFollowee(new User().withId(otherId))
+ .withFollower(new User().withId(id));
+ } else {
+ throw new Exception("endpoint must be set to 'friends' or 'followers'");
+ }
+ ComponentUtils.offerUntilSuccess(new StreamsDatum(follow), provider.providerQueue);
+ } catch (Exception e) {
+ LOGGER.warn(e.getMessage());
+ }
+ }
+ if( ids.hasNext() )
+ curser = ids.getNextCursor();
+ else break;
+ }
+ catch(TwitterException twitterException) {
+ keepTrying += TwitterErrorHandler.handleTwitterError(client, twitterException);
+ }
+ catch(Exception e) {
+ keepTrying += TwitterErrorHandler.handleTwitterError(client, e);
+ }
+ } while (curser != 0 && keepTrying < 10);
+ }
+
protected void getFollowing(String screenName) {
twitter4j.User user = null;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/20e7b661/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java
index fc3f5f2..203df18 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java
@@ -57,7 +57,6 @@ public class TwitterTimelineProvider implements StreamsProvider, Serializable {
private TwitterUserInformationConfiguration config;
- private Class klass;
protected final ReadWriteLock lock = new ReentrantReadWriteLock();
public TwitterUserInformationConfiguration getConfig() {
@@ -94,11 +93,6 @@ public class TwitterTimelineProvider implements StreamsProvider, Serializable {
this.config = config;
}
- public TwitterTimelineProvider(TwitterUserInformationConfiguration config, Class klass) {
- this.config = config;
- this.klass = klass;
- }
-
public Queue<StreamsDatum> getProviderQueue() {
return this.providerQueue;
}
@@ -225,7 +219,6 @@ public class TwitterTimelineProvider implements StreamsProvider, Serializable {
}
Preconditions.checkNotNull(providerQueue);
- Preconditions.checkNotNull(this.klass);
Preconditions.checkNotNull(config.getOauth().getConsumerKey());
Preconditions.checkNotNull(config.getOauth().getConsumerSecret());
Preconditions.checkNotNull(config.getOauth().getAccessToken());
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/20e7b661/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterUserInformationProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterUserInformationProvider.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterUserInformationProvider.java
index 9def663..7da2694 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterUserInformationProvider.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterUserInformationProvider.java
@@ -21,17 +21,18 @@ package org.apache.streams.twitter.provider;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
-import com.typesafe.config.Config;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.streams.config.ComponentConfigurator;
import org.apache.streams.config.StreamsConfigurator;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsProvider;
import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.twitter.TwitterFollowingConfiguration;
import org.apache.streams.twitter.TwitterUserInformationConfiguration;
import org.apache.streams.util.ComponentUtils;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.commons.lang.NotImplementedException;
import twitter4j.Twitter;
import twitter4j.TwitterException;
import twitter4j.TwitterFactory;
@@ -54,14 +55,13 @@ public class TwitterUserInformationProvider implements StreamsProvider, Serializ
public static final String STREAMS_ID = "TwitterUserInformationProvider";
private static final Logger LOGGER = LoggerFactory.getLogger(TwitterUserInformationProvider.class);
- private TwitterUserInformationConfiguration twitterUserInformationConfiguration;
+ private TwitterUserInformationConfiguration config;
- private Class klass;
protected volatile Queue<StreamsDatum> providerQueue = new LinkedBlockingQueue<StreamsDatum>();
- public TwitterUserInformationConfiguration getConfig() { return twitterUserInformationConfiguration; }
+ public TwitterUserInformationConfiguration getConfig() { return config; }
- public void setConfig(TwitterUserInformationConfiguration config) { this.twitterUserInformationConfiguration = config; }
+ public void setConfig(TwitterUserInformationConfiguration config) { this.config = config; }
protected Iterator<Long[]> idsBatches;
protected Iterator<String[]> screenNameBatches;
@@ -80,24 +80,11 @@ public class TwitterUserInformationProvider implements StreamsProvider, Serializ
}
public TwitterUserInformationProvider() {
- Config config = StreamsConfigurator.config.getConfig("twitter");
- this.twitterUserInformationConfiguration = TwitterConfigurator.detectTwitterUserInformationConfiguration(config);
-
+ this.config = new ComponentConfigurator<>(TwitterFollowingConfiguration.class).detectConfiguration(StreamsConfigurator.getConfig().getConfig("twitter"));
}
public TwitterUserInformationProvider(TwitterUserInformationConfiguration config) {
- this.twitterUserInformationConfiguration = config;
- }
-
- public TwitterUserInformationProvider(Class klass) {
- Config config = StreamsConfigurator.config.getConfig("twitter");
- this.twitterUserInformationConfiguration = TwitterConfigurator.detectTwitterUserInformationConfiguration(config);
- this.klass = klass;
- }
-
- public TwitterUserInformationProvider(TwitterUserInformationConfiguration config, Class klass) {
- this.twitterUserInformationConfiguration = config;
- this.klass = klass;
+ this.config = config;
}
public Queue<StreamsDatum> getProviderQueue() {
@@ -229,12 +216,15 @@ public class TwitterUserInformationProvider implements StreamsProvider, Serializ
@Override
public void prepare(Object o) {
+ if( o instanceof TwitterFollowingConfiguration )
+ config = (TwitterUserInformationConfiguration) o;
+
Preconditions.checkNotNull(providerQueue);
- Preconditions.checkNotNull(twitterUserInformationConfiguration.getOauth().getConsumerKey());
- Preconditions.checkNotNull(twitterUserInformationConfiguration.getOauth().getConsumerSecret());
- Preconditions.checkNotNull(twitterUserInformationConfiguration.getOauth().getAccessToken());
- Preconditions.checkNotNull(twitterUserInformationConfiguration.getOauth().getAccessTokenSecret());
- Preconditions.checkNotNull(twitterUserInformationConfiguration.getInfo());
+ Preconditions.checkNotNull(config.getOauth().getConsumerKey());
+ Preconditions.checkNotNull(config.getOauth().getConsumerSecret());
+ Preconditions.checkNotNull(config.getOauth().getAccessToken());
+ Preconditions.checkNotNull(config.getOauth().getAccessTokenSecret());
+ Preconditions.checkNotNull(config.getInfo());
List<String> screenNames = new ArrayList<String>();
List<String[]> screenNameBatches = new ArrayList<String[]>();
@@ -242,7 +232,7 @@ public class TwitterUserInformationProvider implements StreamsProvider, Serializ
List<Long> ids = new ArrayList<Long>();
List<Long[]> idsBatches = new ArrayList<Long[]>();
- for(String s : twitterUserInformationConfiguration.getInfo()) {
+ for(String s : config.getInfo()) {
if(s != null)
{
String potentialScreenName = s.replaceAll("@", "").trim().toLowerCase();
@@ -288,13 +278,13 @@ public class TwitterUserInformationProvider implements StreamsProvider, Serializ
protected Twitter getTwitterClient()
{
- String baseUrl = TwitterProviderUtil.baseUrl(twitterUserInformationConfiguration);
+ String baseUrl = TwitterProviderUtil.baseUrl(config);
ConfigurationBuilder builder = new ConfigurationBuilder()
- .setOAuthConsumerKey(twitterUserInformationConfiguration.getOauth().getConsumerKey())
- .setOAuthConsumerSecret(twitterUserInformationConfiguration.getOauth().getConsumerSecret())
- .setOAuthAccessToken(twitterUserInformationConfiguration.getOauth().getAccessToken())
- .setOAuthAccessTokenSecret(twitterUserInformationConfiguration.getOauth().getAccessTokenSecret())
+ .setOAuthConsumerKey(config.getOauth().getConsumerKey())
+ .setOAuthConsumerSecret(config.getOauth().getConsumerSecret())
+ .setOAuthAccessToken(config.getOauth().getAccessToken())
+ .setOAuthAccessTokenSecret(config.getOauth().getAccessTokenSecret())
.setIncludeEntitiesEnabled(true)
.setJSONStoreEnabled(true)
.setAsyncNumThreads(3)
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/20e7b661/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/TwitterFollowingConfiguration.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/TwitterFollowingConfiguration.json b/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/TwitterFollowingConfiguration.json
new file mode 100644
index 0000000..c72f3cf
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/TwitterFollowingConfiguration.json
@@ -0,0 +1,18 @@
+{
+ "$schema": "http://json-schema.org/draft-03/schema",
+ "$license": [
+ "http://www.apache.org/licenses/LICENSE-2.0"
+ ],
+ "id": "#",
+ "type": "object",
+ "javaType" : "org.apache.streams.twitter.TwitterFollowingConfiguration",
+ "extends": {"$ref":"TwitterUserInformationConfiguration.json"},
+ "javaInterfaces": ["java.io.Serializable"],
+ "properties": {
+ "ids_only": {
+ "type": "boolean",
+ "description": "Whether to collect ids only, or full profiles",
+ "value": "true"
+ }
+ }
+}
\ No newline at end of file
[2/2] incubator-streams git commit: PR feedback
https://github.com/apache/incubator-streams/pull/263
Posted by sb...@apache.org.
PR feedback https://github.com/apache/incubator-streams/pull/263
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/fba147e4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/fba147e4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/fba147e4
Branch: refs/heads/master
Commit: fba147e44b14e0dc58cd3b5fa8da234b195be806
Parents: 20e7b66
Author: Steve Blackmon <sb...@apache.org>
Authored: Mon Oct 19 14:53:29 2015 -0500
Committer: Steve Blackmon <sb...@apache.org>
Committed: Mon Oct 19 14:53:29 2015 -0500
----------------------------------------------------------------------
.../provider/TwitterFollowingProviderTask.java | 31 ++--
.../provider/TwitterFriendsProviderTask.java | 181 -------------------
2 files changed, 17 insertions(+), 195 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fba147e4/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java
index 0bbdb33..1f40953 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFollowingProviderTask.java
@@ -37,7 +37,7 @@ import twitter4j.TwitterObjectFactory;
import java.io.IOException;
/**
- * Retrieve recent posts for a single user id.
+ * Retrieve friend or follower connections for a single user id.
*/
public class TwitterFollowingProviderTask implements Runnable {
@@ -48,26 +48,26 @@ public class TwitterFollowingProviderTask implements Runnable {
protected TwitterFollowingProvider provider;
protected Twitter client;
protected Long id;
- protected Boolean ids_only;
+ protected Boolean idsOnly;
protected String screenName;
protected String endpoint;
private int max_per_page = 200;
- public TwitterFollowingProviderTask(TwitterFollowingProvider provider, Twitter twitter, Long id, String endpoint, Boolean ids_only) {
+ public TwitterFollowingProviderTask(TwitterFollowingProvider provider, Twitter twitter, Long id, String endpoint, Boolean idsOnly) {
this.provider = provider;
this.client = twitter;
this.id = id;
this.endpoint = endpoint;
- this.ids_only = ids_only;
+ this.idsOnly = idsOnly;
}
- public TwitterFollowingProviderTask(TwitterFollowingProvider provider, Twitter twitter, String screenName, String endpoint, Boolean ids_only) {
+ public TwitterFollowingProviderTask(TwitterFollowingProvider provider, Twitter twitter, String screenName, String endpoint, Boolean idsOnly) {
this.provider = provider;
this.client = twitter;
this.screenName = screenName;
this.endpoint = endpoint;
- this.ids_only = ids_only;
+ this.idsOnly = idsOnly;
}
@@ -89,7 +89,7 @@ public class TwitterFollowingProviderTask implements Runnable {
Preconditions.checkArgument(endpoint.equals("friends") || endpoint.equals("followers"));
- if( ids_only == true )
+ if( idsOnly )
collectIds(id);
else
collectUsers(id);
@@ -128,7 +128,7 @@ public class TwitterFollowingProviderTask implements Runnable {
String otherJson = TwitterObjectFactory.getRawJSON(other);
try {
- Follow follow;
+ Follow follow = null;
if( endpoint.equals("followers") ) {
follow = new Follow()
.withFollowee(mapper.readValue(userJson, User.class))
@@ -137,10 +137,12 @@ public class TwitterFollowingProviderTask implements Runnable {
follow = new Follow()
.withFollowee(mapper.readValue(otherJson, User.class))
.withFollower(mapper.readValue(userJson, User.class));
- } else {
- throw new Exception("endpoint must be set to 'friends' or 'followers'");
}
ComponentUtils.offerUntilSuccess(new StreamsDatum(follow), provider.providerQueue);
+
+ Preconditions.checkNotNull(follow);
+
+
} catch (JsonParseException e) {
LOGGER.warn(e.getMessage());
} catch (JsonMappingException e) {
@@ -186,7 +188,7 @@ public class TwitterFollowingProviderTask implements Runnable {
for (long otherId : ids.getIDs()) {
try {
- Follow follow;
+ Follow follow = null;
if( endpoint.equals("followers") ) {
follow = new Follow()
.withFollowee(new User().withId(id))
@@ -195,12 +197,13 @@ public class TwitterFollowingProviderTask implements Runnable {
follow = new Follow()
.withFollowee(new User().withId(otherId))
.withFollower(new User().withId(id));
- } else {
- throw new Exception("endpoint must be set to 'friends' or 'followers'");
}
ComponentUtils.offerUntilSuccess(new StreamsDatum(follow), provider.providerQueue);
+
+ Preconditions.checkNotNull(follow);
+
} catch (Exception e) {
- LOGGER.warn(e.getMessage());
+ LOGGER.warn("Exception: {}", e);
}
}
if( ids.hasNext() )
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/fba147e4/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFriendsProviderTask.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFriendsProviderTask.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFriendsProviderTask.java
deleted file mode 100644
index c5ababe..0000000
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterFriendsProviderTask.java
+++ /dev/null
@@ -1,181 +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
- *
- * 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.streams.twitter.provider;
-
-import com.fasterxml.jackson.core.JsonParseException;
-import com.fasterxml.jackson.databind.JsonMappingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.Lists;
-import org.apache.streams.core.StreamsDatum;
-import org.apache.streams.jackson.StreamsJacksonMapper;
-import org.apache.streams.twitter.pojo.Follow;
-import org.apache.streams.util.ComponentUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import twitter4j.PagableResponseList;
-import twitter4j.Twitter;
-import twitter4j.TwitterException;
-import twitter4j.TwitterObjectFactory;
-import twitter4j.User;
-
-import java.io.IOException;
-
-/**
- * Retrieve recent posts for a single user id.
- */
-public class TwitterFriendsProviderTask implements Runnable {
-
- private final static Logger LOGGER = LoggerFactory.getLogger(TwitterFriendsProviderTask.class);
-
- private final static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
-
- protected TwitterFollowingProvider provider;
- protected Twitter client;
- protected Long id;
- protected String screenName;
-
- public TwitterFriendsProviderTask(TwitterFollowingProvider provider, Twitter twitter, Long id) {
- this.provider = provider;
- this.client = twitter;
- this.id = id;
- }
-
- public TwitterFriendsProviderTask(TwitterFollowingProvider provider, Twitter twitter, String screenName) {
- this.provider = provider;
- this.client = twitter;
- this.screenName = screenName;
- }
-
-
- @Override
- public void run() {
-
- if( id != null )
- getFriends(id);
- if( screenName != null)
- getFriends(screenName);
-
- LOGGER.info(id != null ? id.toString() : screenName + " Thread Finished");
-
- }
-
- protected void getFriends(Long id) {
-
- int keepTrying = 0;
-
- long curser = -1;
-
- do
- {
- try
- {
- twitter4j.User follower4j;
- String followerJson;
- try {
- follower4j = client.users().showUser(id);
- followerJson = TwitterObjectFactory.getRawJSON(follower4j);
- } catch (TwitterException e) {
- LOGGER.error("Failure looking up " + id);
- break;
- }
-
- PagableResponseList<User> followeeList = client.friendsFollowers().getFriendsList(id.longValue(), curser);
-
- for (twitter4j.User followee4j : followeeList ) {
-
- String followeeJson = TwitterObjectFactory.getRawJSON(followee4j);
-
- try {
- Follow follow = new Follow()
- .withFollowee(mapper.readValue(followeeJson, org.apache.streams.twitter.pojo.User.class))
- .withFollower(mapper.readValue(followerJson, org.apache.streams.twitter.pojo.User.class));
-
- ComponentUtils.offerUntilSuccess(new StreamsDatum(follow), provider.providerQueue);
- } catch (JsonParseException e) {
- LOGGER.warn(e.getMessage());
- } catch (JsonMappingException e) {
- LOGGER.warn(e.getMessage());
- } catch (IOException e) {
- LOGGER.warn(e.getMessage());
- }
- }
- curser = followeeList.getNextCursor();
- }
- catch(TwitterException twitterException) {
- keepTrying += TwitterErrorHandler.handleTwitterError(client, twitterException);
- }
- catch(Exception e) {
- keepTrying += TwitterErrorHandler.handleTwitterError(client, e);
- }
- } while (curser != 0 && keepTrying < 10);
- }
-
- protected void getFriends(String screenName) {
-
- int keepTrying = 0;
-
- long curser = -1;
-
- do
- {
- try
- {
- twitter4j.User follower4j;
- String followerJson;
- try {
- follower4j = client.users().showUser(screenName);
- followerJson = TwitterObjectFactory.getRawJSON(follower4j);
- } catch (TwitterException e) {
- LOGGER.error("Failure looking up " + screenName);
- break;
- }
-
- PagableResponseList<User> followeeList = client.friendsFollowers().getFriendsList(screenName, curser);
-
- for (twitter4j.User followee4j : followeeList ) {
-
- String followeeJson = TwitterObjectFactory.getRawJSON(followee4j);
-
- try {
- Follow follow = new Follow()
- .withFollowee(mapper.readValue(followeeJson, org.apache.streams.twitter.pojo.User.class))
- .withFollower(mapper.readValue(followerJson, org.apache.streams.twitter.pojo.User.class));
-
- ComponentUtils.offerUntilSuccess(new StreamsDatum(follow), provider.providerQueue);
- } catch (JsonParseException e) {
- LOGGER.warn(e.getMessage());
- } catch (JsonMappingException e) {
- LOGGER.warn(e.getMessage());
- } catch (IOException e) {
- LOGGER.warn(e.getMessage());
- }
- }
- curser = followeeList.getNextCursor();
- }
- catch(TwitterException twitterException) {
- keepTrying += TwitterErrorHandler.handleTwitterError(client, twitterException);
- }
- catch(Exception e) {
- keepTrying += TwitterErrorHandler.handleTwitterError(client, e);
- }
- } while (curser != 0 && keepTrying < 10);
- }
-
-
-}