class 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());

          /*
           * Deep clone the partitions
           */
          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;
    }
  }

}

class ArraySplitOperationTest  {

  @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);
  }

}