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/03/20 01:53:59 UTC
[33/50] [abbrv] incubator-streams git commit: resolves STREAMS-47
including unit tests of reader and writer
resolves STREAMS-47 including unit tests of reader and writer
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/78105f96
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/78105f96
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/78105f96
Branch: refs/heads/asf-master
Commit: 78105f9690bf9a77872aa00aa7d798804833848a
Parents: 2a16195
Author: sblackmon <sb...@apache.org>
Authored: Wed Feb 18 18:11:14 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Wed Feb 18 18:11:14 2015 -0600
----------------------------------------------------------------------
streams-contrib/streams-persist-mongo/pom.xml | 52 +++++++-
.../streams/mongo/MongoPersistReader.java | 34 ++++--
.../streams/mongo/MongoPersistWriter.java | 17 ++-
.../streams/mongo/test/TestMongoPersist.java | 118 +++++++++++++++++++
4 files changed, 201 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/78105f96/streams-contrib/streams-persist-mongo/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-mongo/pom.xml b/streams-contrib/streams-persist-mongo/pom.xml
index ad9452c..b1c48b1 100644
--- a/streams-contrib/streams-persist-mongo/pom.xml
+++ b/streams-contrib/streams-persist-mongo/pom.xml
@@ -30,7 +30,7 @@
<description>Mongo Module</description>
<properties>
- <mongo-driver.version>2.12.0-rc0</mongo-driver.version>
+ <mongo-driver.version>2.13.0</mongo-driver.version>
</properties>
<dependencies>
@@ -65,6 +65,37 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-pojo</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>com.github.fakemongo</groupId>
+ <artifactId>fongo</artifactId>
+ <version>1.6.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>1.9.5</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-module-junit4</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-api-mockito</artifactId>
+ </dependency>
</dependencies>
<build>
<plugins>
@@ -108,6 +139,25 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <version>2.4</version>
+ <executions>
+ <execution>
+ <id>resource-dependencies</id>
+ <phase>process-test-resources</phase>
+ <goals>
+ <goal>unpack-dependencies</goal>
+ </goals>
+ <configuration>
+ <includeArtifactIds>streams-pojo</includeArtifactIds>
+ <includes>**/*.json</includes>
+ <outputDirectory>${project.build.directory}/test-classes</outputDirectory>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/78105f96/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java
index 4312c6f..fb1fe94 100644
--- a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java
+++ b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java
@@ -59,9 +59,9 @@ public class MongoPersistReader implements StreamsPersistReader {
private ExecutorService executor;
private MongoConfiguration config;
+ private MongoPersistReaderTask readerTask;
protected DB client;
- protected DBAddress dbaddress;
protected DBCollection collection;
protected DBCursor cursor;
@@ -110,7 +110,6 @@ public class MongoPersistReader implements StreamsPersistReader {
connectToMongo();
if( client == null ||
- dbaddress == null ||
collection == null )
throw new RuntimeException("Unable to connect!");
@@ -133,7 +132,18 @@ public class MongoPersistReader implements StreamsPersistReader {
protected StreamsDatum prepareDatum(DBObject dbObject) {
- StreamsDatum datum = new StreamsDatum(dbObject.toString());
+ ObjectNode objectNode;
+ String id;
+
+ try {
+ objectNode = mapper.readValue(dbObject.toString(), ObjectNode.class);
+ id = objectNode.get("_id").get("$oid").asText();
+ objectNode.remove("_id");
+ } catch (IOException e) {
+ LOGGER.warn("document isn't valid JSON.");
+ return null;
+ }
+ StreamsDatum datum = new StreamsDatum(objectNode, id);
return datum;
}
@@ -141,14 +151,12 @@ public class MongoPersistReader implements StreamsPersistReader {
private synchronized void connectToMongo() {
try {
- dbaddress = new DBAddress(config.getHost(), config.getPort().intValue(), config.getDb());
+ client = new MongoClient(config.getHost(), config.getPort().intValue()).getDB(config.getDb());
} catch (UnknownHostException e) {
e.printStackTrace();
return;
}
- client = MongoClient.connect(dbaddress);
-
if (!Strings.isNullOrEmpty(config.getUser()) && !Strings.isNullOrEmpty(config.getPassword()))
client.authenticate(config.getUser(), config.getPassword().toCharArray());
@@ -181,8 +189,16 @@ public class MongoPersistReader implements StreamsPersistReader {
public void startStream() {
LOGGER.debug("startStream");
- MongoPersistReaderTask readerTask = new MongoPersistReaderTask(this);
- executor.submit(readerTask);
+ readerTask = new MongoPersistReaderTask(this);
+ Thread readerTaskThread = new Thread(readerTask);
+ Future future = executor.submit(readerTaskThread);
+
+ while( !future.isDone() && !future.isCancelled()) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {}
+ }
+
executor.shutdown();
}
@@ -236,7 +252,7 @@ public class MongoPersistReader implements StreamsPersistReader {
@Override
public boolean isRunning() {
- return !executor.isTerminated();
+ return !executor.isTerminated() || !executor.isShutdown();
}
private Queue<StreamsDatum> constructQueue() {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/78105f96/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
index 773327c..16bf02a 100644
--- a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
+++ b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
@@ -48,6 +48,7 @@ import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class MongoPersistWriter implements StreamsPersistWriter, Runnable {
+
private final static Logger LOGGER = LoggerFactory.getLogger(MongoPersistWriter.class);
private final static long MAX_WRITE_LATENCY = 1000;
@@ -68,15 +69,11 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable {
protected final ReadWriteLock lock = new ReentrantReadWriteLock();
public MongoPersistWriter() {
- Config config = StreamsConfigurator.config.getConfig("mongo");
- this.config = MongoConfigurator.detectConfiguration(config);
- this.persistQueue = new ConcurrentLinkedQueue<StreamsDatum>();
+ this(MongoConfigurator.detectConfiguration(StreamsConfigurator.config.getConfig("mongo")));
}
- public MongoPersistWriter(Queue<StreamsDatum> persistQueue) {
- Config config = StreamsConfigurator.config.getConfig("mongo");
- this.config = MongoConfigurator.detectConfiguration(config);
- this.persistQueue = persistQueue;
+ public MongoPersistWriter(MongoConfiguration config) {
+ this.config = config;
}
public void setPersistQueue(Queue<StreamsDatum> persistQueue) {
@@ -178,6 +175,7 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable {
@Override
public void prepare(Object configurationObject) {
+ this.persistQueue = new ConcurrentLinkedQueue<StreamsDatum>();
start();
}
@@ -225,15 +223,14 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable {
}
private synchronized void connectToMongo() {
+
try {
- dbaddress = new DBAddress(config.getHost(), config.getPort().intValue(), config.getDb());
+ client = new MongoClient(config.getHost(), config.getPort().intValue()).getDB(config.getDb());
} catch (UnknownHostException e) {
e.printStackTrace();
return;
}
- client = MongoClient.connect(dbaddress);
-
if (!Strings.isNullOrEmpty(config.getUser()) && !Strings.isNullOrEmpty(config.getPassword()))
client.authenticate(config.getUser(), config.getPassword().toCharArray());
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/78105f96/streams-contrib/streams-persist-mongo/src/test/java/org/apache/streams/mongo/test/TestMongoPersist.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-mongo/src/test/java/org/apache/streams/mongo/test/TestMongoPersist.java b/streams-contrib/streams-persist-mongo/src/test/java/org/apache/streams/mongo/test/TestMongoPersist.java
new file mode 100644
index 0000000..d0733e8
--- /dev/null
+++ b/streams-contrib/streams-persist-mongo/src/test/java/org/apache/streams/mongo/test/TestMongoPersist.java
@@ -0,0 +1,118 @@
+/*
+ * 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.mongo.test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.fakemongo.Fongo;
+import com.mongodb.DB;
+import com.mongodb.DBAddress;
+import com.mongodb.MongoClient;
+import org.apache.commons.io.Charsets;
+import org.apache.commons.io.IOUtils;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.mongo.MongoConfiguration;
+import org.apache.streams.mongo.MongoPersistReader;
+import org.apache.streams.mongo.MongoPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.InputStream;
+import java.util.List;
+
+/**
+ * Test copying documents between two indexes on same cluster
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({MongoPersistReader.class, MongoPersistWriter.class, MongoClient.class, DB.class})
+public class TestMongoPersist {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(TestMongoPersist.class);
+
+ ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
+
+ MongoClient mockClient;
+
+ Fongo fongo;
+ DB mockDB;
+
+ int count = 0;
+
+ @Before
+ public void setup() {
+ fongo = new Fongo("testmongo");
+ mockDB = fongo.getDB("test");
+
+ this.mockClient = PowerMockito.mock(MongoClient.class);
+
+ PowerMockito.when(mockClient.getDB(Mockito.anyString()))
+ .thenReturn(mockDB);
+
+ try {
+ PowerMockito.whenNew(MongoClient.class).withAnyArguments().thenReturn(mockClient);
+ } catch (Exception e) {}
+
+ }
+
+ @Test
+ public void testMongoPersist() throws Exception {
+
+ MongoConfiguration mongoConfiguration = new MongoConfiguration().withHost("localhost").withDb("test").withPort(37017l).withCollection("activities");
+
+ MongoPersistWriter writer = new MongoPersistWriter(mongoConfiguration);
+
+ writer.prepare(null);
+
+ InputStream testActivityFolderStream = TestMongoPersist.class.getClassLoader()
+ .getResourceAsStream("activities");
+ List<String> files = IOUtils.readLines(testActivityFolderStream, Charsets.UTF_8);
+
+ for( String file : files) {
+ LOGGER.info("File: " + file );
+ InputStream testActivityFileStream = TestMongoPersist.class.getClassLoader()
+ .getResourceAsStream("activities/" + file);
+ Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
+ activity.getAdditionalProperties().remove("$license");
+ StreamsDatum datum = new StreamsDatum(activity, activity.getVerb());
+ writer.write( datum );
+ LOGGER.info("Wrote: " + activity.getVerb() );
+ count++;
+ }
+
+ writer.cleanUp();
+
+ MongoPersistReader reader = new MongoPersistReader(mongoConfiguration);
+
+ reader.prepare(null);
+
+ StreamsResultSet resultSet = reader.readAll();
+
+ assert( resultSet.size() == count);
+
+ }
+}