// Path: src/test/java/org/apache/flink/embedded/neo4j/Neo4JBaseEmbeddedConfig.java
// public class Neo4JBaseEmbeddedConfig {
// 
// 	public static final String DEFAULT_URL = "bolt://localhost:7687";
// 
// 	public static final String DEFAULT_USERNAME = "neo4j";
// 
// 	public static final String DEFAULT_PASSWORD = "password";
// 	
// 	@ClassRule
// 	public static Neo4jRule neo4jRule;
// 	
// 	public static Neo4JDriverWrapper driverWrapper;
// 	
// 	public static Map<String, String> neo4JConfig;
// 
// 	static {
// 		neo4jRule = new Neo4jRule().withFixture("create (i:Item {description:'an item'})");
// 	}
// 
// 	@Before
// 	public void init() {
// 		neo4JConfig = new HashMap<String, String>();
// 		neo4JConfig.put(Neo4JDriverWrapper.URL, neo4jRule.boltURI().toString());
// 		neo4JConfig.put(Neo4JDriverWrapper.USERNAME_PARAM, "neo4j");
// 		neo4JConfig.put(Neo4JDriverWrapper.PASSWORD_PARAM, "password");
// 		neo4JConfig.put(Neo4JDriverWrapper.SESSION_LIVENESS_TIMEOUT, "4000");
// 		
// 		driverWrapper = new Neo4JDriverWrapper(neo4JConfig);
// 	}
// }
// 
// 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.io.IOException;
import java.util.Map;
import org.apache.flink.embedded.neo4j.Neo4JBaseEmbeddedConfig;
import org.apache.flink.mapping.neo4j.DeserializationMapper;
import org.apache.flink.mapping.neo4j.Neo4JDeserializationMappingStrategy;
/**
 * 
 */
package org.apache.flink.batch.neo4j;



/**
 * @author Alberto De Lazzari
 *
 */
public class Neo4JOutputFormatMock<T> extends Neo4JOutputFormat<T> {

	private static final long serialVersionUID = 1L;

	public Neo4JOutputFormatMock(Neo4JDeserializationMappingStrategy<T, DeserializationMapper<T>> mappingStrategy,
			Map<String, String> config) {
		super(mappingStrategy, config);
	}

	@Override
	public void open(int taskNumber, int numTasks) throws IOException {
		super.open(taskNumber, numTasks);
		// We use a static driver wrapper with an embedded Neo4J instance
// Path: src/test/java/org/apache/flink/embedded/neo4j/Neo4JBaseEmbeddedConfig.java
// public class Neo4JBaseEmbeddedConfig {
// 
// 	public static final String DEFAULT_URL = "bolt://localhost:7687";
// 
// 	public static final String DEFAULT_USERNAME = "neo4j";
// 
// 	public static final String DEFAULT_PASSWORD = "password";
// 	
// 	@ClassRule
// 	public static Neo4jRule neo4jRule;
// 	
// 	public static Neo4JDriverWrapper driverWrapper;
// 	
// 	public static Map<String, String> neo4JConfig;
// 
// 	static {
// 		neo4jRule = new Neo4jRule().withFixture("create (i:Item {description:'an item'})");
// 	}
// 
// 	@Before
// 	public void init() {
// 		neo4JConfig = new HashMap<String, String>();
// 		neo4JConfig.put(Neo4JDriverWrapper.URL, neo4jRule.boltURI().toString());
// 		neo4JConfig.put(Neo4JDriverWrapper.USERNAME_PARAM, "neo4j");
// 		neo4JConfig.put(Neo4JDriverWrapper.PASSWORD_PARAM, "password");
// 		neo4JConfig.put(Neo4JDriverWrapper.SESSION_LIVENESS_TIMEOUT, "4000");
// 		
// 		driverWrapper = new Neo4JDriverWrapper(neo4JConfig);
// 	}
// }
// 
// 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/test/java/org/apache/flink/batch/neo4j/Neo4JOutputFormatMock.java
import java.io.IOException;
import java.util.Map;
import org.apache.flink.embedded.neo4j.Neo4JBaseEmbeddedConfig;
import org.apache.flink.mapping.neo4j.DeserializationMapper;
import org.apache.flink.mapping.neo4j.Neo4JDeserializationMappingStrategy;

/**
 * 
 */
package org.apache.flink.batch.neo4j;



/**
 * @author Alberto De Lazzari
 *
 */
public class Neo4JOutputFormatMock<T> extends Neo4JOutputFormat<T> {

	private static final long serialVersionUID = 1L;

	public Neo4JOutputFormatMock(Neo4JDeserializationMappingStrategy<T, DeserializationMapper<T>> mappingStrategy,
			Map<String, String> config) {
		super(mappingStrategy, config);
	}

	@Override
	public void open(int taskNumber, int numTasks) throws IOException {
		super.open(taskNumber, numTasks);
		// We use a static driver wrapper with an embedded Neo4J instance
		driver = Neo4JBaseEmbeddedConfig.driverWrapper;