Add recordPath option to JsonDecoder (#382)

Instead of using the entire JSON as a single record, provide a
JSON path to query the JSON for the records to process, e.g.
`$.data` to process every entry in a `data` array as a record.

Author: fsteeg

metafacture-json/build.gradle

@@ -20,6 +20,8 @@ description = 'Modules for processing JSON data in Metafacture'
 dependencies {
   api project(':metafacture-framework')
   implementation 'com.fasterxml.jackson.core:jackson-core:2.8.5'
+  implementation 'com.fasterxml.jackson.core:jackson-databind:2.8.5'
+  implementation 'com.jayway.jsonpath:json-path:2.6.0'
   testImplementation 'junit:junit:4.12'
   testImplementation 'org.mockito:mockito-core:2.5.5'
 }

metafacture-json/src/main/java/org/metafacture/json/JsonDecoder.java

@@ -17,12 +17,19 @@
 
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jayway.jsonpath.JsonPath;
+
 import org.metafacture.framework.MetafactureException;
 import org.metafacture.framework.StreamReceiver;
 import org.metafacture.framework.helpers.DefaultObjectPipe;
 
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * Decodes a record in JSON format.
@@ -38,6 +45,8 @@ public final class JsonDecoder extends DefaultObjectPipe<String, StreamReceiver>
 
     public static final String DEFAULT_RECORD_ID = "%d";
 
+    public static final String DEFAULT_ROOT_PATH = "";
+
     private final JsonFactory jsonFactory = new JsonFactory();
 
     private JsonParser jsonParser;
@@ -46,12 +55,15 @@ public final class JsonDecoder extends DefaultObjectPipe<String, StreamReceiver>
     private String recordId;
     private int recordCount;
 
+    private String recordPath;
+
     public JsonDecoder() {
         super();
 
         setArrayMarker(DEFAULT_ARRAY_MARKER);
         setArrayName(DEFAULT_ARRAY_NAME);
         setRecordId(DEFAULT_RECORD_ID);
+        setRecordPath(DEFAULT_ROOT_PATH);
 
         resetRecordCount();
     }
@@ -96,25 +108,45 @@ public int getRecordCount() {
         return recordCount;
     }
 
+    public void setRecordPath(final String recordPath) {
+        this.recordPath = recordPath;
+    }
+
+    public String getRecordPath() {
+        return recordPath;
+    }
+
     public void resetRecordCount() {
         setRecordCount(0);
     }
 
     @Override
-    public void process(final String string) {
+    public void process(final String json) {
         assert !isClosed();
-
-        createParser(string);
-
-        try {
-            decode();
-        }
-        catch (final IOException e) {
-            throw new MetafactureException(e);
-        }
-        finally {
-            closeParser();
-        }
+        final List<String> records = recordPath.isEmpty() ? Arrays.asList(json)
+                : matches(JsonPath.read(json, recordPath));
+        records.forEach(record -> {
+            createParser(record);
+            try {
+                decode();
+            } catch (final IOException e) {
+                throw new MetafactureException(e);
+            } finally {
+                closeParser();
+            }
+        });
+    }
+
+    private List<String> matches(Object obj) {
+        final List<?> records = (obj instanceof List<?>) ? ((List<?>) obj) : Arrays.asList(obj);
+        return records.stream().map(doc -> {
+            try {
+                return new ObjectMapper().writeValueAsString(doc);
+            } catch (JsonProcessingException e) {
+                e.printStackTrace();
+                return doc.toString();
+            }
+        }).collect(Collectors.toList());
     }
 
     @Override

metafacture-json/src/test/java/org/metafacture/json/JsonDecoderTest.java

@@ -151,6 +151,22 @@ public void testShouldProcessConcatenatedRecords() {
         ordered.verify(receiver).endRecord();
     }
 
+    @Test
+    public void testShouldProcessRecordsInArray() {
+        jsonDecoder.setRecordPath("$.data");
+        jsonDecoder.process(
+                "{\"data\":[" + "{\"lit\": \"record 1\"}," +
+                        "{\"lit\": \"record 2\"}" + "]}");
+
+        final InOrder ordered = inOrder(receiver);
+        ordered.verify(receiver).startRecord("1");
+        ordered.verify(receiver).literal("lit", "record 1");
+        ordered.verify(receiver).endRecord();
+        ordered.verify(receiver).startRecord("2");
+        ordered.verify(receiver).literal("lit", "record 2");
+        ordered.verify(receiver).endRecord();
+    }
+
     @Test
     public void testShouldProcessMultipleRecords() {
         jsonDecoder.process("{\"lit\": \"record 1\"}");