// Path: src/main/java/org/apache/flink/mapping/neo4j/DeserializationMapper.java
// @FunctionalInterface
// public interface DeserializationMapper<T> extends Mapper {
// 
// 	/**
// 	 * Convert a generic item to a key-value pairs (used by a cypher statement)
// 	 * 
// 	 * @param item the data stream item
// 	 * @return a key-value set that will be used in a cypher statement
// 	 */
// 	public Map<String, Object> deserialize(T item);
// }
// 
// Path: src/main/java/org/apache/flink/mapping/neo4j/Neo4JDeserializationMappingStrategy.java
// public class Neo4JDeserializationMappingStrategy<T, M extends DeserializationMapper<T>> implements Serializable {
// 
// 	/**
// 	 * 
// 	 */
// 	private static final long serialVersionUID = 1L;
// 
// 	/**
// 	 * The templated cypher query that will be executed on Neo4J
// 	 */
// 	private String templateStatement;
// 
// 	/**
// 	 * The mapper that will map a item from a Flink datastream/dataset to a set
// 	 * of parameters
// 	 * 
// 	 * @see DeserializationMapper
// 	 */
// 	private M mapper;
// 
// 	/**
// 	 * 
// 	 * @param templateStatement
// 	 *            the cypher statement template
// 	 * @param mapper
// 	 *            an actual DeserializationMapper implementation
// 	 */
// 	public Neo4JDeserializationMappingStrategy(String templateStatement, M mapper) {
// 		this.templateStatement = templateStatement;
// 		this.mapper = mapper;
// 	}
// 
// 	/**
// 	 * Generate a statement with parameters (templated cypher query) for a given
// 	 * item and a convert function.
// 	 * 
// 	 * @param item
// 	 *            a data stream item
// 	 * @return the executable statement with its text and parameters
// 	 */
// 	public Statement getStatement(T item) {
// 		Statement statement = new Statement(templateStatement);
// 
// 		Map<String, Object> parameters = mapper.deserialize(item);
// 		return statement.withParameters(parameters);
// 	}
// 
// 	protected void setStatement(String templateStatement) {
// 		this.templateStatement = templateStatement;
// 	}
// }


import java.util.Map;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.mapping.neo4j.DeserializationMapper;
import org.apache.flink.mapping.neo4j.Neo4JDeserializationMappingStrategy;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.neo4j.driver.v1.Session;
import org.neo4j.driver.v1.Statement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 
 */
package org.apache.flink.streaming.connectors.neo4j;



/**
 * @author Alberto De Lazzari
 *
 */
public class Neo4JSink<T> extends RichSinkFunction<T> {

	/**
	 * The default serial version UID
	 */
	private static final long serialVersionUID = 1L;

	private static final Logger LOGGER = LoggerFactory.getLogger(Neo4JSink.class);

	/**
	 * Neo4J driver should be not serialized
	 */
	protected transient Neo4JDriverWrapper driver;

	/**
	 * The mapping strategy that we use to map data from Flink to Neo4J
	 * 
	 * @see Neo4JDeserializationMappingStrategy
	 */
// Path: src/main/java/org/apache/flink/mapping/neo4j/DeserializationMapper.java
// @FunctionalInterface
// public interface DeserializationMapper<T> extends Mapper {
// 
// 	/**
// 	 * Convert a generic item to a key-value pairs (used by a cypher statement)
// 	 * 
// 	 * @param item the data stream item
// 	 * @return a key-value set that will be used in a cypher statement
// 	 */
// 	public Map<String, Object> deserialize(T item);
// }
// 
// Path: src/main/java/org/apache/flink/mapping/neo4j/Neo4JDeserializationMappingStrategy.java
// public class Neo4JDeserializationMappingStrategy<T, M extends DeserializationMapper<T>> implements Serializable {
// 
// 	/**
// 	 * 
// 	 */
// 	private static final long serialVersionUID = 1L;
// 
// 	/**
// 	 * The templated cypher query that will be executed on Neo4J
// 	 */
// 	private String templateStatement;
// 
// 	/**
// 	 * The mapper that will map a item from a Flink datastream/dataset to a set
// 	 * of parameters
// 	 * 
// 	 * @see DeserializationMapper
// 	 */
// 	private M mapper;
// 
// 	/**
// 	 * 
// 	 * @param templateStatement
// 	 *            the cypher statement template
// 	 * @param mapper
// 	 *            an actual DeserializationMapper implementation
// 	 */
// 	public Neo4JDeserializationMappingStrategy(String templateStatement, M mapper) {
// 		this.templateStatement = templateStatement;
// 		this.mapper = mapper;
// 	}
// 
// 	/**
// 	 * Generate a statement with parameters (templated cypher query) for a given
// 	 * item and a convert function.
// 	 * 
// 	 * @param item
// 	 *            a data stream item
// 	 * @return the executable statement with its text and parameters
// 	 */
// 	public Statement getStatement(T item) {
// 		Statement statement = new Statement(templateStatement);
// 
// 		Map<String, Object> parameters = mapper.deserialize(item);
// 		return statement.withParameters(parameters);
// 	}
// 
// 	protected void setStatement(String templateStatement) {
// 		this.templateStatement = templateStatement;
// 	}
// }


// Path: src/main/java/org/apache/flink/streaming/connectors/neo4j/Neo4JSink.java
import java.util.Map;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.mapping.neo4j.DeserializationMapper;
import org.apache.flink.mapping.neo4j.Neo4JDeserializationMappingStrategy;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.neo4j.driver.v1.Session;
import org.neo4j.driver.v1.Statement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 */
package org.apache.flink.streaming.connectors.neo4j;



/**
 * @author Alberto De Lazzari
 *
 */
public class Neo4JSink<T> extends RichSinkFunction<T> {

	/**
	 * The default serial version UID
	 */
	private static final long serialVersionUID = 1L;

	private static final Logger LOGGER = LoggerFactory.getLogger(Neo4JSink.class);

	/**
	 * Neo4J driver should be not serialized
	 */
	protected transient Neo4JDriverWrapper driver;

	/**
	 * The mapping strategy that we use to map data from Flink to Neo4J
	 * 
	 * @see Neo4JDeserializationMappingStrategy
	 */
	private Neo4JDeserializationMappingStrategy<T, DeserializationMapper<T>> mappingStrategy;