ArraySplitOperation implements MultiplexOperation { @Override public List<InternalEvent> perform(InternalEvent ievent) throws OperationException { { if (ievent.getEventObj() == null) { throw new OperationException("Deserialized object is null"); } Object payload; try { payload = ievent.getEventObj().getField(this.path); } catch (FieldNotFoundException e) { throw new OperationException(e); } if (!(payload instanceof JsonArray)) { throw new OperationException("Payload data is not a JsonArray"); } LinkedHashMap<String, String> partitions = ievent.getPartitions(); JsonArray arr = (JsonArray) payload; ArrayList<InternalEvent> output = new ArrayList<InternalEvent>(); for (JsonElement elm : arr) { try { JsonObject newObject = elm.getAsJsonObject(); for (String field : this.fieldsToKeep) { JsonObject jsonObject = (JsonObject) ievent.getEventObj().getPayload(); newObject.add(field, jsonObject.get(field)); } InternalEvent newEvent = new InternalEvent(newObject.toString(), ievent.getCtx(), ievent.getArrivalTime()); DeserializedEvent newDeserEvent = new GenericJsonEvent(newObject); newEvent.setEventObj(newDeserEvent); newEvent.setEventTime(ievent.getEventTime()); if (partitions != null) { LinkedHashMap<String, String> newPartitions = new LinkedHashMap<String, String>(partitions.size()); partitions.entrySet().forEach(kv -> { newPartitions.put(new String(kv.getKey()), new String(kv.getValue())); }); newEvent.setPartitions(newPartitions); } output.add(newEvent); } catch (Exception e) { throw new OperationException(e); } } return output; } } ArraySplitOperation(String path);  ArraySplitOperation(String path,
                             List<String> fieldsToKeep); @Override List<InternalEvent> perform(InternalEvent ievent);  }
@Test public void testEventString() throws JsonSyntaxException, UnsupportedEncodingException, IOException { JsonParser parser = new JsonParser(); JsonElement input = parser.parse(getResourceString("array_input.json")); GenericJsonEvent devent = new GenericJsonEvent(input.getAsJsonObject()); ArraySplitOperation operation = new ArraySplitOperation("$.arr"); InternalEvent ievent = new InternalEvent("", null, 0); ievent.setEventObj(devent); List<String> actual = operation.perform(ievent).stream().map(InternalEvent::getEventString) .collect(Collectors.toList()); List<String> expected = Arrays.asList("{\"foo\":1}", "{\"foo\":2}", "{\"foo\":3}"); assertEquals(expected, actual); }
@Test public void testEventObject() throws JsonSyntaxException, UnsupportedEncodingException, IOException { JsonParser parser = new JsonParser(); JsonElement input = parser.parse(getResourceString("array_input.json")); GenericJsonEvent devent = new GenericJsonEvent(input.getAsJsonObject()); ArraySplitOperation operation = new ArraySplitOperation("$.arr"); InternalEvent ievent = new InternalEvent("", null, 0); ievent.setEventObj(devent); List<String> actual = operation.perform(ievent).stream().map(i -> { return i.getEventObj().getPayload().toString(); }).collect(Collectors.toList()); List<String> expected = Arrays.asList("{\"foo\":1}", "{\"foo\":2}", "{\"foo\":3}"); assertEquals(expected, actual); }
@Test public void testContextMatch() throws JsonSyntaxException, UnsupportedEncodingException, IOException { JsonParser parser = new JsonParser(); TestContext t = new TestContext(); t.setFunctionName("foo"); LambdaContext lctx = new LambdaContext(t); JsonElement input = parser.parse(getResourceString("array_input.json")); GenericJsonEvent devent = new GenericJsonEvent(input.getAsJsonObject()); ArraySplitOperation operation = new ArraySplitOperation("$.arr"); InternalEvent ievent = new InternalEvent("", lctx, 1); ievent.setEventObj(devent); List<InternalEvent> actual = operation.perform(ievent); assertEquals(lctx, actual.get(0).getCtx()); assertEquals(lctx, actual.get(1).getCtx()); assertEquals(lctx, actual.get(2).getCtx()); }
@Test public void testMetadataMatch() throws JsonSyntaxException, UnsupportedEncodingException, IOException { JsonParser parser = new JsonParser(); TestContext t = new TestContext(); t.setFunctionName("foo"); LambdaContext lctx = new LambdaContext(t); JsonElement input = parser.parse(getResourceString("array_input.json")); GenericJsonEvent devent = new GenericJsonEvent(input.getAsJsonObject()); ArraySplitOperation operation = new ArraySplitOperation("$.arr"); InternalEvent ievent = new InternalEvent("", lctx, 1); ievent.setEventObj(devent); List<InternalEvent> events = operation.perform(ievent); assertEquals(new Long(1), events.get(0).getEventMetadata().get("arrivalEpochMs")); assertEquals("6a00541bfa24fd59884de557be71f3a5a1344613", events.get(0).getEventMetadata().get("eventSha1Hash")); assertEquals(new Long(1), events.get(1).getEventMetadata().get("arrivalEpochMs")); assertEquals("5cbc013a144777d9ce51385ac24a95ce05f77075", events.get(1).getEventMetadata().get("eventSha1Hash")); assertEquals(new Long(1), events.get(2).getEventMetadata().get("arrivalEpochMs")); assertEquals("6c3023db7ef03721c27df9f33b26e0d2cbc84d6b", events.get(2).getEventMetadata().get("eventSha1Hash")); }
@Test public void testTimeMatch() throws JsonSyntaxException, UnsupportedEncodingException, IOException { JsonParser parser = new JsonParser(); TestContext t = new TestContext(); t.setFunctionName("foo"); LambdaContext lctx = new LambdaContext(t); JsonElement input = parser.parse(getResourceString("array_input.json")); GenericJsonEvent devent = new GenericJsonEvent(input.getAsJsonObject()); ArraySplitOperation operation = new ArraySplitOperation("$.arr"); InternalEvent ievent = new InternalEvent("", lctx, 123); ievent.setEventObj(devent); ievent.setEventTime(124); List<InternalEvent> events = operation.perform(ievent); assertEquals(ievent.getArrivalTime(), events.get(0).getArrivalTime()); assertEquals(ievent.getEventTime(), events.get(0).getEventTime()); assertEquals(ievent.getArrivalTime(), events.get(1).getArrivalTime()); assertEquals(ievent.getEventTime(), events.get(1).getEventTime()); assertEquals(ievent.getArrivalTime(), events.get(2).getArrivalTime()); assertEquals(ievent.getEventTime(), events.get(2).getEventTime()); }
@Test(expected = OperationException.class) public void testNonArray() throws JsonSyntaxException, UnsupportedEncodingException, IOException { JsonParser parser = new JsonParser(); TestContext t = new TestContext(); t.setFunctionName("foo"); LambdaContext lctx = new LambdaContext(t); JsonElement input = parser.parse(getResourceString("array_input.json")); GenericJsonEvent devent = new GenericJsonEvent(input.getAsJsonObject()); ArraySplitOperation operation = new ArraySplitOperation("$.arr[0]"); InternalEvent ievent = new InternalEvent("", lctx, 123); ievent.setEventObj(devent); ievent.setEventTime(124); List<InternalEvent> events = operation.perform(ievent); }
@Test(expected = OperationException.class) public void testMissingField() throws JsonSyntaxException, UnsupportedEncodingException, IOException { JsonParser parser = new JsonParser(); TestContext t = new TestContext(); t.setFunctionName("foo"); LambdaContext lctx = new LambdaContext(t); JsonElement input = parser.parse(getResourceString("array_input.json")); GenericJsonEvent devent = new GenericJsonEvent(input.getAsJsonObject()); ArraySplitOperation operation = new ArraySplitOperation("$.bar]"); InternalEvent ievent = new InternalEvent("", lctx, 123); ievent.setEventObj(devent); ievent.setEventTime(124); List<InternalEvent> events = operation.perform(ievent); }
@Test public void testArraySplitWithFieldsToKeep() throws IOException { TestContext t = new TestContext(); t.setFunctionName("foo"); LambdaContext lctx = new LambdaContext(t); JsonElement input = JsonParser.parseString(getResourceString("array_input_kinesis_format.json")); GenericJsonEvent devent = new GenericJsonEvent(input.getAsJsonObject()); List<String> fieldsToKeep = Arrays.asList("owner", "logGroup", "logStream"); ArraySplitOperation operation = new ArraySplitOperation("$.logEvents", fieldsToKeep); InternalEvent ievent = new InternalEvent("", lctx, 123); ievent.setEventObj(devent); ievent.setEventTime(124); List<InternalEvent> events = operation.perform(ievent); assertEquals(4, events.size()); JsonElement actual = JsonParser.parseString(events.get(0).getEventString()); assertTrue(actual.getAsJsonObject().has("owner")); assertTrue(actual.getAsJsonObject().has("logGroup")); assertTrue(actual.getAsJsonObject().has("logStream")); assertTrue(actual.getAsJsonObject().has("id")); assertTrue(actual.getAsJsonObject().has("message")); assertTrue(actual.getAsJsonObject().has("timestamp")); actual = JsonParser.parseString(events.get(1).getEventString()); assertTrue(actual.getAsJsonObject().has("owner")); assertTrue(actual.getAsJsonObject().has("logGroup")); assertTrue(actual.getAsJsonObject().has("logStream")); assertTrue(actual.getAsJsonObject().has("id")); assertTrue(actual.getAsJsonObject().has("message")); assertTrue(actual.getAsJsonObject().has("timestamp")); actual = JsonParser.parseString(events.get(2).getEventString()); assertTrue(actual.getAsJsonObject().has("owner")); assertTrue(actual.getAsJsonObject().has("logGroup")); assertTrue(actual.getAsJsonObject().has("logStream")); assertTrue(actual.getAsJsonObject().has("id")); assertTrue(actual.getAsJsonObject().has("message")); assertTrue(actual.getAsJsonObject().has("timestamp")); actual = JsonParser.parseString(events.get(3).getEventString()); assertTrue(actual.getAsJsonObject().has("owner")); assertTrue(actual.getAsJsonObject().has("logGroup")); assertTrue(actual.getAsJsonObject().has("logStream")); assertTrue(actual.getAsJsonObject().has("id")); assertTrue(actual.getAsJsonObject().has("message")); assertTrue(actual.getAsJsonObject().has("timestamp")); }