class JsonExtractStringKudf implements Kudf {

  @Override
  public Object evaluate(Object... args) {
    if (args.length != 2) {
      throw new KsqlFunctionException("getStringFromJson udf should have two input arguments:"
                                      + " JSON document and JSON path.");
    }

    ensureInitialized(args);

    JsonNode currentNode = parseJsonDoc(args[0]);
    for (String token : tokens) {
      if (currentNode instanceof ArrayNode) {
        try {
          final int index = Integer.parseInt(token);
          currentNode = currentNode.get(index);
        } catch (NumberFormatException e) {
          return null;
        }
      } else {
        currentNode = currentNode.get(token);
      }

      if (currentNode == null) {
        return null;
      }
    }

    if (currentNode.isTextual()) {
      return currentNode.asText();
    } else {
      return currentNode.toString();
    }
  }

}