Enhance smart update

src/main/java/com/ontotext/kafka/GraphDBSinkConfig.java

@@ -1,7 +1,7 @@
 package com.ontotext.kafka;
 
-import com.ontotext.kafka.util.ValidateEnum;
-import com.ontotext.kafka.util.ValidateRDFFormat;
+import com.ontotext.kafka.util.EnumValidator;
+import com.ontotext.kafka.util.RDFFormatValidator;
 import com.ontotext.kafka.util.ValueUtil;
 import com.ontotext.kafka.util.VisibleIfRecommender;
 import org.apache.kafka.clients.CommonClientConfigs;
@@ -163,15 +163,15 @@ public static ConfigDef createConfigDef() {
 				RDF_FORMAT,
 				ConfigDef.Type.STRING,
 				DEFAULT_RDF_TYPE,
-				new ValidateRDFFormat(),
+				new RDFFormatValidator(),
 				ConfigDef.Importance.HIGH,
 				RDF_FORMAT_DOC
 			)
 			.define(
 				TRANSACTION_TYPE,
 				ConfigDef.Type.STRING,
 				DEFAULT_TRANSACTION_TYPE,
-				new ValidateEnum(TransactionType.class),
+				new EnumValidator(TransactionType.class),
 				ConfigDef.Importance.HIGH,
 				TRANSACTION_TYPE_DOC
 			)
@@ -200,7 +200,7 @@ public static ConfigDef createConfigDef() {
 				AUTH_TYPE,
 				ConfigDef.Type.STRING,
 				DEFAULT_AUTH_TYPE,
-				new ValidateEnum(AuthenticationType.class),
+				new EnumValidator(AuthenticationType.class),
 				ConfigDef.Importance.HIGH,
 				AUTH_TYPE_DOC
 			)

src/main/java/com/ontotext/kafka/GraphDBSinkConnector.java

@@ -16,7 +16,7 @@
 
 package com.ontotext.kafka;
 
-import com.ontotext.kafka.util.ValidateGraphDBConnection;
+import com.ontotext.kafka.util.GraphDBConnectionValidator;
 import com.ontotext.kafka.util.VersionUtil;
 import org.apache.kafka.common.config.Config;
 import org.apache.kafka.common.config.ConfigDef;
@@ -78,10 +78,10 @@ public ConfigDef config() {
 
 	@Override
 	public Config validate(final Map<String, String> connectorConfigs) {
-		var config = super.validate(connectorConfigs);
+		Config config = super.validate(connectorConfigs);
 		if (config.configValues().stream().anyMatch(cv -> !cv.errorMessages().isEmpty())) {
 			return config;
 		}
-		return ValidateGraphDBConnection.validateGraphDBConnection(config);
+		return GraphDBConnectionValidator.validateGraphDBConnection(config);
 	}
 }

src/main/java/com/ontotext/kafka/convert/JsonDataConverter.java

@@ -0,0 +1,72 @@
+package com.ontotext.kafka.convert;
+
+import com.fasterxml.jackson.core.exc.StreamReadException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DatabindException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.ontotext.kafka.graphdb.template.ValueParser;
+import com.ontotext.kafka.util.ValueUtil;
+import org.eclipse.rdf4j.model.Value;
+import org.eclipse.rdf4j.rio.RDFFormat;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Map;
+
+public final class JsonDataConverter extends RdfFormatConverter {
+
+	private final ObjectMapper objectMapper;
+
+	public JsonDataConverter(RDFFormat inputFormat) {
+		super(inputFormat, RDFFormat.JSONLD);
+		this.objectMapper = new ObjectMapper();
+	}
+
+	public <T> T convert(byte[] data, Class<T> cls) {
+		try {
+			byte[] converted = convertData(data);
+			return objectMapper.readValue(converted, cls);
+		} catch (DatabindException e) {
+			log.error("Could not deserialize data to type {}", cls, e);
+		} catch (StreamReadException e) {
+			log.error("Caught exception while reading data stream", e);
+		} catch (IOException e) {
+			log.error("Could not convert data from {} to JSON", getInputFormat(), e);
+		}
+		return null;
+	}
+
+	public Map<String, Object> convert(byte[] data) {
+		try {
+			byte[] converted = convertData(data);
+			return objectMapper.readValue(converted, new TypeReference<>() {
+			});
+		} catch (DatabindException e) {
+			log.error("Could not deserialize data to a generic map", e);
+		} catch (StreamReadException e) {
+			log.error("Caught exception while reading data stream", e);
+		} catch (IOException e) {
+			log.error("Could not convert data from {} to JSON", getInputFormat(), e);
+		}
+		return null;
+	}
+
+	public static void main(String[] args) throws IOException {
+		byte[] data = Files.readAllBytes(Paths.get(args[0]));
+		JsonDataConverter converter = new JsonDataConverter(ValueUtil.getRDFFormat(args[1]));
+		Map<String, Object> map = converter.convert(data);
+		ObjectMapper objectMapper1 = new ObjectMapper();
+		System.out.println(objectMapper1.writerWithDefaultPrettyPrinter().writeValueAsString(map));
+		for (String key : map.keySet()) {
+			Object obj = map.get(key);
+			Value value = ValueParser.getValue(obj);
+			if (value == null) {
+				continue;
+			}
+			System.out.printf("%s : %s\n", key, value);
+		}
+	}
+
+
+}

src/main/java/com/ontotext/kafka/convert/RdfFormatConverter.java

@@ -0,0 +1,50 @@
+package com.ontotext.kafka.convert;
+
+import org.eclipse.rdf4j.rio.RDFFormat;
+import org.eclipse.rdf4j.rio.RDFParser;
+import org.eclipse.rdf4j.rio.RDFWriter;
+import org.eclipse.rdf4j.rio.Rio;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+
+public class RdfFormatConverter {
+
+	protected final Logger log;
+	private final RDFFormat inputFormat;
+	private final RDFFormat outputFormat;
+
+	public RdfFormatConverter(RDFFormat inputFormat, RDFFormat outputFormat) {
+		this.inputFormat = inputFormat;
+		this.outputFormat = outputFormat;
+		this.log = LoggerFactory.getLogger(String.format("Converter [%s]->[%s]", inputFormat, outputFormat));
+	}
+
+
+	public byte[] convertData(byte[] inputData) throws IOException {
+		if (inputFormat.equals(outputFormat)) {
+			log.debug("Input and output formats [{}] are the same, skipping conversion", inputData);
+			return inputData;
+		}
+		log.debug("Converting data from {} to {}", inputFormat, outputFormat);
+		RDFParser rdfParser = Rio.createParser(inputFormat);
+		try (InputStream is = new ByteArrayInputStream(inputData);
+			 ByteArrayOutputStream baos = new ByteArrayOutputStream(1024);
+			 BufferedOutputStream bos = new BufferedOutputStream(baos)) {
+			RDFWriter rdfWriter = Rio.createWriter(outputFormat, bos);
+			rdfParser.setRDFHandler(rdfWriter);
+			rdfParser.parse(is);
+			return baos.toByteArray();
+		}
+	}
+
+
+	public RDFFormat getInputFormat() {
+		return inputFormat;
+	}
+
+	public RDFFormat getOutputFormat() {
+		return outputFormat;
+	}
+}

src/main/java/com/ontotext/kafka/graphdb/template/TemplateInput.java

@@ -0,0 +1,22 @@
+package com.ontotext.kafka.graphdb.template;
+
+import java.util.Map;
+
+public class TemplateInput {
+
+	private final String templateId;
+	private final Map<String, Object> data;
+
+	public TemplateInput(String templateId, Map<String, Object> data) {
+		this.templateId = templateId;
+		this.data = data;
+	}
+
+	public String getTemplateId() {
+		return templateId;
+	}
+
+	public Map<String, Object> getData() {
+		return data;
+	}
+}

src/main/java/com/ontotext/kafka/graphdb/template/TemplateUtil.java

@@ -0,0 +1,69 @@
+package com.ontotext.kafka.graphdb.template;
+
+import org.apache.kafka.common.config.ConfigException;
+import org.eclipse.rdf4j.http.protocol.UnauthorizedException;
+import org.eclipse.rdf4j.model.Value;
+import org.eclipse.rdf4j.query.QueryLanguage;
+import org.eclipse.rdf4j.query.TupleQueryResult;
+import org.eclipse.rdf4j.query.Update;
+import org.eclipse.rdf4j.repository.RepositoryConnection;
+import org.eclipse.rdf4j.repository.RepositoryException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+public final class TemplateUtil {
+
+	private static final Logger LOG = LoggerFactory.getLogger(TemplateUtil.class);
+
+	private TemplateUtil() {
+		throw new IllegalStateException("Utility class");
+	}
+
+	private static final String TEMPLATE_CONTENT_QUERY = "select ?template {\n <%s> <http://www.ontotext.com/sparql/template> ?template\n}";
+
+	public static void executeUpdate(RepositoryConnection connection, TemplateInput input) {
+		try {
+			Update update = prepareUpdate(input, connection);
+			bindInput(input.getData(), update);
+			update.execute();
+		} catch (Exception e) {
+			e.printStackTrace();
+			// TODO
+		}
+	}
+
+	private static void bindInput(Map<String, Object> input, Update update) {
+		for (String key : input.keySet()) {
+			Object obj = input.get(key);
+			Value value = ValueParser.getValue(obj);
+			if (value == null) {
+				LOG.warn("Binding for {} is null or empty. Skipping binding.", key);
+				continue;
+			}
+			update.setBinding(key, value);
+		}
+	}
+
+
+	private static Update prepareUpdate(TemplateInput input, RepositoryConnection connection) {
+		return connection.prepareUpdate(QueryLanguage.SPARQL, getSparqlTemplateContent(input.getTemplateId(), connection));
+	}
+
+	private static String getSparqlTemplateContent(String templateId, RepositoryConnection connection) {
+		LOG.debug("Querying template ID {}", templateId);
+		try (TupleQueryResult templates = connection.prepareTupleQuery(String.format(TEMPLATE_CONTENT_QUERY, templateId)).evaluate()) {
+			if (templates.hasNext()) {
+				// Only interested in first result
+				return templates.next().getValue("template").stringValue();
+			}
+			throw new ConfigException("Did not find template with ID {}", templateId);
+		} catch (RepositoryException e) {
+			if (e instanceof UnauthorizedException) {
+				throw new ConfigException("Invalid credentials" + e.getMessage());
+			}
+			throw e;
+		}
+	}
+}