fix: add jsonpath Option while parsing template

src/main/java/com/gotocompany/firehose/config/HttpSinkConfig.java

@@ -1,15 +1,11 @@
 package com.gotocompany.firehose.config;
 
-import com.gotocompany.firehose.config.converter.HttpSinkSerializerJsonTypecastConfigConverter;
+import com.gotocompany.firehose.config.converter.*;
 import com.gotocompany.firehose.config.enums.HttpSinkDataFormatType;
 import com.gotocompany.firehose.config.enums.HttpSinkParameterPlacementType;
 import com.gotocompany.firehose.config.enums.HttpSinkParameterSourceType;
 import com.gotocompany.firehose.config.enums.HttpSinkRequestMethodType;
-import com.gotocompany.firehose.config.converter.HttpSinkRequestMethodConverter;
-import com.gotocompany.firehose.config.converter.HttpSinkParameterDataFormatConverter;
-import com.gotocompany.firehose.config.converter.HttpSinkParameterPlacementTypeConverter;
-import com.gotocompany.firehose.config.converter.HttpSinkParameterSourceTypeConverter;
-import com.gotocompany.firehose.config.converter.RangeToHashMapConverter;
+import com.jayway.jsonpath.Option;
 
 import java.util.Map;
 import java.util.function.Function;
@@ -80,6 +76,11 @@ public interface HttpSinkConfig extends AppConfig {
     @DefaultValue("")
     String getSinkHttpJsonBodyTemplate();
 
+    @Key("SINK_HTTP_JSON_BODY_TEMPLATE_PARSE_OPTION")
+    @DefaultValue("")
+    @ConverterClass(HttpJsonBodyTemplateParseOptionConverter.class)
+    Option getSinkHttpJsonBodyTemplateParseOption();
+
     @Key("SINK_HTTP_PARAMETER_PLACEMENT")
     @DefaultValue("header")
     @ConverterClass(HttpSinkParameterPlacementTypeConverter.class)

src/main/java/com/gotocompany/firehose/config/converter/HttpJsonBodyTemplateParseOptionConverter.java

@@ -0,0 +1,14 @@
+package com.gotocompany.firehose.config.converter;
+
+import com.jayway.jsonpath.Option;
+import org.aeonbits.owner.Converter;
+
+import java.lang.reflect.Method;
+
+public class HttpJsonBodyTemplateParseOptionConverter implements Converter<Option> {
+    @Override
+    public Option convert(Method method, String input) {
+        return Option.valueOf(input);
+    }
+}
+

src/main/java/com/gotocompany/firehose/serializer/MessageToTemplatizedJson.java

@@ -9,7 +9,9 @@
 import com.google.protobuf.DynamicMessage;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.util.JsonFormat;
+import com.jayway.jsonpath.Configuration;
 import com.jayway.jsonpath.JsonPath;
+import com.jayway.jsonpath.Option;
 import com.jayway.jsonpath.PathNotFoundException;
 import com.gotocompany.stencil.Parser;
 import org.json.simple.parser.JSONParser;
@@ -32,10 +34,11 @@ public class MessageToTemplatizedJson implements MessageSerializer {
     private Parser protoParser;
     private HashSet<String> pathsToReplace;
     private JSONParser jsonParser;
+    private final Configuration jsonPathConfig;
     private FirehoseInstrumentation firehoseInstrumentation;
 
-    public static MessageToTemplatizedJson create(FirehoseInstrumentation firehoseInstrumentation, String httpSinkJsonBodyTemplate, Parser protoParser) {
-        MessageToTemplatizedJson messageToTemplatizedJson = new MessageToTemplatizedJson(firehoseInstrumentation, httpSinkJsonBodyTemplate, protoParser);
+    public static MessageToTemplatizedJson create(FirehoseInstrumentation firehoseInstrumentation, String httpSinkJsonBodyTemplate, Parser protoParser, Option option) {
+        MessageToTemplatizedJson messageToTemplatizedJson = new MessageToTemplatizedJson(firehoseInstrumentation, httpSinkJsonBodyTemplate, protoParser, option);
         if (messageToTemplatizedJson.isInvalidJson()) {
             throw new ConfigurationException("Given HTTPSink JSON body template :"
                     + httpSinkJsonBodyTemplate
@@ -45,11 +48,12 @@ public static MessageToTemplatizedJson create(FirehoseInstrumentation firehoseIn
         return messageToTemplatizedJson;
     }
 
-    public MessageToTemplatizedJson(FirehoseInstrumentation firehoseInstrumentation, String httpSinkJsonBodyTemplate, Parser protoParser) {
+    public MessageToTemplatizedJson(FirehoseInstrumentation firehoseInstrumentation, String httpSinkJsonBodyTemplate, Parser protoParser, Option option) {
         this.httpSinkJsonBodyTemplate = httpSinkJsonBodyTemplate;
         this.protoParser = protoParser;
         this.jsonParser = new JSONParser();
         this.gson = new Gson();
+        this.jsonPathConfig = Configuration.defaultConfiguration().addOptions(option);
         this.firehoseInstrumentation = firehoseInstrumentation;
     }
 
@@ -90,6 +94,22 @@ public String serialize(Message message) throws DeserializerException {
                 }
                 finalMessage = finalMessage.replace(path, jsonString);
             }
+
+            for (String path : pathsToReplace) {
+                if (path.equals(ALL_FIELDS_FROM_TEMPLATE)) {
+                    jsonString = jsonMessage;
+                } else {
+                    Object element = JsonPath.using(jsonPathConfig).parse(jsonMessage).read(path.replaceAll("\"", ""));
+                    if (element == null && jsonPathConfig.getOptions().contains(Option.SUPPRESS_EXCEPTIONS)) {
+                        firehoseInstrumentation.logWarn("Missing value for path: {}", path);
+                        jsonString = "";
+                    } else {
+                        jsonString = gson.toJson(element);
+                    }
+                }
+                finalMessage = finalMessage.replace(path, jsonString);
+            }
+
             return finalMessage;
         } catch (InvalidProtocolBufferException | PathNotFoundException e) {
             throw new DeserializerException(e.getMessage());

src/main/java/com/gotocompany/firehose/sink/http/factory/SerializerFactory.java

@@ -39,7 +39,7 @@ public MessageSerializer build() {
             } else {
                 firehoseInstrumentation.logDebug("Serializer type: EsbMessageToTemplatizedJson");
                 return getTypecastedJsonSerializer(
-                        MessageToTemplatizedJson.create(new FirehoseInstrumentation(statsDReporter, MessageToTemplatizedJson.class), httpSinkConfig.getSinkHttpJsonBodyTemplate(), protoParser));
+                        MessageToTemplatizedJson.create(new FirehoseInstrumentation(statsDReporter, MessageToTemplatizedJson.class), httpSinkConfig.getSinkHttpJsonBodyTemplate(), protoParser, httpSinkConfig.getSinkHttpJsonBodyTemplateParseOption()));
             }
         }