You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2014/05/20 00:37:04 UTC
[3/3] git commit: updated per pull request feedback
updated per pull request feedback
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/65157bd4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/65157bd4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/65157bd4
Branch: refs/heads/STREAMS-70
Commit: 65157bd4cd642e4fa0b8e2f7236f9dd6cfd73b84
Parents: f24f251
Author: sblackmon <sb...@w2odigital.com>
Authored: Mon May 19 14:49:51 2014 -0500
Committer: sblackmon <sb...@w2odigital.com>
Committed: Mon May 19 14:49:51 2014 -0500
----------------------------------------------------------------------
streams-contrib/streams-processor-json/pom.xml | 17 +++++++
.../apache/streams/json/JsonPathExtractor.java | 49 ++++++++++++++------
.../org/apache/streams/json/JsonPathFilter.java | 40 ++++++++++++----
.../json/test/JsonPathExtractorTest.java | 12 ++---
4 files changed, 86 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/65157bd4/streams-contrib/streams-processor-json/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/pom.xml b/streams-contrib/streams-processor-json/pom.xml
index 27695bc..ad22158 100644
--- a/streams-contrib/streams-processor-json/pom.xml
+++ b/streams-contrib/streams-processor-json/pom.xml
@@ -1,4 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/65157bd4/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
index eab1da1..c4920d8 100644
--- a/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
+++ b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
@@ -1,5 +1,24 @@
package org.apache.streams.json;
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -19,19 +38,11 @@ import java.util.Iterator;
import java.util.List;
/**
- * Created by sblackmon on 12/10/13.
+ * Provides a base implementation for extracting json fields and
+ * objects from datums using JsonPath syntax
*/
public class JsonPathExtractor implements StreamsProcessor {
- public JsonPathExtractor() {
- System.out.println("creating JsonPathExtractor");
- }
-
- public JsonPathExtractor(String pathExpression) {
- this.pathExpression = pathExpression;
- System.out.println("creating JsonPathExtractor for " + this.pathExpression);
- }
-
private final static String STREAMS_ID = "JsonPathExtractor";
private final static Logger LOGGER = LoggerFactory.getLogger(JsonPathExtractor.class);
@@ -41,6 +52,15 @@ public class JsonPathExtractor implements StreamsProcessor {
private String pathExpression;
private JsonPath jsonPath;
+ public JsonPathExtractor() {
+ LOGGER.info("creating JsonPathExtractor");
+ }
+
+ public JsonPathExtractor(String pathExpression) {
+ this.pathExpression = pathExpression;
+ LOGGER.info("creating JsonPathExtractor for " + this.pathExpression);
+ }
+
@Override
public List<StreamsDatum> process(StreamsDatum entry) {
@@ -55,7 +75,7 @@ public class JsonPathExtractor implements StreamsProcessor {
try {
json = mapper.writeValueAsString(node);
} catch (JsonProcessingException e) {
- e.printStackTrace();
+ LOGGER.warn(e.getMessage());
}
} else if( entry.getDocument() instanceof String ) {
json = (String) entry.getDocument();
@@ -85,7 +105,9 @@ public class JsonPathExtractor implements StreamsProcessor {
StreamsDatum matchDatum = new StreamsDatum(match);
result.add(matchDatum);
} else if ( item instanceof JSONObject ) {
- StreamsDatum matchDatum = new StreamsDatum(item);
+ JSONObject match = (JSONObject) item;
+ ObjectNode objectNode = mapper.readValue(mapper.writeValueAsString(match), ObjectNode.class);
+ StreamsDatum matchDatum = new StreamsDatum(objectNode);
result.add(matchDatum);
}
}
@@ -94,7 +116,6 @@ public class JsonPathExtractor implements StreamsProcessor {
}
} catch( Exception e ) {
- e.printStackTrace();
LOGGER.warn(e.getMessage());
}
@@ -114,6 +135,6 @@ public class JsonPathExtractor implements StreamsProcessor {
@Override
public void cleanUp() {
-
+ LOGGER.info("shutting down JsonPathExtractor for " + this.pathExpression);
}
};
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/65157bd4/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java
index 929b95b..820e459 100644
--- a/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java
+++ b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java
@@ -1,5 +1,24 @@
package org.apache.streams.json;
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -23,14 +42,11 @@ import java.util.List;
import java.util.Map;
/**
- * Created by sblackmon on 12/10/13.
+ * Provides a base implementation for filtering datums which
+ * do not contain specific fields using JsonPath syntax
*/
public class JsonPathFilter implements StreamsProcessor {
- public JsonPathFilter() {
- System.out.println("creating JsonPathFilter");
- }
-
private final static String STREAMS_ID = "JsonPathFilter";
private final static Logger LOGGER = LoggerFactory.getLogger(JsonPathFilter.class);
@@ -41,6 +57,15 @@ public class JsonPathFilter implements StreamsProcessor {
private JsonPath jsonPath;
private String destNodeName;
+ public JsonPathFilter() {
+ LOGGER.info("creating JsonPathFilter");
+ }
+
+ public JsonPathFilter(String pathExpression) {
+ this.pathExpression = pathExpression;
+ LOGGER.info("creating JsonPathFilter for " + this.pathExpression);
+ }
+
@Override
public List<StreamsDatum> process(StreamsDatum entry) {
@@ -102,7 +127,6 @@ public class JsonPathFilter implements StreamsProcessor {
node.set(destNodeName, jsonNode);
}
} catch (Exception e) {
- e.printStackTrace();
LOGGER.warn(e.getMessage());
}
} else if( srcResult instanceof JSONObject ) {
@@ -110,14 +134,12 @@ public class JsonPathFilter implements StreamsProcessor {
ObjectNode jsonNode = mapper.convertValue(srcResult, ObjectNode.class);
node.set(destNodeName, jsonNode);
} catch (Exception e) {
- e.printStackTrace();
LOGGER.warn(e.getMessage());
}
} else if( srcResult instanceof String ) {
try {
node.put(destNodeName, (String) srcResult);
} catch (Exception e) {
- e.printStackTrace();
LOGGER.warn(e.getMessage());
}
}
@@ -144,6 +166,6 @@ public class JsonPathFilter implements StreamsProcessor {
@Override
public void cleanUp() {
-
+ LOGGER.info("shutting down JsonPathFilter for " + this.pathExpression);
}
};
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/65157bd4/streams-contrib/streams-processor-json/src/test/java/org/apache/streams/json/test/JsonPathExtractorTest.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/src/test/java/org/apache/streams/json/test/JsonPathExtractorTest.java b/streams-contrib/streams-processor-json/src/test/java/org/apache/streams/json/test/JsonPathExtractorTest.java
index 2484939..7464ab1 100644
--- a/streams-contrib/streams-processor-json/src/test/java/org/apache/streams/json/test/JsonPathExtractorTest.java
+++ b/streams-contrib/streams-processor-json/src/test/java/org/apache/streams/json/test/JsonPathExtractorTest.java
@@ -19,12 +19,9 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
/**
-* Created with IntelliJ IDEA.
-* User: sblackmon
-* Date: 8/20/13
-* Time: 5:57 PM
-* To change this template use File | Settings | File Templates.
-*/
+ * Test for extracting json fields and
+ * objects from datums using JsonPath syntax
+ */
public class JsonPathExtractorTest {
private final static Logger LOGGER = LoggerFactory.getLogger(JsonPathExtractorTest.class);
@@ -52,7 +49,6 @@ public class JsonPathExtractorTest {
assertTrue(result.get(1).getDocument() instanceof String);
}
- @Ignore
@Test
public void test2()
{
@@ -63,7 +59,6 @@ public class JsonPathExtractorTest {
assertTrue(result.get(0).getDocument() instanceof ObjectNode);
}
- @Ignore
@Test
public void test3()
{
@@ -74,7 +69,6 @@ public class JsonPathExtractorTest {
assertTrue(result.get(0).getDocument() instanceof ObjectNode);
}
- @Ignore
@Test
public void test4()
{