Build failed in Jenkins: streams-examples-site #66

[Apache Jenkins Server]

See 

Changes:

[sblackmon] switch to bolt protocol

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on ubuntu-4 (ubuntu) in workspace 

Wiping out workspace first.
Cloning the remote Git repository
Cloning repository 
https://git-wip-us.apache.org/repos/asf/incubator-streams-examples.git
 > git init  # 
 > timeout=10
Fetching upstream changes from 
https://git-wip-us.apache.org/repos/asf/incubator-streams-examples.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams-examples.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams-examples.git # 
 > timeout=10
 > git config --add remote.origin.fetch +refs/heads/*:refs/remotes/origin/* # 
 > timeout=10
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams-examples.git # 
 > timeout=10
Cleaning workspace
 > git rev-parse --verify HEAD # timeout=10
No valid HEAD. Skipping the resetting
 > git clean -fdx # timeout=10
Fetching upstream changes from 
https://git-wip-us.apache.org/repos/asf/incubator-streams-examples.git
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams-examples.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse origin/master^{commit} # timeout=10
Checking out Revision b430357222faacfa27f5db37dee198ed5115ad3d (origin/master)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f b430357222faacfa27f5db37dee198ed5115ad3d
 > git rev-list 27f63c1d1df595289e49e1417055cc2b0342c4da # timeout=10
Parsing POMs
using settings config with name simple-deploy-settings-no-mirror
Replacing all maven server entries not found in credentials list is true
Downloaded artifact 
https://repository.apache.org/content/repositories/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/maven-metadata.xml
Downloaded artifact 
https://repository.apache.org/content/repositories/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/streams-master-0.5-incubating-20170107.212356-47.pom
Downloaded artifact 
https://repository.apache.org/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/maven-metadata.xml
Modules changed, recalculating dependency graph
Established TCP socket on 56104
maven32-agent.jar already up to date
maven32-interceptor.jar already up to date
maven3-interceptor-commons.jar already up to date
[streams-examples-site] $ /home/jenkins/tools/java/latest1.8/bin/java -Xmx2g 
-Xms256m -XX:MaxPermSize=512m -Dfile.encoding=UTF-8 -cp 
/home/jenkins/jenkins-slave/maven32-agent.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/boot/plexus-classworlds-2.5.2.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/conf/logging
 jenkins.maven3.agent.Maven32Main /home/jenkins/tools/maven/apache-maven-3.3.9 
/home/jenkins/jenkins-slave/slave.jar 
/home/jenkins/jenkins-slave/maven32-interceptor.jar 
/home/jenkins/jenkins-slave/maven3-interceptor-commons.jar 56104
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=512m; 
support was removed in 8.0
<===[JENKINS REMOTING CAPACITY]===>   channel started
using settings config with name simple-deploy-settings-no-mirror
Replacing all maven server entries not found in credentials list is true
Executing Maven:  -B -f 
 
-Dmaven.repo.local=/home/jenkins/jenkins-slave/maven-repositories/0 -s 
/tmp/settings666523304563013700.xml -q -U generate-sources site:site site:stage
Saving to 
outputFile=
Processed 0 file(s)
Found 0 errors
Found 0 warnings
Found 0 infos
Finished in 155 ms
[JENKINS] Archiving site from 
 to 
/x1/jenkins/jenkins-home/jobs/streams-examples-site/site
file:// 
- Session: Opened  
file:// 
- Session: Disconnecting  
file:// 
- Session: Disconnected
Saving to 
outputFile=
Processed 0 file(s)
Found 0 errors
Found 0 warnings
Found 0 infos
Finished in 1 ms
[JENKINS] Archiving site from 
 to 
/x1/jenkins/jenkins-home/jobs/streams-examples-site/site/streams-examples-flink

Build failed in Jenkins: streams-project-site #128

[Apache Jenkins Server]

See 

--
Started by upstream project "streams-project" build number 342
originally caused by:
 Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on H11 (ubuntu) in workspace 

Wiping out workspace first.
Cloning the remote Git repository
Cloning repository https://git-wip-us.apache.org/repos/asf/incubator-streams.git
 > git init  # 
 > timeout=10
Fetching upstream changes from 
https://git-wip-us.apache.org/repos/asf/incubator-streams.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams.git # timeout=10
 > git config --add remote.origin.fetch +refs/heads/*:refs/remotes/origin/* # 
 > timeout=10
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams.git # timeout=10
Cleaning workspace
 > git rev-parse --verify HEAD # timeout=10
No valid HEAD. Skipping the resetting
 > git clean -fdx # timeout=10
Fetching upstream changes from 
https://git-wip-us.apache.org/repos/asf/incubator-streams.git
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse origin/master^{commit} # timeout=10
Checking out Revision 9ef999423584ce4e004443f0b704b843e8fa0234 (origin/master)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 9ef999423584ce4e004443f0b704b843e8fa0234
 > git rev-list 9ef999423584ce4e004443f0b704b843e8fa0234 # timeout=10
Parsing POMs
using settings config with name simple-deploy-settings-no-mirror
Replacing all maven server entries not found in credentials list is true
Downloaded artifact 
https://repository.apache.org/content/repositories/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/maven-metadata.xml
Failed to transfer Could not find metadata 
org.apache.streams:streams-master:0.5-incubating-SNAPSHOT/maven-metadata.xml in 
sonatype (https://oss.sonatype.org/content/repositories/snapshots)
Failed to transfer Could not find metadata 
org.apache.streams:streams-master:0.5-incubating-SNAPSHOT/maven-metadata.xml in 
typesafe (http://repo.typesafe.com/typesafe/repo)
Failed to transfer Could not find metadata 
org.apache.streams:streams-master:0.5-incubating-SNAPSHOT/maven-metadata.xml in 
google-api-services (http://google-api-client-libraries.appspot.com/mavenrepo)
Downloaded artifact 
https://repository.apache.org/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/maven-metadata.xml
Modules changed, recalculating dependency graph
Established TCP socket on 50520
maven32-agent.jar already up to date
maven32-interceptor.jar already up to date
maven3-interceptor-commons.jar already up to date
[streams-project-site] $ /home/jenkins/tools/java/latest1.8/bin/java -Xmx2g 
-Xms256m -XX:MaxPermSize=512m -Dfile.encoding=UTF-8 -cp 
/home/jenkins/jenkins-slave/maven32-agent.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/boot/plexus-classworlds-2.5.2.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/conf/logging
 jenkins.maven3.agent.Maven32Main /home/jenkins/tools/maven/apache-maven-3.3.9 
/home/jenkins/jenkins-slave/slave.jar 
/home/jenkins/jenkins-slave/maven32-interceptor.jar 
/home/jenkins/jenkins-slave/maven3-interceptor-commons.jar 50520
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=512m; 
support was removed in 8.0
<===[JENKINS REMOTING CAPACITY]===>   channel started
using settings config with name simple-deploy-settings-no-mirror
Replacing all maven server entries not found in credentials list is true
Executing Maven:  -B -f 
 
-Dmaven.repo.local=/home/jenkins/jenkins-slave/maven-repositories/0 -s 
/tmp/settings4463282315120187356.xml -q -U generate-sources site:site site:stage
Saving to 
outputFile=
Processed 0 file(s)
Found 0 errors
Found 0 warnings
Found 0 infos
Finished in 145 ms
[JENKINS] Archiving site from 
 to 
/x1/jenkins/jenkins-home/jobs/streams-project-site/site
file:// 
- Session: Opened  
file:// 
- Session: Disconnecting  
file:// 
- Session: Disconnected
Saving to 
outputFile=

Build failed in Jenkins: streams-examples #181

[Apache Jenkins Server]

See 

--
[...truncated 17 lines...]
using settings config with name simple-deploy-settings-no-mirror
Replacing all maven server entries not found in credentials list is true
Modules changed, recalculating dependency graph
Established TCP socket on 40631
maven32-agent.jar already up to date
maven32-interceptor.jar already up to date
maven3-interceptor-commons.jar already up to date
[streams-examples] $ /home/jenkins/tools/java/latest1.8/bin/java -Xmx2g 
-Xms256m -XX:MaxPermSize=512m -Dfile.encoding=ISO-8859-1 -cp 
/home/jenkins/jenkins-slave/maven32-agent.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/boot/plexus-classworlds-2.5.2.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/conf/logging
 jenkins.maven3.agent.Maven32Main /home/jenkins/tools/maven/apache-maven-3.3.9 
/home/jenkins/jenkins-slave/slave.jar 
/home/jenkins/jenkins-slave/maven32-interceptor.jar 
/home/jenkins/jenkins-slave/maven3-interceptor-commons.jar 40631
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=512m; 
support was removed in 8.0
<===[JENKINS REMOTING CAPACITY]===>   channel started
using settings config with name simple-deploy-settings-no-mirror
Replacing all maven server entries not found in credentials list is true
Executing Maven:  -B -f 
 
-Dmaven.repo.local=/home/jenkins/jenkins-slave/maven-repositories/1 -s 
/tmp/settings6201961430641665702.xml clean install verify
[INFO] Scanning for projects...
[INFO] 
[INFO] Reactor Build Order:
[INFO] 
[INFO] streams-examples
[INFO] streams-examples-flink
[INFO] flink-twitter-collection
[INFO] streams-examples-local
[INFO] elasticsearch-hdfs
[INFO] elasticsearch-reindex
[INFO] mongo-elasticsearch-sync
[INFO] twitter-follow-neo4j
[INFO] twitter-history-elasticsearch
[INFO] twitter-userstream-elasticsearch
[INFO] 
[INFO] 
[INFO] Building streams-examples 0.5-incubating-SNAPSHOT
[INFO] 
[INFO] 
[INFO] --- maven-clean-plugin:2.6:clean (default-clean) @ streams-examples ---
[INFO] Deleting 
[INFO] 
[INFO] --- maven-resources-plugin:2.7:copy-resources (copy-resources) @ 
streams-examples ---
[INFO] Using 'UTF-8' encoding to copy filtered resources.
[INFO] skip non existing resourceDirectory 

[INFO] 
[INFO] --- maven-enforcer-plugin:1.4.1:enforce (enforce-java) @ 
streams-examples ---
[INFO] 
[INFO] --- maven-enforcer-plugin:1.4.1:enforce (enforce-banned-dependencies) @ 
streams-examples ---
[INFO] 
[INFO] --- maven-enforcer-plugin:1.4.1:enforce (enforce-no-snapshots) @ 
streams-examples ---
[INFO] 
[INFO] --- maven-checkstyle-plugin:2.17:check (validate) @ streams-examples ---
[INFO] Starting audit...
Audit done.
[INFO] 
[INFO] --- scalastyle-maven-plugin:0.8.0:check (validate) @ streams-examples ---
[WARNING] sourceDirectory is not specified or does not exist 
value=
[WARNING] testSourceDirectory is not specified or does not exist 
value=
Saving to 
outputFile=
Processed 0 file(s)
Found 0 errors
Found 0 warnings
Found 0 infos
Finished in 132 ms
[INFO] 
[INFO] --- maven-remote-resources-plugin:1.4:process (default) @ 
streams-examples ---
[INFO] 
[INFO] --- maven-remote-resources-plugin:1.4:process (process-resource-bundles) 
@ streams-examples ---
[INFO] 
[INFO] --- maven-surefire-plugin:2.19.1:test (test) @ streams-examples ---
[INFO] No tests to run.
[JENKINS] Recording test results[INFO] 
[INFO] --- maven-site-plugin:3.5:attach-descriptor (attach-descriptor) @ 
streams-examples ---

[INFO] 
[INFO] --- maven-failsafe-plugin:2.19.1:integration-test (integration-tests) @ 
streams-examples ---
[INFO] Tests are skipped.
[INFO] 
[INFO] --- maven-failsafe-plugin:2.19.1:verify (integration-tests) @ 
streams-examples ---
[INFO] Tests are skipped.
[INFO] 
[INFO] --- maven-install-plugin:2.5.2:install (default-install) @ 
streams-examples ---
[INFO] Installing  
to 
/home/jenkins/jenkins-slave/maven-repositories/1/org/apache/streams/streams-examples/0.5-incubating-SNAPSHOT/streams-examples-0.5-incubating-SNAPSHOT.pom
[INFO] Installing 

 to 

Build failed in Jenkins: streams-examples #180

[Apache Jenkins Server]

See 

Changes:

[sblackmon] switch to bolt protocol

--
[...truncated 18 lines...]
Downloaded artifact 
https://repository.apache.org/content/repositories/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/streams-master-0.5-incubating-20170107.212356-47.pom
Downloaded artifact 
https://repository.apache.org/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/maven-metadata.xml
Modules changed, recalculating dependency graph
Established TCP socket on 39176
maven32-agent.jar already up to date
maven32-interceptor.jar already up to date
maven3-interceptor-commons.jar already up to date
[streams-examples] $ /home/jenkins/tools/java/latest1.8/bin/java -Xmx2g 
-Xms256m -XX:MaxPermSize=512m -Dfile.encoding=ISO-8859-1 -cp 
/home/jenkins/jenkins-slave/maven32-agent.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/boot/plexus-classworlds-2.5.2.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/conf/logging
 jenkins.maven3.agent.Maven32Main /home/jenkins/tools/maven/apache-maven-3.3.9 
/home/jenkins/jenkins-slave/slave.jar 
/home/jenkins/jenkins-slave/maven32-interceptor.jar 
/home/jenkins/jenkins-slave/maven3-interceptor-commons.jar 39176
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=512m; 
support was removed in 8.0
<===[JENKINS REMOTING CAPACITY]===>   channel started
using settings config with name simple-deploy-settings-no-mirror
Replacing all maven server entries not found in credentials list is true
Executing Maven:  -B -f 
 
-Dmaven.repo.local=/home/jenkins/jenkins-slave/maven-repositories/1 -s 
/tmp/settings2663867164171599509.xml clean install verify
[INFO] Scanning for projects...
[INFO] 
[INFO] Reactor Build Order:
[INFO] 
[INFO] streams-examples
[INFO] streams-examples-flink
[INFO] flink-twitter-collection
[INFO] streams-examples-local
[INFO] elasticsearch-hdfs
[INFO] elasticsearch-reindex
[INFO] mongo-elasticsearch-sync
[INFO] twitter-follow-neo4j
[INFO] twitter-history-elasticsearch
[INFO] twitter-userstream-elasticsearch
[INFO] 
[INFO] 
[INFO] Building streams-examples 0.5-incubating-SNAPSHOT
[INFO] 
[INFO] 
[INFO] --- maven-clean-plugin:2.6:clean (default-clean) @ streams-examples ---
[INFO] Deleting 
[INFO] 
[INFO] --- maven-resources-plugin:2.7:copy-resources (copy-resources) @ 
streams-examples ---
[INFO] Using 'UTF-8' encoding to copy filtered resources.
[INFO] skip non existing resourceDirectory 

[INFO] 
[INFO] --- maven-enforcer-plugin:1.4.1:enforce (enforce-java) @ 
streams-examples ---
[INFO] 
[INFO] --- maven-enforcer-plugin:1.4.1:enforce (enforce-banned-dependencies) @ 
streams-examples ---
[INFO] 
[INFO] --- maven-enforcer-plugin:1.4.1:enforce (enforce-no-snapshots) @ 
streams-examples ---
[INFO] 
[INFO] --- maven-checkstyle-plugin:2.17:check (validate) @ streams-examples ---
[INFO] Starting audit...
Audit done.
[INFO] 
[INFO] --- scalastyle-maven-plugin:0.8.0:check (validate) @ streams-examples ---
[WARNING] sourceDirectory is not specified or does not exist 
value=
[WARNING] testSourceDirectory is not specified or does not exist 
value=
Saving to 
outputFile=
Processed 0 file(s)
Found 0 errors
Found 0 warnings
Found 0 infos
Finished in 148 ms
[INFO] 
[INFO] --- maven-remote-resources-plugin:1.4:process (default) @ 
streams-examples ---
[INFO] 
[INFO] --- maven-remote-resources-plugin:1.4:process (process-resource-bundles) 
@ streams-examples ---
[INFO] 
[INFO] --- maven-surefire-plugin:2.19.1:test (test) @ streams-examples ---
[INFO] No tests to run.
[JENKINS] Recording test results[INFO] 
[INFO] --- maven-site-plugin:3.5:attach-descriptor (attach-descriptor) @ 
streams-examples ---

[INFO] 
[INFO] --- maven-failsafe-plugin:2.19.1:integration-test (integration-tests) @ 
streams-examples ---
[INFO] Tests are skipped.
[INFO] 
[INFO] --- maven-failsafe-plugin:2.19.1:verify (integration-tests) @ 
streams-examples ---
[INFO] Tests are skipped.
[INFO] 
[INFO] --- maven-install-plugin:2.5.2:install (default-install) @ 
streams-examples ---
[INFO] Installing  
to 

incubator-streams-examples git commit: switch to bolt protocol

[sblackmon]

Repository: incubator-streams-examples
Updated Branches:
  refs/heads/master 27f63c1d1 -> b43035722


switch to bolt protocol


Project: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/repo
Commit: 
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/commit/b4303572
Tree: 
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/tree/b4303572
Diff: 
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/diff/b4303572

Branch: refs/heads/master
Commit: b430357222faacfa27f5db37dee198ed5115ad3d
Parents: 27f63c1
Author: Steve Blackmon 
Authored: Mon Jan 2 19:44:09 2017 -0600
Committer: Steve Blackmon 
Committed: Mon Jan 2 19:44:09 2017 -0600

--
 local/twitter-follow-neo4j/pom.xml  |  4 +-
 .../streams/example/TwitterFollowNeo4j.java | 18 +++---
 .../TwitterFollowNeo4jConfiguration.json|  2 +-
 .../example/test/TwitterFollowNeo4jIT.java  | 62 ++--
 .../test/resources/TwitterFollowNeo4jIT.conf|  9 ++-
 pom.xml |  4 +-
 6 files changed, 77 insertions(+), 22 deletions(-)
--


http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/b4303572/local/twitter-follow-neo4j/pom.xml
--
diff --git a/local/twitter-follow-neo4j/pom.xml 
b/local/twitter-follow-neo4j/pom.xml
index 3fd9dd1..def7ae7 100644
--- a/local/twitter-follow-neo4j/pom.xml
+++ b/local/twitter-follow-neo4j/pom.xml
@@ -78,7 +78,7 @@
 
 
 org.apache.streams
-streams-persist-graph
+streams-persist-neo4j
 
 
 org.apache.streams
@@ -175,7 +175,7 @@
 
 
 org.apache.streams
-streams-persist-graph
+streams-persist-neo4j
 ${project.version}
 
 

http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/b4303572/local/twitter-follow-neo4j/src/main/java/org/apache/streams/example/TwitterFollowNeo4j.java
--
diff --git 
a/local/twitter-follow-neo4j/src/main/java/org/apache/streams/example/TwitterFollowNeo4j.java
 
b/local/twitter-follow-neo4j/src/main/java/org/apache/streams/example/TwitterFollowNeo4j.java
index e00e4a0..5098061 100644
--- 
a/local/twitter-follow-neo4j/src/main/java/org/apache/streams/example/TwitterFollowNeo4j.java
+++ 
b/local/twitter-follow-neo4j/src/main/java/org/apache/streams/example/TwitterFollowNeo4j.java
@@ -26,11 +26,11 @@ import org.apache.streams.converter.TypeConverterProcessor;
 import org.apache.streams.core.StreamBuilder;
 import org.apache.streams.data.ActivityConverter;
 import org.apache.streams.data.DocumentClassifier;
-import org.apache.streams.graph.GraphHttpConfiguration;
-import org.apache.streams.graph.GraphHttpPersistWriter;
 import org.apache.streams.jackson.StreamsJacksonMapper;
 import org.apache.streams.local.LocalRuntimeConfiguration;
 import org.apache.streams.local.builders.LocalStreamBuilder;
+import org.apache.streams.neo4j.Neo4jConfiguration;
+import org.apache.streams.neo4j.bolt.Neo4jBoltPersistWriter;
 import org.apache.streams.twitter.TwitterFollowingConfiguration;
 import org.apache.streams.twitter.converter.TwitterDocumentClassifier;
 import org.apache.streams.twitter.converter.TwitterFollowActivityConverter;
@@ -39,6 +39,7 @@ import 
org.apache.streams.twitter.provider.TwitterFollowingProvider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -66,14 +67,17 @@ public class TwitterFollowNeo4j implements Runnable {
 TwitterFollowingProvider followingProvider = new 
TwitterFollowingProvider(twitterFollowingConfiguration);
 TypeConverterProcessor converter = new 
TypeConverterProcessor(String.class);
 
+List classifiers = Stream.of((DocumentClassifier) new 
TwitterDocumentClassifier()).collect(Collectors.toList());
+List converters = Stream.of((ActivityConverter) new 
TwitterFollowActivityConverter()).collect(Collectors.toList());
 ActivityConverterProcessorConfiguration 
activityConverterProcessorConfiguration =
 new ActivityConverterProcessorConfiguration()
-.withClassifiers(Stream.of((DocumentClassifier) new 
TwitterDocumentClassifier()).collect(Collectors.toList()))
-.withConverters(Stream.of((ActivityConverter) new 
TwitterFollowActivityConverter()).collect(Collectors.toList()));
+.withClassifiers(classifiers)
+.withConverters(converters);
 ActivityConverterProcessor activity = new 

Build failed in Jenkins: streams-project-site #127

[Apache Jenkins Server]

See 

Changes:

[sblackmon] STREAMS-472: fixes root cause

[sblackmon] STREAMS:344: streams-persist-neo4j

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on H11 (ubuntu) in workspace 

Wiping out workspace first.
Cloning the remote Git repository
Cloning repository https://git-wip-us.apache.org/repos/asf/incubator-streams.git
 > git init  # 
 > timeout=10
Fetching upstream changes from 
https://git-wip-us.apache.org/repos/asf/incubator-streams.git
 > git --version # timeout=10
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams.git # timeout=10
 > git config --add remote.origin.fetch +refs/heads/*:refs/remotes/origin/* # 
 > timeout=10
 > git config remote.origin.url 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams.git # timeout=10
Cleaning workspace
 > git rev-parse --verify HEAD # timeout=10
No valid HEAD. Skipping the resetting
 > git clean -fdx # timeout=10
Fetching upstream changes from 
https://git-wip-us.apache.org/repos/asf/incubator-streams.git
 > git -c core.askpass=true fetch --tags --progress 
 > https://git-wip-us.apache.org/repos/asf/incubator-streams.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse origin/master^{commit} # timeout=10
Checking out Revision 9ef999423584ce4e004443f0b704b843e8fa0234 (origin/master)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 9ef999423584ce4e004443f0b704b843e8fa0234
 > git rev-list 7810361d2e5a2f236967b3208053ec80b0e5885a # timeout=10
Parsing POMs
using settings config with name simple-deploy-settings-no-mirror
Replacing all maven server entries not found in credentials list is true
Downloaded artifact 
https://repository.apache.org/content/repositories/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/maven-metadata.xml
Downloaded artifact 
https://repository.apache.org/content/repositories/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/streams-master-0.5-incubating-20170107.212356-47.pom
Failed to transfer Could not find metadata 
org.apache.streams:streams-master:0.5-incubating-SNAPSHOT/maven-metadata.xml in 
sonatype (https://oss.sonatype.org/content/repositories/snapshots)
Failed to transfer Could not find metadata 
org.apache.streams:streams-master:0.5-incubating-SNAPSHOT/maven-metadata.xml in 
typesafe (http://repo.typesafe.com/typesafe/repo)
Failed to transfer Could not find metadata 
org.apache.streams:streams-master:0.5-incubating-SNAPSHOT/maven-metadata.xml in 
google-api-services (http://google-api-client-libraries.appspot.com/mavenrepo)
Downloaded artifact 
https://repository.apache.org/snapshots/org/apache/streams/streams-master/0.5-incubating-SNAPSHOT/maven-metadata.xml
Discovered a new module org.apache.streams:streams-persist-neo4j 
streams-persist-neo4j
Modules changed, recalculating dependency graph
Established TCP socket on 49086
maven32-agent.jar already up to date
maven32-interceptor.jar already up to date
maven3-interceptor-commons.jar already up to date
[streams-project-site] $ /home/jenkins/tools/java/latest1.8/bin/java -Xmx2g 
-Xms256m -XX:MaxPermSize=512m -Dfile.encoding=UTF-8 -cp 
/home/jenkins/jenkins-slave/maven32-agent.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/boot/plexus-classworlds-2.5.2.jar:/home/jenkins/tools/maven/apache-maven-3.3.9/conf/logging
 jenkins.maven3.agent.Maven32Main /home/jenkins/tools/maven/apache-maven-3.3.9 
/home/jenkins/jenkins-slave/slave.jar 
/home/jenkins/jenkins-slave/maven32-interceptor.jar 
/home/jenkins/jenkins-slave/maven3-interceptor-commons.jar 49086
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=512m; 
support was removed in 8.0
<===[JENKINS REMOTING CAPACITY]===>   channel started
using settings config with name simple-deploy-settings-no-mirror
Replacing all maven server entries not found in credentials list is true
Executing Maven:  -B -f 
 
-Dmaven.repo.local=/home/jenkins/jenkins-slave/maven-repositories/0 -s 
/tmp/settings8166769403848047059.xml -q -U generate-sources site:site site:stage
Saving to 
outputFile=
Processed 0 file(s)
Found 0 errors
Found 0 warnings
Found 0 infos
Finished in 396 ms
[JENKINS] Archiving site from 
 to 
/x1/jenkins/jenkins-home/jobs/streams-project-site/site
file:// 
- Session: Opened  

[jira] [Commented] (STREAMS-472) Fix failing IT in streams-plugins

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-472?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809831#comment-15809831
 ] 

ASF GitHub Bot commented on STREAMS-472:


Github user asfgit closed the pull request at:

https://github.com/apache/incubator-streams/pull/349


> Fix failing IT in streams-plugins
> -
>
> Key: STREAMS-472
> URL: https://issues.apache.org/jira/browse/STREAMS-472
> Project: Streams
>  Issue Type: Bug
>Affects Versions: 0.4.1
>Reporter: Steve Blackmon
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>   Original Estimate: 4h
>  Remaining Estimate: 4h
>
> Appears that these test poms have an outdated version dependency, 
> 0.4-incubating-SNAPSHOT instead of 0.5-incubating-SNAPSHO



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-streams pull request #349: STREAMS-472: fixes root cause

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/incubator-streams/pull/349


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[2/2] incubator-streams git commit: Merge branch 'STREAMS-472.2'

[sblackmon]

Merge branch 'STREAMS-472.2'


Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/9ef99942
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/9ef99942
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/9ef99942

Branch: refs/heads/master
Commit: 9ef999423584ce4e004443f0b704b843e8fa0234
Parents: 08b6fac 16edd38
Author: Steve Blackmon @steveblackmon 
Authored: Sun Jan 8 12:43:54 2017 -0600
Committer: Steve Blackmon @steveblackmon 
Committed: Sun Jan 8 12:43:54 2017 -0600

--
 .../java/org/apache/streams/util/schema/SchemaStoreImpl.java| 5 -
 .../org/apache/streams/util/schema/test/SchemaOrderingTest.java | 3 ++-
 2 files changed, 6 insertions(+), 2 deletions(-)
--

incubator-streams git commit: resolves #348

[sblackmon]

Repository: incubator-streams
Updated Branches:
  refs/heads/master 4bd22317e -> 08b6fac0d


resolves #348


Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/08b6fac0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/08b6fac0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/08b6fac0

Branch: refs/heads/master
Commit: 08b6fac0da52eda7495dcf48a51f56bfe0541345
Parents: 4bd2231
Author: Steve Blackmon @steveblackmon 
Authored: Sun Jan 8 12:42:30 2017 -0600
Committer: Steve Blackmon @steveblackmon 
Committed: Sun Jan 8 12:42:30 2017 -0600

--

--

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/incubator-streams/pull/348


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809824#comment-15809824
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user smarthi commented on the issue:

https://github.com/apache/incubator-streams/pull/348
  
+1 to merge


> Support binary protocol in streams-persist-graph
> 
>
> Key: STREAMS-344
> URL: https://issues.apache.org/jira/browse/STREAMS-344
> Project: Streams
>  Issue Type: Improvement
>  Components: Persist
>Reporter: Steve Blackmon
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>
> Support batch writes in GraphHttpPersistWriter.
> Using a separate HTTP Post for every datum is inefficient.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-streams issue #348: STREAMS-344: streams-persist-neo4j

[smarthi]

Github user smarthi commented on the issue:

https://github.com/apache/incubator-streams/pull/348
  
+1 to merge


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[2/2] incubator-streams git commit: STREAMS:344: streams-persist-neo4j

[sblackmon]

STREAMS:344: streams-persist-neo4j

Squashed commit of the following:

commit 76207b1577a0fb6f05992c8700151223db20e4b3
Author: Steve Blackmon @steveblackmon 
Date:   Sun Jan 8 12:29:32 2017 -0600

STREAMS-344: address PR feedback

from https://github.com/apache/incubator-streams/pull/348

commit ee700fd16e8631bdb0fb453d686beef4167af13b
Author: Steve Blackmon 
Date:   Mon Jan 2 19:42:33 2017 -0600

add constructor

commit 1f4e175cf84a208252d488c2858ea420af0642f9
Author: Steve Blackmon 
Date:   Mon Jan 2 18:11:01 2017 -0600

new neo4j module with bolt:// and http:// support, and tests


Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/4bd22317
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/4bd22317
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/4bd22317

Branch: refs/heads/master
Commit: 4bd22317ea3a67b7dfdc0c9d3aba96a71f712e3a
Parents: 7810361
Author: Steve Blackmon @steveblackmon 
Authored: Sun Jan 8 12:36:18 2017 -0600
Committer: Steve Blackmon @steveblackmon 
Committed: Sun Jan 8 12:36:18 2017 -0600

--
 streams-contrib/pom.xml |   3 +-
 streams-contrib/streams-persist-graph/pom.xml   |  20 +-
 .../streams/graph/GraphHttpPersistWriter.java   | 250 --
 .../apache/streams/graph/GraphVertexReader.java | 126 ---
 .../apache/streams/graph/HttpGraphHelper.java   |   4 +-
 .../apache/streams/graph/QueryGraphHelper.java  |   4 +-
 .../graph/neo4j/CypherQueryGraphHelper.java | 238 -
 .../graph/neo4j/Neo4jHttpGraphHelper.java   |  75 
 .../streams/graph/GraphBinaryConfiguration.json |  28 --
 .../streams/graph/GraphConfiguration.json   |  22 --
 .../streams/graph/GraphHttpConfiguration.json   |  22 --
 .../graph/neo4j/CypherQueryResponse.json|  43 ---
 .../graph/test/TestCypherQueryGraphHelper.java  | 116 ---
 .../graph/test/TestNeo4jHttpVertexReader.java   |  81 -
 streams-contrib/streams-persist-neo4j/pom.xml   | 263 ++
 .../streams/neo4j/CypherQueryGraphHelper.java   | 344 +++
 .../apache/streams/neo4j/Neo4jPersistUtil.java  | 151 
 .../streams/neo4j/bolt/Neo4jBoltClient.java |  92 +
 .../neo4j/bolt/Neo4jBoltPersistReader.java  | 326 ++
 .../neo4j/bolt/Neo4jBoltPersistWriter.java  |  77 +
 .../streams/neo4j/http/Neo4jHttpClient.java |  74 
 .../neo4j/http/Neo4jHttpGraphHelper.java| 104 ++
 .../neo4j/http/Neo4jHttpPersistReader.java  | 173 ++
 .../neo4j/http/Neo4jHttpPersistWriter.java  | 171 +
 .../streams/neo4j/CypherQueryResponse.json  |  43 +++
 .../streams/neo4j/Neo4jConfiguration.json   |  27 ++
 .../streams/neo4j/Neo4jReaderConfiguration.json |  17 +
 .../streams/neo4j/test/Neo4jBoltPersistIT.java  | 156 +
 .../streams/neo4j/test/Neo4jHttpPersistIT.java  | 138 
 .../neo4j/test/TestCypherQueryGraphHelper.java  | 150 
 .../src/test/resources/Neo4jBoltPersistIT.conf  |  20 ++
 .../src/test/resources/Neo4jHttpPersistIT.conf  |  20 ++
 .../apache/streams/data/util/PropertyUtil.java  | 124 ---
 .../org/apache/streams/util/PropertyUtil.java   | 130 +++
 .../util/schema/test/PropertyUtilTest.java  |  25 ++
 35 files changed, 2510 insertions(+), 1147 deletions(-)
--


http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/4bd22317/streams-contrib/pom.xml
--
diff --git a/streams-contrib/pom.xml b/streams-contrib/pom.xml
index 8408cef..aed60c9 100644
--- a/streams-contrib/pom.xml
+++ b/streams-contrib/pom.xml
@@ -37,8 +37,8 @@
 
 
 
-streams-persist-console
 streams-persist-cassandra
+streams-persist-console
 streams-persist-elasticsearch
 streams-persist-filebuffer
 streams-persist-hbase
@@ -46,6 +46,7 @@
 streams-persist-graph
 streams-persist-kafka
 streams-persist-mongo
+streams-persist-neo4j
 streams-amazon-aws
 streams-processor-jackson
 streams-processor-json

http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/4bd22317/streams-contrib/streams-persist-graph/pom.xml
--
diff --git a/streams-contrib/streams-persist-graph/pom.xml 
b/streams-contrib/streams-persist-graph/pom.xml
index b8db538..996c706 100644
--- a/streams-contrib/streams-persist-graph/pom.xml
+++ b/streams-contrib/streams-persist-graph/pom.xml
@@ -147,25 +147,7 @@
 
 
 
-
-

[1/2] incubator-streams git commit: STREAMS:344: streams-persist-neo4j

[sblackmon]

Repository: incubator-streams
Updated Branches:
  refs/heads/master 7810361d2 -> 4bd22317e


http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/4bd22317/streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/bolt/Neo4jBoltPersistReader.java
--
diff --git 
a/streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/bolt/Neo4jBoltPersistReader.java
 
b/streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/bolt/Neo4jBoltPersistReader.java
new file mode 100644
index 000..171dde4
--- /dev/null
+++ 
b/streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/bolt/Neo4jBoltPersistReader.java
@@ -0,0 +1,326 @@
+/*
+ * 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.neo4j.bolt;
+
+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.StreamsPersistReader;
+import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.neo4j.Neo4jReaderConfiguration;
+import org.apache.streams.util.PropertyUtil;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Queues;
+
+import org.joda.time.DateTime;
+import org.neo4j.driver.internal.value.NodeValue;
+import org.neo4j.driver.internal.value.RelationshipValue;
+import org.neo4j.driver.internal.value.StringValue;
+import org.neo4j.driver.v1.Record;
+import org.neo4j.driver.v1.Session;
+import org.neo4j.driver.v1.StatementResult;
+import org.neo4j.driver.v1.Transaction;
+import org.neo4j.driver.v1.Value;
+import org.neo4j.driver.v1.types.Node;
+import org.neo4j.driver.v1.types.Relationship;
+import org.neo4j.driver.v1.util.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import javax.annotation.Nullable;
+
+
+/**
+ * Neo4jBoltPersistReader reads documents from neo4j.
+ */
+public class Neo4jBoltPersistReader implements StreamsPersistReader {
+
+  public static final String STREAMS_ID = "CassandraPersistReader";
+
+  public static final Logger LOGGER = 
LoggerFactory.getLogger(Neo4jBoltPersistReader.class);
+
+  protected volatile Queue persistQueue;
+
+  private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private ExecutorService executor;
+  private CompletableFuture readerTaskFuture = new 
CompletableFuture<>();
+
+  private Neo4jReaderConfiguration config;
+
+  protected Neo4jBoltClient client;
+
+//  protected Cluster cluster;
+//  protected Session session;
+//
+//  protected String keyspace;
+//  protected String table;
+  protected StatementResult statementResult;
+
+  protected final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+  /**
+   * Neo4jBoltPersistReader constructor - resolves Neo4jReaderConfiguration 
from JVM 'neo4j'.
+   */
+  public Neo4jBoltPersistReader() {
+this.config = new ComponentConfigurator<>(Neo4jReaderConfiguration.class)
+  .detectConfiguration(StreamsConfigurator.getConfig().getConfig("neo4j"));
+  }
+
+  /**
+   * Neo4jBoltPersistReader constructor - uses supplied 
Neo4jReaderConfiguration.
+   * @param config config
+   */
+  public Neo4jBoltPersistReader(Neo4jReaderConfiguration config) {
+this.config = config;
+  }
+
+  /**
+   * Neo4jBoltPersistReader constructor - uses supplied persistQueue.
+   * @param persistQueue persistQueue
+   */
+  public Neo4jBoltPersistReader(Queue persistQueue) {
+this();
+this.persistQueue = persistQueue;
+  }
+
+  public void setPersistQueue(Queue 

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809812#comment-15809812
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085434
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/http/Neo4jHttpClient.java
 ---
@@ -0,0 +1,84 @@
+package org.apache.streams.neo4j.http;
+
+import org.apache.streams.neo4j.Neo4jConfiguration;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.http.client.HttpClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class Neo4jHttpClient {
+
+private static final Logger LOGGER = LoggerFactory
+.getLogger(Neo4jHttpClient.class);
+
+public Neo4jConfiguration config;
+
+private HttpClient client;
+
+private Neo4jHttpClient(Neo4jConfiguration neo4jConfiguration) {
+this.config = neo4jConfiguration;
+try {
+this.start();
+} catch (Exception e) {
+e.printStackTrace();
+this.client = null;
+}
+}
+
+private static Map INSTANCE_MAP = 
new ConcurrentHashMap();
+
+public static Neo4jHttpClient getInstance(Neo4jConfiguration 
neo4jConfiguration) {
+if (INSTANCE_MAP != null &&
+INSTANCE_MAP.size() > 0 &&
+INSTANCE_MAP.containsKey(neo4jConfiguration)
+)
+return INSTANCE_MAP.get(neo4jConfiguration);
+else {
+Neo4jHttpClient instance = new 
Neo4jHttpClient(neo4jConfiguration);
+if( instance != null && instance.client != null ) {
+INSTANCE_MAP.put(neo4jConfiguration, instance);
+return instance;
+} else {
+return null;
+}
+}
+}
+
+public void start() throws Exception {
+
+Preconditions.checkNotNull(config);
+Preconditions.checkArgument(
+config.getScheme().startsWith("http")
+);
+
+LOGGER.info("Neo4jConfiguration.start {}", config);
+
+Preconditions.checkNotNull(client);
--- End diff --

✔️ 


> Support binary protocol in streams-persist-graph
> 
>
> Key: STREAMS-344
> URL: https://issues.apache.org/jira/browse/STREAMS-344
> Project: Streams
>  Issue Type: Improvement
>  Components: Persist
>Reporter: Steve Blackmon
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>
> Support batch writes in GraphHttpPersistWriter.
> Using a separate HTTP Post for every datum is inefficient.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809811#comment-15809811
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085431
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/bolt/Neo4jBoltClient.java
 ---
@@ -0,0 +1,100 @@
+package org.apache.streams.neo4j.bolt;
+
+import org.apache.streams.neo4j.Neo4jConfiguration;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.commons.lang3.StringUtils;
+import org.neo4j.driver.v1.AuthToken;
+import org.neo4j.driver.v1.AuthTokens;
+import org.neo4j.driver.v1.Driver;
+import org.neo4j.driver.v1.GraphDatabase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class Neo4jBoltClient {
+
+private static final Logger LOGGER = LoggerFactory
+.getLogger(Neo4jBoltClient.class);
+
+private Driver client;
+
+public Neo4jConfiguration config;
+
+private Neo4jBoltClient(Neo4jConfiguration neo4jConfiguration) {
+this.config = neo4jConfiguration;
+try {
+this.start();
+} catch (Exception e) {
+e.printStackTrace();
+this.client = null;
+}
+}
+
+private static Map INSTANCE_MAP = 
new ConcurrentHashMap();
+
+public static Neo4jBoltClient getInstance(Neo4jConfiguration 
neo4jConfiguration) {
+if (INSTANCE_MAP != null &&
--- End diff --

✔️ 


> Support binary protocol in streams-persist-graph
> 
>
> Key: STREAMS-344
> URL: https://issues.apache.org/jira/browse/STREAMS-344
> Project: Streams
>  Issue Type: Improvement
>  Components: Persist
>Reporter: Steve Blackmon
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>
> Support batch writes in GraphHttpPersistWriter.
> Using a separate HTTP Post for every datum is inefficient.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809813#comment-15809813
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085437
  
--- Diff: 
streams-util/src/main/java/org/apache/streams/util/PropertyUtil.java ---
@@ -98,22 +102,24 @@ public static ObjectNode 
unflattenObjectNode(ObjectNode flatObject, char seperat
 while (iter.hasNext()) {
   Map.Entry item = iter.next();
   String fullKey = item.getKey();
-  if ( !fullKey.contains(Character.valueOf(seperator).toString())) {
+  if( !fullKey.contains(Character.valueOf(seperator).toString())) {
 root.put(item.getKey(), item.getValue());
   } else {
 ObjectNode currentNode = root;
-List keyParts = new ArrayList<>();
+List keyParts = Lists.newArrayList();
--- End diff --

✔️ removed guava from this class



> Support binary protocol in streams-persist-graph
> 
>
> Key: STREAMS-344
> URL: https://issues.apache.org/jira/browse/STREAMS-344
> Project: Streams
>  Issue Type: Improvement
>  Components: Persist
>Reporter: Steve Blackmon
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>
> Support batch writes in GraphHttpPersistWriter.
> Using a separate HTTP Post for every datum is inefficient.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[steveblackmon]

Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085434
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/http/Neo4jHttpClient.java
 ---
@@ -0,0 +1,84 @@
+package org.apache.streams.neo4j.http;
+
+import org.apache.streams.neo4j.Neo4jConfiguration;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.http.client.HttpClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class Neo4jHttpClient {
+
+private static final Logger LOGGER = LoggerFactory
+.getLogger(Neo4jHttpClient.class);
+
+public Neo4jConfiguration config;
+
+private HttpClient client;
+
+private Neo4jHttpClient(Neo4jConfiguration neo4jConfiguration) {
+this.config = neo4jConfiguration;
+try {
+this.start();
+} catch (Exception e) {
+e.printStackTrace();
+this.client = null;
+}
+}
+
+private static Map INSTANCE_MAP = 
new ConcurrentHashMap();
+
+public static Neo4jHttpClient getInstance(Neo4jConfiguration 
neo4jConfiguration) {
+if (INSTANCE_MAP != null &&
+INSTANCE_MAP.size() > 0 &&
+INSTANCE_MAP.containsKey(neo4jConfiguration)
+)
+return INSTANCE_MAP.get(neo4jConfiguration);
+else {
+Neo4jHttpClient instance = new 
Neo4jHttpClient(neo4jConfiguration);
+if( instance != null && instance.client != null ) {
+INSTANCE_MAP.put(neo4jConfiguration, instance);
+return instance;
+} else {
+return null;
+}
+}
+}
+
+public void start() throws Exception {
+
+Preconditions.checkNotNull(config);
+Preconditions.checkArgument(
+config.getScheme().startsWith("http")
+);
+
+LOGGER.info("Neo4jConfiguration.start {}", config);
+
+Preconditions.checkNotNull(client);
--- End diff --

✔️ 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[steveblackmon]

Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085431
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/bolt/Neo4jBoltClient.java
 ---
@@ -0,0 +1,100 @@
+package org.apache.streams.neo4j.bolt;
+
+import org.apache.streams.neo4j.Neo4jConfiguration;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.commons.lang3.StringUtils;
+import org.neo4j.driver.v1.AuthToken;
+import org.neo4j.driver.v1.AuthTokens;
+import org.neo4j.driver.v1.Driver;
+import org.neo4j.driver.v1.GraphDatabase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class Neo4jBoltClient {
+
+private static final Logger LOGGER = LoggerFactory
+.getLogger(Neo4jBoltClient.class);
+
+private Driver client;
+
+public Neo4jConfiguration config;
+
+private Neo4jBoltClient(Neo4jConfiguration neo4jConfiguration) {
+this.config = neo4jConfiguration;
+try {
+this.start();
+} catch (Exception e) {
+e.printStackTrace();
+this.client = null;
+}
+}
+
+private static Map INSTANCE_MAP = 
new ConcurrentHashMap();
+
+public static Neo4jBoltClient getInstance(Neo4jConfiguration 
neo4jConfiguration) {
+if (INSTANCE_MAP != null &&
--- End diff --

✔️ 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[steveblackmon]

Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085437
  
--- Diff: 
streams-util/src/main/java/org/apache/streams/util/PropertyUtil.java ---
@@ -98,22 +102,24 @@ public static ObjectNode 
unflattenObjectNode(ObjectNode flatObject, char seperat
 while (iter.hasNext()) {
   Map.Entry item = iter.next();
   String fullKey = item.getKey();
-  if ( !fullKey.contains(Character.valueOf(seperator).toString())) {
+  if( !fullKey.contains(Character.valueOf(seperator).toString())) {
 root.put(item.getKey(), item.getValue());
   } else {
 ObjectNode currentNode = root;
-List keyParts = new ArrayList<>();
+List keyParts = Lists.newArrayList();
--- End diff --

✔️ removed guava from this class



---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809807#comment-15809807
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085389
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/Neo4jPersistUtil.java
 ---
@@ -0,0 +1,151 @@
+package org.apache.streams.neo4j;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.neo4j.bolt.Neo4jBoltPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+import org.apache.commons.lang3.StringUtils;
+import org.javatuples.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by steve on 1/2/17.
+ */
+public class Neo4jPersistUtil {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(Neo4jBoltPersistWriter.class);
+
+  private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private static CypherQueryGraphHelper helper = new 
CypherQueryGraphHelper();
+
+  public static List>> 
prepareStatements(StreamsDatum entry) throws Exception {
+
+List>> statements = new ArrayList<>();
+
+String id = entry.getId();
+Activity activity = null;
+ActivityObject activityObject = null;
+Object document = entry.getDocument();
+
+if (document instanceof Activity) {
+  activity = (Activity) document;
+} else if (document instanceof ActivityObject) {
+  activityObject = (ActivityObject) document;
+} else {
+  ObjectNode objectNode;
+  if (document instanceof ObjectNode) {
+objectNode = (ObjectNode) document;
+  } else if ( document instanceof String) {
+try {
+  objectNode = mapper.readValue((String) document, 
ObjectNode.class);
+} catch (IOException ex) {
+  LOGGER.error("Can't handle input: ", entry);
+  throw ex;
+}
+  } else {
+LOGGER.error("Can't handle input: ", entry);
+throw new Exception("Can't create statements from datum.");
+  }
+
+  if ( objectNode.get("verb") != null ) {
+try {
+  activity = mapper.convertValue(objectNode, Activity.class);
+  activityObject = activity.getObject();
+} catch (Exception ex) {
+  activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+}
+  } else {
+activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+  }
+
+}
+
+Preconditions.checkArgument(activity != null ^ activityObject != null);
+
+if ( activityObject != null && 
!Strings.isNullOrEmpty(activityObject.getId())) {
+
+  statements.add(vertexStatement(activityObject));
+
+} else if ( activity != null && 
!Strings.isNullOrEmpty(activity.getId())) {
+
+  statements.addAll(vertexStatements(activity));
+
+  statements.addAll(edgeStatements(activity));
+
+}
+
+return statements;
+  }
+
+  public static List>> 
vertexStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+ActivityObject actor = activity.getActor();
+ActivityObject object = activity.getObject();
+ActivityObject target = activity.getTarget();
+
+if (actor != null && StringUtils.isNotBlank(actor.getId())) {
+  Pair> actorStatement = 
vertexStatement(actor);
+  statements.add(actorStatement);
+}
+
+if (object != null && StringUtils.isNotBlank(object.getId())) {
+  Pair> objectStatement = 
vertexStatement(object);
+  statements.add(objectStatement);
+}
+
+if (target != null && StringUtils.isNotBlank(target.getId())) {
+  Pair> targetStatement = 
vertexStatement(target);
+  statements.add(targetStatement);
+

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[steveblackmon]

Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085386
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/Neo4jPersistUtil.java
 ---
@@ -0,0 +1,151 @@
+package org.apache.streams.neo4j;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.neo4j.bolt.Neo4jBoltPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+import org.apache.commons.lang3.StringUtils;
+import org.javatuples.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by steve on 1/2/17.
+ */
+public class Neo4jPersistUtil {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(Neo4jBoltPersistWriter.class);
+
+  private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private static CypherQueryGraphHelper helper = new 
CypherQueryGraphHelper();
+
+  public static List>> 
prepareStatements(StreamsDatum entry) throws Exception {
+
+List>> statements = new ArrayList<>();
+
+String id = entry.getId();
+Activity activity = null;
+ActivityObject activityObject = null;
+Object document = entry.getDocument();
+
+if (document instanceof Activity) {
+  activity = (Activity) document;
+} else if (document instanceof ActivityObject) {
+  activityObject = (ActivityObject) document;
+} else {
+  ObjectNode objectNode;
+  if (document instanceof ObjectNode) {
+objectNode = (ObjectNode) document;
+  } else if ( document instanceof String) {
+try {
+  objectNode = mapper.readValue((String) document, 
ObjectNode.class);
+} catch (IOException ex) {
+  LOGGER.error("Can't handle input: ", entry);
+  throw ex;
+}
+  } else {
+LOGGER.error("Can't handle input: ", entry);
+throw new Exception("Can't create statements from datum.");
+  }
+
+  if ( objectNode.get("verb") != null ) {
+try {
+  activity = mapper.convertValue(objectNode, Activity.class);
+  activityObject = activity.getObject();
+} catch (Exception ex) {
+  activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+}
+  } else {
+activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+  }
+
+}
+
+Preconditions.checkArgument(activity != null ^ activityObject != null);
+
+if ( activityObject != null && 
!Strings.isNullOrEmpty(activityObject.getId())) {
+
+  statements.add(vertexStatement(activityObject));
+
+} else if ( activity != null && 
!Strings.isNullOrEmpty(activity.getId())) {
+
+  statements.addAll(vertexStatements(activity));
+
+  statements.addAll(edgeStatements(activity));
+
+}
+
+return statements;
+  }
+
+  public static List>> 
vertexStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+ActivityObject actor = activity.getActor();
+ActivityObject object = activity.getObject();
+ActivityObject target = activity.getTarget();
+
+if (actor != null && StringUtils.isNotBlank(actor.getId())) {
+  Pair> actorStatement = 
vertexStatement(actor);
+  statements.add(actorStatement);
+}
+
+if (object != null && StringUtils.isNotBlank(object.getId())) {
+  Pair> objectStatement = 
vertexStatement(object);
+  statements.add(objectStatement);
+}
+
+if (target != null && StringUtils.isNotBlank(target.getId())) {
+  Pair> targetStatement = 
vertexStatement(target);
+  statements.add(targetStatement);
+}
+
+return statements;
+  }
+
+  public static List>> 
edgeStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809806#comment-15809806
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085386
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/Neo4jPersistUtil.java
 ---
@@ -0,0 +1,151 @@
+package org.apache.streams.neo4j;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.neo4j.bolt.Neo4jBoltPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+import org.apache.commons.lang3.StringUtils;
+import org.javatuples.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by steve on 1/2/17.
+ */
+public class Neo4jPersistUtil {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(Neo4jBoltPersistWriter.class);
+
+  private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private static CypherQueryGraphHelper helper = new 
CypherQueryGraphHelper();
+
+  public static List>> 
prepareStatements(StreamsDatum entry) throws Exception {
+
+List>> statements = new ArrayList<>();
+
+String id = entry.getId();
+Activity activity = null;
+ActivityObject activityObject = null;
+Object document = entry.getDocument();
+
+if (document instanceof Activity) {
+  activity = (Activity) document;
+} else if (document instanceof ActivityObject) {
+  activityObject = (ActivityObject) document;
+} else {
+  ObjectNode objectNode;
+  if (document instanceof ObjectNode) {
+objectNode = (ObjectNode) document;
+  } else if ( document instanceof String) {
+try {
+  objectNode = mapper.readValue((String) document, 
ObjectNode.class);
+} catch (IOException ex) {
+  LOGGER.error("Can't handle input: ", entry);
+  throw ex;
+}
+  } else {
+LOGGER.error("Can't handle input: ", entry);
+throw new Exception("Can't create statements from datum.");
+  }
+
+  if ( objectNode.get("verb") != null ) {
+try {
+  activity = mapper.convertValue(objectNode, Activity.class);
+  activityObject = activity.getObject();
+} catch (Exception ex) {
+  activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+}
+  } else {
+activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+  }
+
+}
+
+Preconditions.checkArgument(activity != null ^ activityObject != null);
+
+if ( activityObject != null && 
!Strings.isNullOrEmpty(activityObject.getId())) {
+
+  statements.add(vertexStatement(activityObject));
+
+} else if ( activity != null && 
!Strings.isNullOrEmpty(activity.getId())) {
+
+  statements.addAll(vertexStatements(activity));
+
+  statements.addAll(edgeStatements(activity));
+
+}
+
+return statements;
+  }
+
+  public static List>> 
vertexStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+ActivityObject actor = activity.getActor();
+ActivityObject object = activity.getObject();
+ActivityObject target = activity.getTarget();
+
+if (actor != null && StringUtils.isNotBlank(actor.getId())) {
+  Pair> actorStatement = 
vertexStatement(actor);
+  statements.add(actorStatement);
+}
+
+if (object != null && StringUtils.isNotBlank(object.getId())) {
+  Pair> objectStatement = 
vertexStatement(object);
+  statements.add(objectStatement);
+}
+
+if (target != null && StringUtils.isNotBlank(target.getId())) {
+  Pair> targetStatement = 
vertexStatement(target);
+  statements.add(targetStatement);
+

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[steveblackmon]

Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95085389
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/Neo4jPersistUtil.java
 ---
@@ -0,0 +1,151 @@
+package org.apache.streams.neo4j;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.neo4j.bolt.Neo4jBoltPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+import org.apache.commons.lang3.StringUtils;
+import org.javatuples.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by steve on 1/2/17.
+ */
+public class Neo4jPersistUtil {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(Neo4jBoltPersistWriter.class);
+
+  private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private static CypherQueryGraphHelper helper = new 
CypherQueryGraphHelper();
+
+  public static List>> 
prepareStatements(StreamsDatum entry) throws Exception {
+
+List>> statements = new ArrayList<>();
+
+String id = entry.getId();
+Activity activity = null;
+ActivityObject activityObject = null;
+Object document = entry.getDocument();
+
+if (document instanceof Activity) {
+  activity = (Activity) document;
+} else if (document instanceof ActivityObject) {
+  activityObject = (ActivityObject) document;
+} else {
+  ObjectNode objectNode;
+  if (document instanceof ObjectNode) {
+objectNode = (ObjectNode) document;
+  } else if ( document instanceof String) {
+try {
+  objectNode = mapper.readValue((String) document, 
ObjectNode.class);
+} catch (IOException ex) {
+  LOGGER.error("Can't handle input: ", entry);
+  throw ex;
+}
+  } else {
+LOGGER.error("Can't handle input: ", entry);
+throw new Exception("Can't create statements from datum.");
+  }
+
+  if ( objectNode.get("verb") != null ) {
+try {
+  activity = mapper.convertValue(objectNode, Activity.class);
+  activityObject = activity.getObject();
+} catch (Exception ex) {
+  activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+}
+  } else {
+activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+  }
+
+}
+
+Preconditions.checkArgument(activity != null ^ activityObject != null);
+
+if ( activityObject != null && 
!Strings.isNullOrEmpty(activityObject.getId())) {
+
+  statements.add(vertexStatement(activityObject));
+
+} else if ( activity != null && 
!Strings.isNullOrEmpty(activity.getId())) {
+
+  statements.addAll(vertexStatements(activity));
+
+  statements.addAll(edgeStatements(activity));
+
+}
+
+return statements;
+  }
+
+  public static List>> 
vertexStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+ActivityObject actor = activity.getActor();
+ActivityObject object = activity.getObject();
+ActivityObject target = activity.getTarget();
+
+if (actor != null && StringUtils.isNotBlank(actor.getId())) {
+  Pair> actorStatement = 
vertexStatement(actor);
+  statements.add(actorStatement);
+}
+
+if (object != null && StringUtils.isNotBlank(object.getId())) {
+  Pair> objectStatement = 
vertexStatement(object);
+  statements.add(objectStatement);
+}
+
+if (target != null && StringUtils.isNotBlank(target.getId())) {
+  Pair> targetStatement = 
vertexStatement(target);
+  statements.add(targetStatement);
+}
+
+return statements;
+  }
+
+  public static List>> 
edgeStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+

[jira] [Commented] (STREAMS-472) Fix failing IT in streams-plugins

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-472?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809586#comment-15809586
 ] 

ASF GitHub Bot commented on STREAMS-472:


Github user smarthi commented on the issue:

https://github.com/apache/incubator-streams/pull/349
  
+1 lgtm


> Fix failing IT in streams-plugins
> -
>
> Key: STREAMS-472
> URL: https://issues.apache.org/jira/browse/STREAMS-472
> Project: Streams
>  Issue Type: Bug
>Affects Versions: 0.4.1
>Reporter: Steve Blackmon
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>   Original Estimate: 4h
>  Remaining Estimate: 4h
>
> Appears that these test poms have an outdated version dependency, 
> 0.4-incubating-SNAPSHOT instead of 0.5-incubating-SNAPSHO



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-streams issue #349: STREAMS-472: fixes root cause

[smarthi]

Github user smarthi commented on the issue:

https://github.com/apache/incubator-streams/pull/349
  
+1 lgtm


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809582#comment-15809582
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082300
  
--- Diff: 
streams-util/src/main/java/org/apache/streams/util/PropertyUtil.java ---
@@ -98,22 +102,24 @@ public static ObjectNode 
unflattenObjectNode(ObjectNode flatObject, char seperat
 while (iter.hasNext()) {
   Map.Entry item = iter.next();
   String fullKey = item.getKey();
-  if ( !fullKey.contains(Character.valueOf(seperator).toString())) {
+  if( !fullKey.contains(Character.valueOf(seperator).toString())) {
 root.put(item.getKey(), item.getValue());
   } else {
 ObjectNode currentNode = root;
-List keyParts = new ArrayList<>();
+List keyParts = Lists.newArrayList();
--- End diff --

any reason for switching back to using Guava here, don't see the benefit


> Support binary protocol in streams-persist-graph
> 
>
> Key: STREAMS-344
> URL: https://issues.apache.org/jira/browse/STREAMS-344
> Project: Streams
>  Issue Type: Improvement
>  Components: Persist
>Reporter: Steve Blackmon
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>
> Support batch writes in GraphHttpPersistWriter.
> Using a separate HTTP Post for every datum is inefficient.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[smarthi]

Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082300
  
--- Diff: 
streams-util/src/main/java/org/apache/streams/util/PropertyUtil.java ---
@@ -98,22 +102,24 @@ public static ObjectNode 
unflattenObjectNode(ObjectNode flatObject, char seperat
 while (iter.hasNext()) {
   Map.Entry item = iter.next();
   String fullKey = item.getKey();
-  if ( !fullKey.contains(Character.valueOf(seperator).toString())) {
+  if( !fullKey.contains(Character.valueOf(seperator).toString())) {
 root.put(item.getKey(), item.getValue());
   } else {
 ObjectNode currentNode = root;
-List keyParts = new ArrayList<>();
+List keyParts = Lists.newArrayList();
--- End diff --

any reason for switching back to using Guava here, don't see the benefit


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[smarthi]

Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082122
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/http/Neo4jHttpClient.java
 ---
@@ -0,0 +1,84 @@
+package org.apache.streams.neo4j.http;
+
+import org.apache.streams.neo4j.Neo4jConfiguration;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.http.client.HttpClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class Neo4jHttpClient {
+
+private static final Logger LOGGER = LoggerFactory
+.getLogger(Neo4jHttpClient.class);
+
+public Neo4jConfiguration config;
+
+private HttpClient client;
+
+private Neo4jHttpClient(Neo4jConfiguration neo4jConfiguration) {
+this.config = neo4jConfiguration;
+try {
+this.start();
+} catch (Exception e) {
+e.printStackTrace();
+this.client = null;
+}
+}
+
+private static Map INSTANCE_MAP = 
new ConcurrentHashMap();
+
+public static Neo4jHttpClient getInstance(Neo4jConfiguration 
neo4jConfiguration) {
+if (INSTANCE_MAP != null &&
+INSTANCE_MAP.size() > 0 &&
+INSTANCE_MAP.containsKey(neo4jConfiguration)
+)
+return INSTANCE_MAP.get(neo4jConfiguration);
+else {
+Neo4jHttpClient instance = new 
Neo4jHttpClient(neo4jConfiguration);
+if( instance != null && instance.client != null ) {
+INSTANCE_MAP.put(neo4jConfiguration, instance);
+return instance;
+} else {
+return null;
+}
+}
+}
+
+public void start() throws Exception {
+
+Preconditions.checkNotNull(config);
+Preconditions.checkArgument(
+config.getScheme().startsWith("http")
+);
+
+LOGGER.info("Neo4jConfiguration.start {}", config);
+
+Preconditions.checkNotNull(client);
--- End diff --

Replace by Objects.requireNotNull()


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809574#comment-15809574
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082122
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/http/Neo4jHttpClient.java
 ---
@@ -0,0 +1,84 @@
+package org.apache.streams.neo4j.http;
+
+import org.apache.streams.neo4j.Neo4jConfiguration;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.http.client.HttpClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class Neo4jHttpClient {
+
+private static final Logger LOGGER = LoggerFactory
+.getLogger(Neo4jHttpClient.class);
+
+public Neo4jConfiguration config;
+
+private HttpClient client;
+
+private Neo4jHttpClient(Neo4jConfiguration neo4jConfiguration) {
+this.config = neo4jConfiguration;
+try {
+this.start();
+} catch (Exception e) {
+e.printStackTrace();
+this.client = null;
+}
+}
+
+private static Map INSTANCE_MAP = 
new ConcurrentHashMap();
+
+public static Neo4jHttpClient getInstance(Neo4jConfiguration 
neo4jConfiguration) {
+if (INSTANCE_MAP != null &&
+INSTANCE_MAP.size() > 0 &&
+INSTANCE_MAP.containsKey(neo4jConfiguration)
+)
+return INSTANCE_MAP.get(neo4jConfiguration);
+else {
+Neo4jHttpClient instance = new 
Neo4jHttpClient(neo4jConfiguration);
+if( instance != null && instance.client != null ) {
+INSTANCE_MAP.put(neo4jConfiguration, instance);
+return instance;
+} else {
+return null;
+}
+}
+}
+
+public void start() throws Exception {
+
+Preconditions.checkNotNull(config);
+Preconditions.checkArgument(
+config.getScheme().startsWith("http")
+);
+
+LOGGER.info("Neo4jConfiguration.start {}", config);
+
+Preconditions.checkNotNull(client);
--- End diff --

Replace by Objects.requireNotNull()


> Support binary protocol in streams-persist-graph
> 
>
> Key: STREAMS-344
> URL: https://issues.apache.org/jira/browse/STREAMS-344
> Project: Streams
>  Issue Type: Improvement
>  Components: Persist
>Reporter: Steve Blackmon
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>
> Support batch writes in GraphHttpPersistWriter.
> Using a separate HTTP Post for every datum is inefficient.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809570#comment-15809570
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082081
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/Neo4jPersistUtil.java
 ---
@@ -0,0 +1,151 @@
+package org.apache.streams.neo4j;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.neo4j.bolt.Neo4jBoltPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+import org.apache.commons.lang3.StringUtils;
+import org.javatuples.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by steve on 1/2/17.
+ */
+public class Neo4jPersistUtil {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(Neo4jBoltPersistWriter.class);
+
+  private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private static CypherQueryGraphHelper helper = new 
CypherQueryGraphHelper();
+
+  public static List>> 
prepareStatements(StreamsDatum entry) throws Exception {
+
+List>> statements = new ArrayList<>();
+
+String id = entry.getId();
+Activity activity = null;
+ActivityObject activityObject = null;
+Object document = entry.getDocument();
+
+if (document instanceof Activity) {
+  activity = (Activity) document;
+} else if (document instanceof ActivityObject) {
+  activityObject = (ActivityObject) document;
+} else {
+  ObjectNode objectNode;
+  if (document instanceof ObjectNode) {
+objectNode = (ObjectNode) document;
+  } else if ( document instanceof String) {
+try {
+  objectNode = mapper.readValue((String) document, 
ObjectNode.class);
+} catch (IOException ex) {
+  LOGGER.error("Can't handle input: ", entry);
+  throw ex;
+}
+  } else {
+LOGGER.error("Can't handle input: ", entry);
+throw new Exception("Can't create statements from datum.");
+  }
+
+  if ( objectNode.get("verb") != null ) {
+try {
+  activity = mapper.convertValue(objectNode, Activity.class);
+  activityObject = activity.getObject();
+} catch (Exception ex) {
+  activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+}
+  } else {
+activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+  }
+
+}
+
+Preconditions.checkArgument(activity != null ^ activityObject != null);
+
+if ( activityObject != null && 
!Strings.isNullOrEmpty(activityObject.getId())) {
+
+  statements.add(vertexStatement(activityObject));
+
+} else if ( activity != null && 
!Strings.isNullOrEmpty(activity.getId())) {
+
+  statements.addAll(vertexStatements(activity));
+
+  statements.addAll(edgeStatements(activity));
+
+}
+
+return statements;
+  }
+
+  public static List>> 
vertexStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+ActivityObject actor = activity.getActor();
+ActivityObject object = activity.getObject();
+ActivityObject target = activity.getTarget();
+
+if (actor != null && StringUtils.isNotBlank(actor.getId())) {
+  Pair> actorStatement = 
vertexStatement(actor);
+  statements.add(actorStatement);
+}
+
+if (object != null && StringUtils.isNotBlank(object.getId())) {
+  Pair> objectStatement = 
vertexStatement(object);
+  statements.add(objectStatement);
+}
+
+if (target != null && StringUtils.isNotBlank(target.getId())) {
+  Pair> targetStatement = 
vertexStatement(target);
+  statements.add(targetStatement);
+}

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809569#comment-15809569
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082073
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/Neo4jPersistUtil.java
 ---
@@ -0,0 +1,151 @@
+package org.apache.streams.neo4j;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.neo4j.bolt.Neo4jBoltPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+import org.apache.commons.lang3.StringUtils;
+import org.javatuples.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by steve on 1/2/17.
+ */
+public class Neo4jPersistUtil {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(Neo4jBoltPersistWriter.class);
+
+  private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private static CypherQueryGraphHelper helper = new 
CypherQueryGraphHelper();
+
+  public static List>> 
prepareStatements(StreamsDatum entry) throws Exception {
+
+List>> statements = new ArrayList<>();
+
+String id = entry.getId();
+Activity activity = null;
+ActivityObject activityObject = null;
+Object document = entry.getDocument();
+
+if (document instanceof Activity) {
+  activity = (Activity) document;
+} else if (document instanceof ActivityObject) {
+  activityObject = (ActivityObject) document;
+} else {
+  ObjectNode objectNode;
+  if (document instanceof ObjectNode) {
+objectNode = (ObjectNode) document;
+  } else if ( document instanceof String) {
+try {
+  objectNode = mapper.readValue((String) document, 
ObjectNode.class);
+} catch (IOException ex) {
+  LOGGER.error("Can't handle input: ", entry);
+  throw ex;
+}
+  } else {
+LOGGER.error("Can't handle input: ", entry);
+throw new Exception("Can't create statements from datum.");
+  }
+
+  if ( objectNode.get("verb") != null ) {
+try {
+  activity = mapper.convertValue(objectNode, Activity.class);
+  activityObject = activity.getObject();
+} catch (Exception ex) {
+  activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+}
+  } else {
+activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+  }
+
+}
+
+Preconditions.checkArgument(activity != null ^ activityObject != null);
+
+if ( activityObject != null && 
!Strings.isNullOrEmpty(activityObject.getId())) {
+
+  statements.add(vertexStatement(activityObject));
+
+} else if ( activity != null && 
!Strings.isNullOrEmpty(activity.getId())) {
+
+  statements.addAll(vertexStatements(activity));
+
+  statements.addAll(edgeStatements(activity));
+
+}
+
+return statements;
+  }
+
+  public static List>> 
vertexStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+ActivityObject actor = activity.getActor();
+ActivityObject object = activity.getObject();
+ActivityObject target = activity.getTarget();
+
+if (actor != null && StringUtils.isNotBlank(actor.getId())) {
+  Pair> actorStatement = 
vertexStatement(actor);
+  statements.add(actorStatement);
+}
+
+if (object != null && StringUtils.isNotBlank(object.getId())) {
+  Pair> objectStatement = 
vertexStatement(object);
+  statements.add(objectStatement);
+}
+
+if (target != null && StringUtils.isNotBlank(target.getId())) {
+  Pair> targetStatement = 
vertexStatement(target);
+  statements.add(targetStatement);
+}

[jira] [Commented] (STREAMS-344) Support binary protocol in streams-persist-graph

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/STREAMS-344?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15809571#comment-15809571
 ] 

ASF GitHub Bot commented on STREAMS-344:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082088
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/bolt/Neo4jBoltClient.java
 ---
@@ -0,0 +1,100 @@
+package org.apache.streams.neo4j.bolt;
+
+import org.apache.streams.neo4j.Neo4jConfiguration;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.commons.lang3.StringUtils;
+import org.neo4j.driver.v1.AuthToken;
+import org.neo4j.driver.v1.AuthTokens;
+import org.neo4j.driver.v1.Driver;
+import org.neo4j.driver.v1.GraphDatabase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class Neo4jBoltClient {
+
+private static final Logger LOGGER = LoggerFactory
+.getLogger(Neo4jBoltClient.class);
+
+private Driver client;
+
+public Neo4jConfiguration config;
+
+private Neo4jBoltClient(Neo4jConfiguration neo4jConfiguration) {
+this.config = neo4jConfiguration;
+try {
+this.start();
+} catch (Exception e) {
+e.printStackTrace();
+this.client = null;
+}
+}
+
+private static Map INSTANCE_MAP = 
new ConcurrentHashMap();
+
+public static Neo4jBoltClient getInstance(Neo4jConfiguration 
neo4jConfiguration) {
+if (INSTANCE_MAP != null &&
--- End diff --

extra space after openning brace - style check nitpick


> Support binary protocol in streams-persist-graph
> 
>
> Key: STREAMS-344
> URL: https://issues.apache.org/jira/browse/STREAMS-344
> Project: Streams
>  Issue Type: Improvement
>  Components: Persist
>Reporter: Steve Blackmon
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>
> Support batch writes in GraphHttpPersistWriter.
> Using a separate HTTP Post for every datum is inefficient.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[smarthi]

Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082088
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/bolt/Neo4jBoltClient.java
 ---
@@ -0,0 +1,100 @@
+package org.apache.streams.neo4j.bolt;
+
+import org.apache.streams.neo4j.Neo4jConfiguration;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.commons.lang3.StringUtils;
+import org.neo4j.driver.v1.AuthToken;
+import org.neo4j.driver.v1.AuthTokens;
+import org.neo4j.driver.v1.Driver;
+import org.neo4j.driver.v1.GraphDatabase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class Neo4jBoltClient {
+
+private static final Logger LOGGER = LoggerFactory
+.getLogger(Neo4jBoltClient.class);
+
+private Driver client;
+
+public Neo4jConfiguration config;
+
+private Neo4jBoltClient(Neo4jConfiguration neo4jConfiguration) {
+this.config = neo4jConfiguration;
+try {
+this.start();
+} catch (Exception e) {
+e.printStackTrace();
+this.client = null;
+}
+}
+
+private static Map INSTANCE_MAP = 
new ConcurrentHashMap();
+
+public static Neo4jBoltClient getInstance(Neo4jConfiguration 
neo4jConfiguration) {
+if (INSTANCE_MAP != null &&
--- End diff --

extra space after openning brace - style check nitpick


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[smarthi]

Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082081
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/Neo4jPersistUtil.java
 ---
@@ -0,0 +1,151 @@
+package org.apache.streams.neo4j;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.neo4j.bolt.Neo4jBoltPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+import org.apache.commons.lang3.StringUtils;
+import org.javatuples.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by steve on 1/2/17.
+ */
+public class Neo4jPersistUtil {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(Neo4jBoltPersistWriter.class);
+
+  private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private static CypherQueryGraphHelper helper = new 
CypherQueryGraphHelper();
+
+  public static List>> 
prepareStatements(StreamsDatum entry) throws Exception {
+
+List>> statements = new ArrayList<>();
+
+String id = entry.getId();
+Activity activity = null;
+ActivityObject activityObject = null;
+Object document = entry.getDocument();
+
+if (document instanceof Activity) {
+  activity = (Activity) document;
+} else if (document instanceof ActivityObject) {
+  activityObject = (ActivityObject) document;
+} else {
+  ObjectNode objectNode;
+  if (document instanceof ObjectNode) {
+objectNode = (ObjectNode) document;
+  } else if ( document instanceof String) {
+try {
+  objectNode = mapper.readValue((String) document, 
ObjectNode.class);
+} catch (IOException ex) {
+  LOGGER.error("Can't handle input: ", entry);
+  throw ex;
+}
+  } else {
+LOGGER.error("Can't handle input: ", entry);
+throw new Exception("Can't create statements from datum.");
+  }
+
+  if ( objectNode.get("verb") != null ) {
+try {
+  activity = mapper.convertValue(objectNode, Activity.class);
+  activityObject = activity.getObject();
+} catch (Exception ex) {
+  activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+}
+  } else {
+activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+  }
+
+}
+
+Preconditions.checkArgument(activity != null ^ activityObject != null);
+
+if ( activityObject != null && 
!Strings.isNullOrEmpty(activityObject.getId())) {
+
+  statements.add(vertexStatement(activityObject));
+
+} else if ( activity != null && 
!Strings.isNullOrEmpty(activity.getId())) {
+
+  statements.addAll(vertexStatements(activity));
+
+  statements.addAll(edgeStatements(activity));
+
+}
+
+return statements;
+  }
+
+  public static List>> 
vertexStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+ActivityObject actor = activity.getActor();
+ActivityObject object = activity.getObject();
+ActivityObject target = activity.getTarget();
+
+if (actor != null && StringUtils.isNotBlank(actor.getId())) {
+  Pair> actorStatement = 
vertexStatement(actor);
+  statements.add(actorStatement);
+}
+
+if (object != null && StringUtils.isNotBlank(object.getId())) {
+  Pair> objectStatement = 
vertexStatement(object);
+  statements.add(objectStatement);
+}
+
+if (target != null && StringUtils.isNotBlank(target.getId())) {
+  Pair> targetStatement = 
vertexStatement(target);
+  statements.add(targetStatement);
+}
+
+return statements;
+  }
+
+  public static List>> 
edgeStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+

[GitHub] incubator-streams pull request #348: STREAMS-344: streams-persist-neo4j

[smarthi]

Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/348#discussion_r95082073
  
--- Diff: 
streams-contrib/streams-persist-neo4j/src/main/java/org/apache/streams/neo4j/Neo4jPersistUtil.java
 ---
@@ -0,0 +1,151 @@
+package org.apache.streams.neo4j;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.neo4j.bolt.Neo4jBoltPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+import org.apache.commons.lang3.StringUtils;
+import org.javatuples.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by steve on 1/2/17.
+ */
+public class Neo4jPersistUtil {
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(Neo4jBoltPersistWriter.class);
+
+  private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private static CypherQueryGraphHelper helper = new 
CypherQueryGraphHelper();
+
+  public static List>> 
prepareStatements(StreamsDatum entry) throws Exception {
+
+List>> statements = new ArrayList<>();
+
+String id = entry.getId();
+Activity activity = null;
+ActivityObject activityObject = null;
+Object document = entry.getDocument();
+
+if (document instanceof Activity) {
+  activity = (Activity) document;
+} else if (document instanceof ActivityObject) {
+  activityObject = (ActivityObject) document;
+} else {
+  ObjectNode objectNode;
+  if (document instanceof ObjectNode) {
+objectNode = (ObjectNode) document;
+  } else if ( document instanceof String) {
+try {
+  objectNode = mapper.readValue((String) document, 
ObjectNode.class);
+} catch (IOException ex) {
+  LOGGER.error("Can't handle input: ", entry);
+  throw ex;
+}
+  } else {
+LOGGER.error("Can't handle input: ", entry);
+throw new Exception("Can't create statements from datum.");
+  }
+
+  if ( objectNode.get("verb") != null ) {
+try {
+  activity = mapper.convertValue(objectNode, Activity.class);
+  activityObject = activity.getObject();
+} catch (Exception ex) {
+  activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+}
+  } else {
+activityObject = mapper.convertValue(objectNode, 
ActivityObject.class);
+  }
+
+}
+
+Preconditions.checkArgument(activity != null ^ activityObject != null);
+
+if ( activityObject != null && 
!Strings.isNullOrEmpty(activityObject.getId())) {
+
+  statements.add(vertexStatement(activityObject));
+
+} else if ( activity != null && 
!Strings.isNullOrEmpty(activity.getId())) {
+
+  statements.addAll(vertexStatements(activity));
+
+  statements.addAll(edgeStatements(activity));
+
+}
+
+return statements;
+  }
+
+  public static List>> 
vertexStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+ActivityObject actor = activity.getActor();
+ActivityObject object = activity.getObject();
+ActivityObject target = activity.getTarget();
+
+if (actor != null && StringUtils.isNotBlank(actor.getId())) {
+  Pair> actorStatement = 
vertexStatement(actor);
+  statements.add(actorStatement);
+}
+
+if (object != null && StringUtils.isNotBlank(object.getId())) {
+  Pair> objectStatement = 
vertexStatement(object);
+  statements.add(objectStatement);
+}
+
+if (target != null && StringUtils.isNotBlank(target.getId())) {
+  Pair> targetStatement = 
vertexStatement(target);
+  statements.add(targetStatement);
+}
+
+return statements;
+  }
+
+  public static List>> 
edgeStatements(Activity activity) {
+List>> statements = new 
ArrayList<>();;
+