// Path: src/main/java/org/apache/flink/mapping/neo4j/Neo4JSerializationMappingStrategy.java
// public class Neo4JSerializationMappingStrategy<T, M extends SerializationMapper<T>> implements Serializable {
// 
// 	private static final long serialVersionUID = 1L;
// 
// 	/**
// 	 * The templated cypher query that will be executed on Neo4J
// 	 */
// 	private String templateStatement;
// 
// 	/**
// 	 * 
// 	 */
// 	private M mapper;
// 
// 	/**
// 	 * 
// 	 * @param templateStatement the cypher statement template
// 	 * @param mapper an actual SerializationMapper implementation
// 	 */
// 	public Neo4JSerializationMappingStrategy(final String templateStatement, M mapper) {
// 		this.templateStatement = templateStatement;
// 		this.mapper = mapper;
// 	}
// 
// 	/**
// 	 * 
// 	 * @return the cypher statement template
// 	 */
// 	public Statement getStatement() {
// 		return new Statement(templateStatement);
// 	}
// 
// 	/**
// 	 * Maps a single result record into a T object
// 	 * 
// 	 * @param record a record that will be returned by the cypher query
// 	 * @return an object of type T
// 	 */
// 	public T map(Record record) {
// 		return this.mapper.serialize(record.asMap());
// 	}
// }
// 
// Path: src/main/java/org/apache/flink/mapping/neo4j/SerializationMapper.java
// @FunctionalInterface
// public interface SerializationMapper<T> extends Mapper {
// 
// 	/**
// 	 * Convert a record (that is returned by the cypher query) into an object of
// 	 * type T.
// 	 * 
// 	 * Tips: Maybe record should be of type org.neo4j.driver.v1.Record
// 	 * 
// 	 * @param record
// 	 *            a record (as a Map) that will be returned by the cypher query
// 	 * @return an item of type T that is a transformation of a record into an
// 	 *         item T
// 	 */
// 	public T serialize(Map<String, Object> record);
// }


import java.util.Map;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.mapping.neo4j.Neo4JSerializationMappingStrategy;
import org.apache.flink.mapping.neo4j.SerializationMapper;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.neo4j.driver.v1.Record;
import org.neo4j.driver.v1.Session;
import org.neo4j.driver.v1.Statement;
import org.neo4j.driver.v1.StatementResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 
 */
package org.apache.flink.streaming.connectors.neo4j;



/**
 * @author Alberto De Lazzari
 *
 */
public class Neo4JSource<T> extends RichSourceFunction<T> {

	private static final long serialVersionUID = 1L;

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

	protected transient Neo4JDriverWrapper driver;

	/**
	 * The mapping strategy that we use to map data from Flink to Neo4J
	 * 
	 * @see Neo4JSerializationMappingStrategy
	 */
// Path: src/main/java/org/apache/flink/mapping/neo4j/Neo4JSerializationMappingStrategy.java
// public class Neo4JSerializationMappingStrategy<T, M extends SerializationMapper<T>> implements Serializable {
// 
// 	private static final long serialVersionUID = 1L;
// 
// 	/**
// 	 * The templated cypher query that will be executed on Neo4J
// 	 */
// 	private String templateStatement;
// 
// 	/**
// 	 * 
// 	 */
// 	private M mapper;
// 
// 	/**
// 	 * 
// 	 * @param templateStatement the cypher statement template
// 	 * @param mapper an actual SerializationMapper implementation
// 	 */
// 	public Neo4JSerializationMappingStrategy(final String templateStatement, M mapper) {
// 		this.templateStatement = templateStatement;
// 		this.mapper = mapper;
// 	}
// 
// 	/**
// 	 * 
// 	 * @return the cypher statement template
// 	 */
// 	public Statement getStatement() {
// 		return new Statement(templateStatement);
// 	}
// 
// 	/**
// 	 * Maps a single result record into a T object
// 	 * 
// 	 * @param record a record that will be returned by the cypher query
// 	 * @return an object of type T
// 	 */
// 	public T map(Record record) {
// 		return this.mapper.serialize(record.asMap());
// 	}
// }
// 
// Path: src/main/java/org/apache/flink/mapping/neo4j/SerializationMapper.java
// @FunctionalInterface
// public interface SerializationMapper<T> extends Mapper {
// 
// 	/**
// 	 * Convert a record (that is returned by the cypher query) into an object of
// 	 * type T.
// 	 * 
// 	 * Tips: Maybe record should be of type org.neo4j.driver.v1.Record
// 	 * 
// 	 * @param record
// 	 *            a record (as a Map) that will be returned by the cypher query
// 	 * @return an item of type T that is a transformation of a record into an
// 	 *         item T
// 	 */
// 	public T serialize(Map<String, Object> record);
// }


// Path: src/main/java/org/apache/flink/streaming/connectors/neo4j/Neo4JSource.java
import java.util.Map;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.mapping.neo4j.Neo4JSerializationMappingStrategy;
import org.apache.flink.mapping.neo4j.SerializationMapper;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.neo4j.driver.v1.Record;
import org.neo4j.driver.v1.Session;
import org.neo4j.driver.v1.Statement;
import org.neo4j.driver.v1.StatementResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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



/**
 * @author Alberto De Lazzari
 *
 */
public class Neo4JSource<T> extends RichSourceFunction<T> {

	private static final long serialVersionUID = 1L;

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

	protected transient Neo4JDriverWrapper driver;

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