// 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/mapping/neo4j/SimpleValuesMapper.java
// public class SimpleValuesMapper implements DeserializationMapper<Tuple2<String, Integer>> {
// 
// 	/**
// 	 * 
// 	 */
// 	private static final long serialVersionUID = 1681921169214823084L;
// 
// 	public SimpleValuesMapper() {
// 		super();
// 	}
// 
// 	@Override
// 	public Map<String, Object> deserialize(Tuple2<String, Integer> item) {
// 		HashMap<String, Object> values = new HashMap<String, Object>();
// 
// 		values.put("t1", item.f1);
// 		values.put("t2", item.f0);
// 
// 		return values;
// 	}
// 
// }


import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.ArrayList;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.embedded.neo4j.Neo4JBaseEmbeddedConfig;
import org.apache.flink.mapping.neo4j.DeserializationMapper;
import org.apache.flink.mapping.neo4j.Neo4JDeserializationMappingStrategy;
import org.apache.flink.mapping.neo4j.SimpleValuesMapper;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 
 */
package org.apache.flink.batch.neo4j;




/**
 * @author Alberto De Lazzari
 *
 */
public class Neo4JOutputFormatTest extends Neo4JBaseEmbeddedConfig {

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

	private static final ArrayList<Tuple2<String, Integer>> collection = new ArrayList<>(20);

	static {
		for (int i = 0; i < 20; i++) {
			collection.add(new Tuple2<>("neo4j-" + i, i));
		}
	}

	@Test
	public void testOutputFormatCount() throws Exception {
		ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();

		String statementTemplate = "MERGE (tuple:Tuple {name: {t1}, index: {t2}}) RETURN tuple";
		DeserializationMapper<Tuple2<String, Integer>> mapper = new SimpleValuesMapper();
// 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/mapping/neo4j/SimpleValuesMapper.java
// public class SimpleValuesMapper implements DeserializationMapper<Tuple2<String, Integer>> {
// 
// 	/**
// 	 * 
// 	 */
// 	private static final long serialVersionUID = 1681921169214823084L;
// 
// 	public SimpleValuesMapper() {
// 		super();
// 	}
// 
// 	@Override
// 	public Map<String, Object> deserialize(Tuple2<String, Integer> item) {
// 		HashMap<String, Object> values = new HashMap<String, Object>();
// 
// 		values.put("t1", item.f1);
// 		values.put("t2", item.f0);
// 
// 		return values;
// 	}
// 
// }


// Path: src/test/java/org/apache/flink/batch/neo4j/Neo4JOutputFormatTest.java
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.ArrayList;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.embedded.neo4j.Neo4JBaseEmbeddedConfig;
import org.apache.flink.mapping.neo4j.DeserializationMapper;
import org.apache.flink.mapping.neo4j.Neo4JDeserializationMappingStrategy;
import org.apache.flink.mapping.neo4j.SimpleValuesMapper;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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




/**
 * @author Alberto De Lazzari
 *
 */
public class Neo4JOutputFormatTest extends Neo4JBaseEmbeddedConfig {

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

	private static final ArrayList<Tuple2<String, Integer>> collection = new ArrayList<>(20);

	static {
		for (int i = 0; i < 20; i++) {
			collection.add(new Tuple2<>("neo4j-" + i, i));
		}
	}

	@Test
	public void testOutputFormatCount() throws Exception {
		ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();

		String statementTemplate = "MERGE (tuple:Tuple {name: {t1}, index: {t2}}) RETURN tuple";
		DeserializationMapper<Tuple2<String, Integer>> mapper = new SimpleValuesMapper();
		Neo4JDeserializationMappingStrategy<Tuple2<String, Integer>, DeserializationMapper<Tuple2<String, Integer>>> mappingStrategy = new Neo4JDeserializationMappingStrategy<Tuple2<String, Integer>, DeserializationMapper<Tuple2<String, Integer>>>(