JsonParser extends Parser<byte[], KeyValPair<String, String>> { @Override public void beginWindow(long windowId) { super.beginWindow(windowId); parsedOutputCount = 0; }  @Override void beginWindow(long windowId); @Override void setup(OperatorContext context); @Override void processTuple(byte[] tuple); @Override Object convert(byte[] tuple); @Override KeyValPair<String, String> processErrorTuple(byte[] input); String getJsonSchema(); void setJsonSchema(String jsonSchema); @VisibleForTesting long getErrorTupleCount(); @VisibleForTesting long getEmittedObjectCount(); @VisibleForTesting long getIncomingTuplesCount(); @VisibleForTesting void setSchema(JsonSchema schema); transient DefaultOutputPort<JSONObject> parsedOutput; }
@Test public void testValidInput() throws JSONException { String tuple = "{" + "\"id\": 2," + "\"name\": \"An ice sculpture\"," + "\"price\": 1," + "\"tags\": [\"cold\", \"ice\"]," + "\"dimensions\": {" + "\"length\": 7.0," + "\"width\" : 8.0," + "\"height\": 9.5" + "}," + "\"warehouseLocation\": {" + "\"latitude\": -78.75," + "\"longitude\": 20.4" + "}," + "\"dateOfManufacture\": \"2013/09/29\"," + "\"dateOfExpiry\": \"2013\"" + "}"; parser.beginWindow(0); parser.in.process(tuple.getBytes()); parser.endWindow(); Assert.assertEquals(1, objectPort.collectedTuples.size()); Assert.assertEquals(1, pojoPort.collectedTuples.size()); Assert.assertEquals(0, error.collectedTuples.size()); Object obj = pojoPort.collectedTuples.get(0); Assert.assertNotNull(obj); Assert.assertEquals(Product.class, obj.getClass()); Product pojo = (Product)obj; JSONObject jsonObject = (JSONObject)objectPort.collectedTuples.get(0); Assert.assertEquals(2, jsonObject.getInt("id")); Assert.assertEquals(1, jsonObject.getInt("price")); Assert.assertEquals("An ice sculpture", jsonObject.get("name")); Assert.assertEquals(7.0, jsonObject.getJSONObject("dimensions").getDouble("length"), 0); Assert.assertEquals(2, pojo.getId()); Assert.assertEquals(1, pojo.getPrice()); Assert.assertEquals("An ice sculpture", pojo.getName()); Assert.assertEquals(7.0, (double)pojo.getDimensions().get("length"), 0); }
@Test public void testInvalidPrice() throws JSONException { String tuple = "{" + "\"id\": 2," + "\"name\": \"An ice sculpture\"," + "\"price\": -1," + "\"tags\": [\"cold\", \"ice\"]," + "\"dimensions\": {" + "\"length\": 7.0," + "\"width\" : 8.0," + "\"height\": 9.5" + "}," + "\"warehouseLocation\": {" + "\"latitude\": -78.75," + "\"longitude\": 20.4" + "}," + "\"dateOfManufacture\": \"2013/09/29\"," + "\"dateOfExpiry\": \"2013\"" + "}"; parser.beginWindow(0); parser.in.process(tuple.getBytes()); parser.endWindow(); Assert.assertEquals(0, objectPort.collectedTuples.size()); Assert.assertEquals(0, pojoPort.collectedTuples.size()); Assert.assertEquals(1, error.collectedTuples.size()); KeyValPair<String, String> errorKeyValPair = (KeyValPair<String, String>)error.collectedTuples.get(0); Assert.assertEquals(tuple, errorKeyValPair.getKey()); Assert.assertEquals("\"/price\":\"number is lower than the required minimum\"", errorKeyValPair.getValue()); }
@Test public void testMultipleViolations() throws JSONException { String tuple = "{" + "\"id\": 2," + "\"name\": \"An ice sculpture\"," + "\"price\": -1," + "\"tags\": [\"cold\", \"ice\"]," + "\"dimensions\": {" + "\"width\" : 8.0," + "\"height\": 9.5" + "}," + "\"warehouseLocation\": {" + "\"latitude\": -78.75," + "\"longitude\": 20.4" + "}," + "\"dateOfManufacture\": \"2013/09/29\"," + "\"dateOfExpiry\": \"2013\"" + "}"; parser.beginWindow(0); parser.in.process(tuple.getBytes()); parser.endWindow(); Assert.assertEquals(0, objectPort.collectedTuples.size()); Assert.assertEquals(0, pojoPort.collectedTuples.size()); Assert.assertEquals(1, error.collectedTuples.size()); KeyValPair<String, String> errorKeyValPair = (KeyValPair<String, String>)error.collectedTuples.get(0); Assert.assertEquals(tuple, errorKeyValPair.getKey()); Assert.assertEquals( "\"/dimensions\":\"missing required property(ies)\",\"/price\":\"number is lower than the required minimum\"", errorKeyValPair.getValue()); }
@Test public void testJsonSyntaxError() throws JSONException { String tuple = "{" + "\"id\": 2," + "\"name\": \"An ice sculpture\"," + "\"price\": -1," + "\"tags\": [\"cold\", \"ice\"]," + "\"dimensions\": {" + "\"width\" : 8.0," + "\"height\": 9.5" + "}" + "\"warehouseLocation\": {" + "\"latitude\": -78.75," + "\"longitude\": 20.4" + "}," + "\"dateOfManufacture\": \"2013/09/29\"," + "\"dateOfExpiry\": \"2013\"" + "}"; parser.beginWindow(0); parser.in.process(tuple.getBytes()); parser.endWindow(); Assert.assertEquals(0, objectPort.collectedTuples.size()); Assert.assertEquals(0, pojoPort.collectedTuples.size()); Assert.assertEquals(1, error.collectedTuples.size()); KeyValPair<String, String> errorKeyValPair = (KeyValPair<String, String>)error.collectedTuples.get(0); Assert.assertEquals(tuple, errorKeyValPair.getKey()); }
@Test public void testValidInputPojoPortNotConnected() throws JSONException { parser.out.setSink(null); String tuple = "{" + "\"id\": 2," + "\"name\": \"An ice sculpture\"," + "\"price\": 1," + "\"tags\": [\"cold\", \"ice\"]," + "\"dimensions\": {" + "\"length\": 7.0," + "\"width\" : 8.0," + "\"height\": 9.5" + "}," + "\"warehouseLocation\": {" + "\"latitude\": -78.75," + "\"longitude\": 20.4" + "}," + "\"dateOfManufacture\": \"2013/09/29\"," + "\"dateOfExpiry\": \"2013\"" + "}"; parser.beginWindow(0); parser.in.process(tuple.getBytes()); parser.endWindow(); Assert.assertEquals(1, objectPort.collectedTuples.size()); Assert.assertEquals(0, pojoPort.collectedTuples.size()); Assert.assertEquals(0, error.collectedTuples.size()); JSONObject jsonObject = (JSONObject)objectPort.collectedTuples.get(0); Assert.assertEquals(2, jsonObject.getInt("id")); Assert.assertEquals(1, jsonObject.getInt("price")); Assert.assertEquals("An ice sculpture", jsonObject.get("name")); Assert.assertEquals(7.0, jsonObject.getJSONObject("dimensions").getDouble("length"), 0); }
@Test public void testValidInputParsedOutputPortNotConnected() throws JSONException { parser.parsedOutput.setSink(null); String tuple = "{" + "\"id\": 2," + "\"name\": \"An ice sculpture\"," + "\"price\": 1," + "\"tags\": [\"cold\", \"ice\"]," + "\"dimensions\": {" + "\"length\": 7.0," + "\"width\" : 8.0," + "\"height\": 9.5" + "}," + "\"warehouseLocation\": {" + "\"latitude\": -78.75," + "\"longitude\": 20.4" + "}," + "\"dateOfManufacture\": \"2013/09/29\"," + "\"dateOfExpiry\": \"2013\"" + "}"; parser.beginWindow(0); parser.in.process(tuple.getBytes()); parser.endWindow(); Assert.assertEquals(0, objectPort.collectedTuples.size()); Assert.assertEquals(1, pojoPort.collectedTuples.size()); Assert.assertEquals(0, error.collectedTuples.size()); Object obj = pojoPort.collectedTuples.get(0); Assert.assertNotNull(obj); Assert.assertEquals(Product.class, obj.getClass()); Product pojo = (Product)obj; Assert.assertEquals(2, pojo.getId()); Assert.assertEquals(1, pojo.getPrice()); Assert.assertEquals("An ice sculpture", pojo.getName()); Assert.assertEquals(7.0, (double)pojo.getDimensions().get("length"), 0); }