JsonByteArrayOperator extends BaseOperator { public void setConcatenationCharacter(char c) { JsonByteArrayOperator.this.CONCAT_CHAR = c; }  void setConcatenationCharacter(char c); final transient DefaultInputPort<byte[]> input; final transient DefaultOutputPort<HashMap<String, Object>> outputMap; final transient DefaultOutputPort<JSONObject> outputJsonObject; final transient DefaultOutputPort<HashMap<String, Object>> outputFlatMap; }
@SuppressWarnings({"rawtypes", "unchecked"}) @Test public void testOperator() throws Exception { JsonByteArrayOperator oper = new JsonByteArrayOperator(); oper.setConcatenationCharacter('.'); CollectorTestSink mapSink = new CollectorTestSink(); CollectorTestSink jsonObjectSink = new CollectorTestSink(); CollectorTestSink flatMapSink = new CollectorTestSink(); oper.outputMap.setSink(mapSink); oper.outputJsonObject.setSink(jsonObjectSink); oper.outputFlatMap.setSink(flatMapSink); oper.beginWindow(0); String inputJson = " { \"@timestamp\":\"2013-09-25T19:37:23.569Z\"" + " ,\"@version\":\"1\"" + " ,\"type\":\"apache-logs\"" + " ,\"host\":\"node1001\"" + " ,\"clientip\":192.168.150.120" + " ,\"verb\":\"GET\"" + " ,\"request\":\"/reset.css\"" + " ,\"httpversion\":\"1.1\"" + " ,\"response\":200" + " ,\"agentinfo\": {\"browser\":Firefox" + " ,\"os\": { \"name\":\"Ubuntu\"" + " ,\"version\":\"10.04\"" + " }" + " }" + " ,\"bytes\":909.1" + " }"; byte[] inputByteArray = inputJson.getBytes(); int numtuples = 1000; for (int i = 0; i < numtuples; i++) { oper.input.process(inputByteArray); } oper.endWindow(); Assert.assertEquals("number emitted tuples", numtuples, mapSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", numtuples, jsonObjectSink.collectedTuples.size()); Assert.assertEquals("number emitted tuples", numtuples, flatMapSink.collectedTuples.size()); Object map = mapSink.collectedTuples.get(510); String expectedClientip = "192.168.150.120"; Assert.assertEquals("emitted tuple", expectedClientip, ((Map)map).get("clientip")); Object jsonObject = jsonObjectSink.collectedTuples.get(433); Number expectedResponse = 200; Assert.assertEquals("emitted tuple", expectedResponse, ((JSONObject)jsonObject).get("response")); Map flatMap = (Map)flatMapSink.collectedTuples.get(511); String expectedBrowser = "Firefox"; String expectedOsName = "Ubuntu"; Assert.assertEquals("emitted tuple", expectedBrowser, flatMap.get("agentinfo.browser")); Assert.assertEquals("emitted tuple", expectedOsName, flatMap.get("agentinfo.os.name")); }