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 2016/10/19 20:32:36 UTC
[1/3] incubator-streams git commit: level up youtube provider
Repository: incubator-streams
Updated Branches:
refs/heads/master 006234fb8 -> 104f29b1e
level up youtube provider
update google client version (STREAMS-413)
use provider without a runtime (STREAMS-403)
add main methods to each Provider (STREAMS-411)
add real integration tests (STREAMS-415)
isolate isRunning and readCurrent (STREAMS-425)
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/762ce8c7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/762ce8c7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/762ce8c7
Branch: refs/heads/master
Commit: 762ce8c7c58aff54d296afbfbc863e586406f915
Parents: 3234cdb
Author: Steve Blackmon @steveblackmon <sb...@apache.org>
Authored: Thu Oct 13 13:21:59 2016 -0500
Committer: Steve Blackmon @steveblackmon <sb...@apache.org>
Committed: Thu Oct 13 13:21:59 2016 -0500
----------------------------------------------------------------------
.../streams-provider-youtube/pom.xml | 21 +++--
.../provider/YoutubeChannelDataCollector.java | 5 +-
.../provider/YoutubeChannelProvider.java | 84 +++++++++++++++++++
.../com/youtube/provider/YoutubeProvider.java | 86 ++++++++++++++------
.../provider/YoutubeUserActivityCollector.java | 5 +-
.../provider/YoutubeUserActivityProvider.java | 81 ++++++++++++++++++
.../src/site/markdown/index.md | 23 ++++++
.../providers/YoutubeChannelProviderIT.java | 55 +++++++++++++
.../YoutubeUserActivityProviderIT.java | 55 +++++++++++++
.../resources/YoutubeChannelProviderIT.conf | 22 +++++
.../YoutubeUserActivityProviderIT.conf | 22 +++++
11 files changed, 423 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/pom.xml b/streams-contrib/streams-provider-youtube/pom.xml
index 8d028af..037d4b6 100644
--- a/streams-contrib/streams-provider-youtube/pom.xml
+++ b/streams-contrib/streams-provider-youtube/pom.xml
@@ -30,10 +30,8 @@
<artifactId>streams-provider-youtube</artifactId>
<properties>
- <project.youtube.version>v3-rev107-1.18.0-rc</project.youtube.version>
- <project.youtube.analytics.version>v1-rev24-1.17.0-rc</project.youtube.analytics.version>
- <project.http.version>1.18.0-rc</project.http.version>
- <project.oauth.version>1.18.0-rc</project.oauth.version>
+ <youtube.client.version>v3-rev178-1.22.0</youtube.client.version>
+ <google.client.version>1.22.0</google.client.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
@@ -53,7 +51,7 @@
<dependency>
<groupId>com.google.http-client</groupId>
<artifactId>google-http-client-jackson2</artifactId>
- <version>${project.http.version}</version>
+ <version>${google.client.version}</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
@@ -69,7 +67,7 @@
<dependency>
<groupId>com.google.oauth-client</groupId>
<artifactId>google-oauth-client-jetty</artifactId>
- <version>${project.oauth.version}</version>
+ <version>${google.client.version}</version>
</dependency>
<dependency>
@@ -101,7 +99,7 @@
<dependency>
<groupId>com.google.api-client</groupId>
<artifactId>google-api-client</artifactId>
- <version>1.17.0-rc</version>
+ <version>${google.client.version}</version>
</dependency>
<dependency>
<groupId>org.apache.streams</groupId>
@@ -111,7 +109,7 @@
<dependency>
<groupId>com.google.apis</groupId>
<artifactId>google-api-services-youtube</artifactId>
- <version>v3-rev126-1.19.1</version>
+ <version>${youtube.client.version}</version>
</dependency>
<dependency>
<groupId>org.apache.streams</groupId>
@@ -166,6 +164,13 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <configuration>
+ <skipTests>${skipITs}</skipTests>
+ </configuration>
+ </plugin>
</plugins>
</build>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelDataCollector.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelDataCollector.java b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelDataCollector.java
index e0f9d7b..8e980a7 100644
--- a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelDataCollector.java
+++ b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelDataCollector.java
@@ -25,6 +25,7 @@ import com.google.api.client.http.HttpRequest;
import com.google.api.services.youtube.YouTube;
import com.google.api.services.youtube.model.Channel;
import com.google.api.services.youtube.model.ChannelListResponse;
+import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.google.gplus.configuration.UserInfo;
@@ -63,6 +64,7 @@ public class YoutubeChannelDataCollector extends YoutubeDataCollector{
@Override
public void run() {
+ Gson gson = new Gson();
try {
int attempt = 0;
YouTube.Channels.List channelLists = this.youTube.channels().list(CONTENT).setId(this.userInfo.getUserId()).setKey(this.youtubeConfig.getApiKey());
@@ -71,7 +73,8 @@ public class YoutubeChannelDataCollector extends YoutubeDataCollector{
try {
List<Channel> channels = channelLists.execute().getItems();
for (Channel channel : channels) {
- this.queue.put(new StreamsDatum(MAPPER.writeValueAsString(channel), channel.getId()));
+ String json = gson.toJson(channel);
+ this.queue.put(new StreamsDatum(json, channel.getId()));
}
if (StringUtils.isEmpty(channelLists.getPageToken())) {
channelLists = null;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelProvider.java b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelProvider.java
index 6a5fda1..817c98e 100644
--- a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelProvider.java
+++ b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeChannelProvider.java
@@ -19,20 +19,104 @@
package com.youtube.provider;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.api.services.youtube.YouTube;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Uninterruptibles;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigParseOptions;
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfiguration;
+import org.apache.streams.config.StreamsConfigurator;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.google.gplus.configuration.UserInfo;
+import org.apache.streams.jackson.StreamsJacksonMapper;
import org.apache.streams.util.api.requests.backoff.BackOffStrategy;
+import org.apache.youtube.pojo.YoutubeConfiguration;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.PrintStream;
+import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
/**
+ * Retrieve recent activity from a list of channels.
*
+ * To use from command line:
+ *
+ * Supply (at least) the following required configuration in application.conf:
+ *
+ * youtube.oauth.pathToP12KeyFile
+ * youtube.oauth.serviceAccountEmailAddress
+ * youtube.apiKey
+ * youtube.youtubeUsers
+ *
+ * Launch using:
+ *
+ * mvn exec:java -Dexec.mainClass=org.apache.streams.youtube.provider.YoutubeUserActivityProvider -Dexec.args="application.conf tweets.json"
*/
public class YoutubeChannelProvider extends YoutubeProvider {
+ public YoutubeChannelProvider() {
+ super();
+ }
+
+ public YoutubeChannelProvider(YoutubeConfiguration config) {
+ super(config);
+ }
+
@Override
protected Runnable getDataCollector(BackOffStrategy strategy, BlockingQueue<StreamsDatum> queue, YouTube youtube, UserInfo userInfo) {
return new YoutubeChannelDataCollector(youtube, queue, strategy, userInfo, this.config);
}
+
+ public static void main(String[] args) throws Exception {
+
+ Preconditions.checkArgument(args.length >= 2);
+
+ String configfile = args[0];
+ String outfile = args[1];
+
+ Config reference = ConfigFactory.load();
+ File conf_file = new File(configfile);
+ assert(conf_file.exists());
+ Config testResourceConfig = ConfigFactory.parseFileAnySyntax(conf_file, ConfigParseOptions.defaults().setAllowMissing(false));
+
+ Config typesafe = testResourceConfig.withFallback(reference).resolve();
+
+ StreamsConfiguration streamsConfiguration = StreamsConfigurator.detectConfiguration(typesafe);
+ YoutubeConfiguration config = new ComponentConfigurator<>(YoutubeConfiguration.class).detectConfiguration(typesafe, "youtube");
+ YoutubeChannelProvider provider = new YoutubeChannelProvider(config);
+
+ ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+ PrintStream outStream = new PrintStream(new BufferedOutputStream(new FileOutputStream(outfile)));
+ provider.prepare(config);
+ provider.startStream();
+ do {
+ Uninterruptibles.sleepUninterruptibly(streamsConfiguration.getBatchFrequencyMs(), TimeUnit.MILLISECONDS);
+ Iterator<StreamsDatum> iterator = provider.readCurrent().iterator();
+ while(iterator.hasNext()) {
+ StreamsDatum datum = iterator.next();
+ String json;
+ try {
+ if( datum.getDocument() instanceof String )
+ json = (String)datum.getDocument();
+ else
+ json = mapper.writeValueAsString(datum.getDocument());
+ outStream.println(json);
+ } catch (JsonProcessingException e) {
+ System.err.println(e.getMessage());
+ }
+ }
+ } while( provider.isRunning());
+ provider.cleanUp();
+ outStream.flush();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeProvider.java b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeProvider.java
index b04a781..e16f4bb 100644
--- a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeProvider.java
+++ b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeProvider.java
@@ -19,14 +19,21 @@
package com.youtube.provider;
+import com.google.api.client.auth.oauth2.Credential;
+import com.google.api.client.googleapis.auth.oauth2.GoogleCredential;
import com.google.api.client.http.HttpTransport;
import com.google.api.client.http.javanet.NetHttpTransport;
import com.google.api.client.json.JsonFactory;
import com.google.api.client.json.jackson2.JacksonFactory;
+import com.google.api.client.repackaged.com.google.common.base.Strings;
import com.google.api.services.youtube.YouTube;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import com.typesafe.config.Config;
import org.apache.streams.config.StreamsConfigurator;
import org.apache.streams.core.StreamsDatum;
@@ -41,9 +48,11 @@ import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
import java.io.IOException;
import java.math.BigInteger;
import java.security.GeneralSecurityException;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -53,6 +62,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
+import static java.util.concurrent.Executors.newSingleThreadExecutor;
+
public abstract class YoutubeProvider implements StreamsProvider {
public static final String STREAMS_ID = "YoutubeProvider";
@@ -60,6 +71,10 @@ public abstract class YoutubeProvider implements StreamsProvider {
private final static Logger LOGGER = LoggerFactory.getLogger(YoutubeProvider.class);
private final static int MAX_BATCH_SIZE = 1000;
+ // This OAuth 2.0 access scope allows for full read/write access to the
+ // authenticated user's account.
+ List<String> scopes = Lists.newArrayList("https://www.googleapis.com/auth/youtube");
+
/**
* Define a global instance of the HTTP transport.
*/
@@ -72,7 +87,9 @@ public abstract class YoutubeProvider implements StreamsProvider {
private static final int DEFAULT_THREAD_POOL_SIZE = 5;
- private ExecutorService executor;
+ List<ListenableFuture<Object>> futures = new ArrayList<>();
+
+ private ListeningExecutorService executor;
private BlockingQueue<StreamsDatum> datumQueue;
private AtomicBoolean isComplete;
private boolean previousPullWasEmpty;
@@ -99,6 +116,21 @@ public abstract class YoutubeProvider implements StreamsProvider {
}
@Override
+ public void prepare(Object configurationObject) {
+ try {
+ this.youtube = createYouTubeClient();
+ } catch (IOException |GeneralSecurityException e) {
+ LOGGER.error("Failed to created oauth for YouTube : {}", e);
+ throw new RuntimeException(e);
+ }
+
+ this.executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE));
+ this.datumQueue = new LinkedBlockingQueue<>(1000);
+ this.isComplete = new AtomicBoolean(false);
+ this.previousPullWasEmpty = false;
+ }
+
+ @Override
public void startStream() {
BackOffStrategy backOffStrategy = new ExponentialBackOffStrategy(2);
@@ -109,7 +141,9 @@ public abstract class YoutubeProvider implements StreamsProvider {
if(this.config.getDefaultBeforeDate() != null && user.getBeforeDate() == null) {
user.setBeforeDate(this.config.getDefaultBeforeDate());
}
- this.executor.submit(getDataCollector(backOffStrategy, this.datumQueue, this.youtube, user));
+
+ ListenableFuture future = executor.submit(getDataCollector(backOffStrategy, this.datumQueue, this.youtube, user));
+ futures.add(future);
}
this.executor.shutdown();
@@ -128,9 +162,6 @@ public abstract class YoutubeProvider implements StreamsProvider {
ComponentUtils.offerUntilSuccess(datum, batch);
}
}
- boolean pullIsEmpty = batch.isEmpty() && this.datumQueue.isEmpty() &&this.executor.isTerminated();
- this.isComplete.set(this.previousPullWasEmpty && pullIsEmpty);
- this.previousPullWasEmpty = pullIsEmpty;
return new StreamsResultSet(batch);
}
@@ -144,29 +175,22 @@ public abstract class YoutubeProvider implements StreamsProvider {
return null;
}
- @Override
- public boolean isRunning() {
- return !this.isComplete.get();
- }
-
- @Override
- public void prepare(Object configurationObject) {
- try {
- this.youtube = createYouTubeClient();
- } catch (IOException |GeneralSecurityException e) {
- LOGGER.error("Failed to created oauth for GPlus : {}", e);
- throw new RuntimeException(e);
- }
-
- this.executor = Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE);
- this.datumQueue = new LinkedBlockingQueue<>(1000);
- this.isComplete = new AtomicBoolean(false);
- this.previousPullWasEmpty = false;
- }
-
@VisibleForTesting
protected YouTube createYouTubeClient() throws IOException, GeneralSecurityException {
- return new YouTube.Builder(HTTP_TRANSPORT, JSON_FACTORY, null).setApplicationName("Streams Application").build();
+ GoogleCredential.Builder credentialBuilder = new GoogleCredential.Builder()
+ .setTransport(HTTP_TRANSPORT)
+ .setJsonFactory(JSON_FACTORY)
+ .setServiceAccountId(getConfig().getOauth().getServiceAccountEmailAddress())
+ .setServiceAccountScopes(scopes);
+
+ if( !Strings.isNullOrEmpty(getConfig().getOauth().getPathToP12KeyFile())) {
+ File p12KeyFile = new File(getConfig().getOauth().getPathToP12KeyFile());
+ if( p12KeyFile.exists() && p12KeyFile.isFile() && p12KeyFile.canRead()) {
+ credentialBuilder = credentialBuilder.setServiceAccountPrivateKeyFromP12File(p12KeyFile);
+ }
+ }
+ Credential credential = credentialBuilder.build();
+ return new YouTube.Builder(HTTP_TRANSPORT, JSON_FACTORY, credential).setApplicationName("Streams Application").build();
}
@Override
@@ -233,4 +257,14 @@ public abstract class YoutubeProvider implements StreamsProvider {
this.config.setYoutubeUsers(youtubeUsers);
}
+
+ @Override
+ public boolean isRunning() {
+ if (datumQueue.isEmpty() && executor.isTerminated() && Futures.allAsList(futures).isDone()) {
+ LOGGER.info("Completed");
+ isComplete.set(true);
+ LOGGER.info("Exiting");
+ }
+ return !isComplete.get();
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityCollector.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityCollector.java b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityCollector.java
index 5e7e9fa..76a69f3 100644
--- a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityCollector.java
+++ b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityCollector.java
@@ -28,6 +28,7 @@ import com.google.api.services.youtube.YouTube;
import com.google.api.services.youtube.model.ActivityListResponse;
import com.google.api.services.youtube.model.Video;
import com.google.api.services.youtube.model.VideoListResponse;
+import com.google.gson.Gson;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.google.gplus.configuration.UserInfo;
import org.apache.streams.jackson.StreamsJacksonMapper;
@@ -63,6 +64,8 @@ public class YoutubeUserActivityCollector extends YoutubeDataCollector {
private UserInfo userInfo;
private YoutubeConfiguration config;
+ Gson gson = new Gson();
+
public YoutubeUserActivityCollector(YouTube youtube, BlockingQueue<StreamsDatum> datumQueue, BackOffStrategy backOff, UserInfo userInfo, YoutubeConfiguration config) {
this.youtube = youtube;
this.datumQueue = datumQueue;
@@ -169,7 +172,7 @@ public class YoutubeUserActivityCollector extends YoutubeDataCollector {
|| (afterDate == null && beforeDate.isAfter(published))
|| ((afterDate != null && beforeDate != null) && (afterDate.isAfter(published) && beforeDate.isBefore(published)))) {
LOGGER.debug("Providing Youtube Activity: {}", MAPPER.writeValueAsString(video));
- this.datumQueue.put(new StreamsDatum(MAPPER.writeValueAsString(video), activity.getId()));
+ this.datumQueue.put(new StreamsDatum(gson.toJson(video), activity.getId()));
} else if (afterDate != null && afterDate.isAfter(published)) {
feed.setNextPageToken(null); // do not fetch next page
break;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityProvider.java b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityProvider.java
index 99820c1..ed3dc63 100644
--- a/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityProvider.java
+++ b/streams-contrib/streams-provider-youtube/src/main/java/com/youtube/provider/YoutubeUserActivityProvider.java
@@ -19,16 +19,53 @@
package com.youtube.provider;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.api.services.youtube.YouTube;
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.Uninterruptibles;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigParseOptions;
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfiguration;
+import org.apache.streams.config.StreamsConfigurator;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.google.gplus.configuration.UserInfo;
+import org.apache.streams.jackson.StreamsJacksonMapper;
import org.apache.streams.util.api.requests.backoff.BackOffStrategy;
import org.apache.youtube.pojo.YoutubeConfiguration;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.PrintStream;
+import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+/**
+ * Retrieve recent activity from a list of user ids or names.
+ *
+ * To use from command line:
+ *
+ * Supply (at least) the following required configuration in application.conf:
+ *
+ * youtube.oauth.pathToP12KeyFile
+ * youtube.oauth.serviceAccountEmailAddress
+ * youtube.apiKey
+ * youtube.youtubeUsers
+ *
+ * Launch using:
+ *
+ * mvn exec:java -Dexec.mainClass=org.apache.streams.youtube.provider.YoutubeUserActivityProvider -Dexec.args="application.conf tweets.json"
+ */
public class YoutubeUserActivityProvider extends YoutubeProvider {
+ public YoutubeUserActivityProvider() {
+ super();
+ }
+
public YoutubeUserActivityProvider(YoutubeConfiguration config) {
super(config);
}
@@ -37,4 +74,48 @@ public class YoutubeUserActivityProvider extends YoutubeProvider {
protected Runnable getDataCollector(BackOffStrategy strategy, BlockingQueue<StreamsDatum> queue, YouTube youtube, UserInfo userInfo) {
return new YoutubeUserActivityCollector(youtube, queue, strategy, userInfo, config);
}
+
+ public static void main(String[] args) throws Exception {
+
+ Preconditions.checkArgument(args.length >= 2);
+
+ String configfile = args[0];
+ String outfile = args[1];
+
+ Config reference = ConfigFactory.load();
+ File conf_file = new File(configfile);
+ assert(conf_file.exists());
+ Config testResourceConfig = ConfigFactory.parseFileAnySyntax(conf_file, ConfigParseOptions.defaults().setAllowMissing(false));
+
+ Config typesafe = testResourceConfig.withFallback(reference).resolve();
+
+ StreamsConfiguration streamsConfiguration = StreamsConfigurator.detectConfiguration(typesafe);
+ YoutubeConfiguration config = new ComponentConfigurator<>(YoutubeConfiguration.class).detectConfiguration(typesafe, "youtube");
+ YoutubeUserActivityProvider provider = new YoutubeUserActivityProvider(config);
+
+ ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+ PrintStream outStream = new PrintStream(new BufferedOutputStream(new FileOutputStream(outfile)));
+ provider.prepare(config);
+ provider.startStream();
+ do {
+ Uninterruptibles.sleepUninterruptibly(streamsConfiguration.getBatchFrequencyMs(), TimeUnit.MILLISECONDS);
+ Iterator<StreamsDatum> iterator = provider.readCurrent().iterator();
+ while(iterator.hasNext()) {
+ StreamsDatum datum = iterator.next();
+ String json;
+ try {
+ if( datum.getDocument() instanceof String )
+ json = (String)datum.getDocument();
+ else
+ json = mapper.writeValueAsString(datum.getDocument());
+ outStream.println(json);
+ } catch (JsonProcessingException e) {
+ System.err.println(e.getMessage());
+ }
+ }
+ } while( provider.isRunning());
+ provider.cleanUp();
+ outStream.flush();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/site/markdown/index.md
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/site/markdown/index.md b/streams-contrib/streams-provider-youtube/src/site/markdown/index.md
index cea90fe..de050e9 100644
--- a/streams-contrib/streams-provider-youtube/src/site/markdown/index.md
+++ b/streams-contrib/streams-provider-youtube/src/site/markdown/index.md
@@ -22,6 +22,29 @@ This module relies on classes from com.google.apis:google-api-services-youtube
| YoutubeChannelProvider [YoutubeChannelProvider.html](apidocs/com/youtube/provider/YoutubeChannelProvider.html "javadoc") | [YoutubeConfiguration.json](com/youtube/YoutubeConfiguration.json "YoutubeConfiguration.json") [YoutubeConfiguration.html](apidocs/com/youtube/YoutubeConfiguration.html "javadoc")
| YoutubeUserActivityProvider [YoutubeUserActivityProvider.html](apidocs/com/youtube/provider/YoutubeUserActivityProvider.html "javadoc") | [YoutubeConfiguration.json](com/youtube/YoutubeConfiguration.json "YoutubeConfiguration.json") [YoutubeConfiguration.html](apidocs/com/youtube/YoutubeConfiguration.html "javadoc")
+Test:
+-----
+
+Log into admin console
+Create project
+Enable Data API on project
+Create service account
+Download p12 file
+
+Create a local file `youtube.conf` with valid youtube credentials
+
+ youtube {
+ apiKey = ""
+ oauth {
+ serviceAccountEmailAddress = ""
+ pathToP12KeyFile = ""
+ }
+ }
+
+Build with integration testing enabled, using your credentials
+
+ mvn clean test verify -DskipITs=false -DargLine="-Dconfig.file=`pwd`/youtube.conf"
+
[JavaDocs](apidocs/index.html "JavaDocs")
###### Licensed under Apache License 2.0 - http://www.apache.org/licenses/LICENSE-2.0
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeChannelProviderIT.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeChannelProviderIT.java b/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeChannelProviderIT.java
new file mode 100644
index 0000000..cc1e811
--- /dev/null
+++ b/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeChannelProviderIT.java
@@ -0,0 +1,55 @@
+/*
+ * 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.youtube.test.providers;
+
+import com.google.common.collect.Lists;
+import com.youtube.provider.YoutubeChannelProvider;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.LineNumberReader;
+
+/**
+ * Created by sblackmon on 10/13/16.
+ */
+public class YoutubeChannelProviderIT {
+
+ @Test
+ public void testYoutubeChannelProvider() throws Exception {
+
+ String configfile = "./target/test-classes/YoutubeChannelProviderIT.conf";
+ String outfile = "./target/test-classes/YoutubeChannelProviderIT.stdout.txt";
+
+ YoutubeChannelProvider.main(Lists.newArrayList(configfile, outfile).toArray(new String[2]));
+
+ File out = new File(outfile);
+ assert (out.exists());
+ assert (out.canRead());
+ assert (out.isFile());
+
+ FileReader outReader = new FileReader(out);
+ LineNumberReader outCounter = new LineNumberReader(outReader);
+
+ while(outCounter.readLine() != null) {}
+
+ assert (outCounter.getLineNumber() > 1);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeUserActivityProviderIT.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeUserActivityProviderIT.java b/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeUserActivityProviderIT.java
new file mode 100644
index 0000000..e71a6d1
--- /dev/null
+++ b/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeUserActivityProviderIT.java
@@ -0,0 +1,55 @@
+/*
+ * 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.youtube.test.providers;
+
+import com.google.common.collect.Lists;
+import com.youtube.provider.YoutubeUserActivityProvider;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.LineNumberReader;
+
+/**
+ * Created by sblackmon on 10/13/16.
+ */
+public class YoutubeUserActivityProviderIT {
+
+ @Test
+ public void testYoutubeUserActivityProvider() throws Exception {
+
+ String configfile = "./target/test-classes/YoutubeUserActivityProviderIT.conf";
+ String outfile = "./target/test-classes/YoutubeUserActivityProviderIT.stdout.txt";
+
+ YoutubeUserActivityProvider.main(Lists.newArrayList(configfile, outfile).toArray(new String[2]));
+
+ File out = new File(outfile);
+ assert (out.exists());
+ assert (out.canRead());
+ assert (out.isFile());
+
+ FileReader outReader = new FileReader(out);
+ LineNumberReader outCounter = new LineNumberReader(outReader);
+
+ while(outCounter.readLine() != null) {}
+
+ assert (outCounter.getLineNumber() >= 250);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/test/resources/YoutubeChannelProviderIT.conf
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/test/resources/YoutubeChannelProviderIT.conf b/streams-contrib/streams-provider-youtube/src/test/resources/YoutubeChannelProviderIT.conf
new file mode 100644
index 0000000..8565c45
--- /dev/null
+++ b/streams-contrib/streams-provider-youtube/src/test/resources/YoutubeChannelProviderIT.conf
@@ -0,0 +1,22 @@
+# 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.
+youtube {
+ youtubeUsers = [
+ {
+ userId = "UCLDJ_V9KUOdOFSbDvPfGBxw"
+ }
+ ]
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/762ce8c7/streams-contrib/streams-provider-youtube/src/test/resources/YoutubeUserActivityProviderIT.conf
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/test/resources/YoutubeUserActivityProviderIT.conf b/streams-contrib/streams-provider-youtube/src/test/resources/YoutubeUserActivityProviderIT.conf
new file mode 100644
index 0000000..8565c45
--- /dev/null
+++ b/streams-contrib/streams-provider-youtube/src/test/resources/YoutubeUserActivityProviderIT.conf
@@ -0,0 +1,22 @@
+# 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.
+youtube {
+ youtubeUsers = [
+ {
+ userId = "UCLDJ_V9KUOdOFSbDvPfGBxw"
+ }
+ ]
+}
\ No newline at end of file
[3/3] incubator-streams git commit: Merge branch '0.4-youtube'
Posted by sb...@apache.org.
Merge branch '0.4-youtube'
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/104f29b1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/104f29b1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/104f29b1
Branch: refs/heads/master
Commit: 104f29b1eb52c2843456bbb0466e346e682a6859
Parents: 006234f 9688d32
Author: Steve Blackmon @steveblackmon <sb...@apache.org>
Authored: Wed Oct 19 15:18:38 2016 -0500
Committer: Steve Blackmon @steveblackmon <sb...@apache.org>
Committed: Wed Oct 19 15:18:38 2016 -0500
----------------------------------------------------------------------
.../streams-provider-youtube/pom.xml | 21 +++--
.../provider/YoutubeChannelDataCollector.java | 5 +-
.../provider/YoutubeChannelProvider.java | 84 +++++++++++++++++++
.../com/youtube/provider/YoutubeProvider.java | 86 ++++++++++++++------
.../provider/YoutubeUserActivityCollector.java | 5 +-
.../provider/YoutubeUserActivityProvider.java | 81 ++++++++++++++++++
.../src/site/markdown/index.md | 23 ++++++
.../providers/YoutubeChannelProviderIT.java | 55 +++++++++++++
.../YoutubeUserActivityProviderIT.java | 55 +++++++++++++
.../resources/YoutubeChannelProviderIT.conf | 22 +++++
.../YoutubeUserActivityProviderIT.conf | 22 +++++
11 files changed, 423 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
[2/3] incubator-streams git commit: just 1 line is OK
Posted by sb...@apache.org.
just 1 line is OK
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/9688d32b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/9688d32b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/9688d32b
Branch: refs/heads/master
Commit: 9688d32bfb9f4526cc13d6eaa507481769a8e477
Parents: 762ce8c
Author: Steve Blackmon @steveblackmon <sb...@apache.org>
Authored: Thu Oct 13 13:30:20 2016 -0500
Committer: Steve Blackmon @steveblackmon <sb...@apache.org>
Committed: Thu Oct 13 13:30:20 2016 -0500
----------------------------------------------------------------------
.../streams/youtube/test/providers/YoutubeChannelProviderIT.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/9688d32b/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeChannelProviderIT.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeChannelProviderIT.java b/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeChannelProviderIT.java
index cc1e811..320024a 100644
--- a/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeChannelProviderIT.java
+++ b/streams-contrib/streams-provider-youtube/src/test/java/org/apache/streams/youtube/test/providers/YoutubeChannelProviderIT.java
@@ -49,7 +49,7 @@ public class YoutubeChannelProviderIT {
while(outCounter.readLine() != null) {}
- assert (outCounter.getLineNumber() > 1);
+ assert (outCounter.getLineNumber() >= 1);
}
}