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

import java.io.Serializable;
import java.util.Map;

import org.neo4j.driver.v1.AuthToken;
import org.neo4j.driver.v1.AuthTokens;
import org.neo4j.driver.v1.Config;
import org.neo4j.driver.v1.Config.EncryptionLevel;
import org.neo4j.driver.v1.Driver;
import org.neo4j.driver.v1.GraphDatabase;
import org.neo4j.driver.v1.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Alberto De Lazzari
 *
 */
public class Neo4JDriverWrapper implements Serializable {

	private static final long serialVersionUID = 1L;

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

	public static final String USERNAME_PARAM = "neo4j.auth.username";

	public static final String PASSWORD_PARAM = "neo4j.auth.password";

	public static final String URL = "neo4j.url";

	public static final String SESSION_LIVENESS_TIMEOUT = "neo4j.session.livetimeout";

	/**
	 * The default session liveness timeout as it's defined by the ConfigBuilder class.
	 * 
	 * @see org.neo4j.driver.v1.Config.ConfigBuilder
	 */
	private final String DEFAULT_SESSION_LIVENESS_TIMEOUT = "200";

	private Map<String, String> parameters;

	protected transient Driver driver;

	/**
	 * The defaut constructor. Always pass the connection url, username and
	 * password
	 * 
	 * @param parameters
	 *            the connection parameters
	 */
	public Neo4JDriverWrapper(final Map<String, String> parameters) {
		// We want to ensure that all the mandatory parameters are defined
		boolean urlDefined = parameters.containsKey(URL);
		assert urlDefined;
		boolean usernameDefined = parameters.containsKey(USERNAME_PARAM);
		assert usernameDefined;
		boolean passwordDefined = parameters.containsKey(PASSWORD_PARAM);
		assert passwordDefined;

		this.parameters = parameters;
		this.initDriver();
	}

	/**
	 * Init the internal Neo4J driver
	 * 
	 * @see org.neo4j.driver.v1.Driver
	 */
	protected void initDriver() {
		String url = parameters.get(URL);
		String username = parameters.get(USERNAME_PARAM);
		String password = parameters.get(PASSWORD_PARAM);
		String timeout = parameters.getOrDefault(SESSION_LIVENESS_TIMEOUT, DEFAULT_SESSION_LIVENESS_TIMEOUT);

		AuthToken authToken = AuthTokens.basic(username, password);
		LOGGER.debug("Basic authentication token with username {}", username);

		Config config = Config.build().withSessionLivenessCheckTimeout(getLongValue(timeout))
				.withEncryptionLevel(EncryptionLevel.NONE).toConfig();
		driver = GraphDatabase.driver(url, authToken, config);
		LOGGER.debug("A driver has been created to {}", url);
	}

	/**
	 * 
	 * @param longValue
	 *            a long number as a string
	 * @return
	 */
	private long getLongValue(String longValue) {
		return Long.valueOf(longValue);
	}

	/**
	 * Establish a session
	 * 
	 * @see org.neo4j.driver.v1.Session
	 * 
	 * @return a Neo4J session
	 */
	public Session session() {
		return driver.session();
	}

	/**
	 * Close all the resources assigned to the internal Neo4J Driver
	 * 
	 * @see org.neo4j.driver.v1.Driver
	 */
	public void close() {
		driver.close();
	}
}
@@ -1,16 +1,4 @@
-/**%0A * %0A */%0A
 pack
@@ -674,16 +674,57 @@
 lass);%0A%0A
+%09/**%0A%09 * The username property key.%0A%09 */%0A
 %09public 
@@ -780,24 +780,65 @@
 username%22;%0A%0A
+%09/**%0A%09 * The password property key.%0A%09 */%0A
 %09public stat
@@ -899,54 +899,160 @@
 ;%0A%0A%09
-public static final String URL = %22neo4j.url%22;%0A
+/**%0A%09 * The Neo4J server url property key.%0A%09 */%0A%09public static final String URL = %22neo4j.url%22;%0A%0A%09/**%0A%09 * The session liveness timeout property key.%0A%09 */
 %0A%09pu
@@ -1223,21 +1223,16 @@
  class.%0A
-%09 * %0A
 %09 * @see
@@ -1287,16 +1287,23 @@
 private 
+static 
 final St
@@ -1355,48 +1355,174 @@
 ;%0A%0A%09
-private Map%3CString, String%3E parameters;%0A
+/**%0A%09 * The configuration parameters.%0A%09 */%0A%09private Map%3CString, String%3E parameters;%0A%0A%09/**%0A%09 * The internal Neo4J Driver.%0A%09 * %0A%09 * @see org.neo4j.driver.v1.Driver%0A%09 */
 %0A%09pr
@@ -2209,24 +2209,25 @@
 Neo4J driver
+.
 %0A%09 * %0A%09 * @s
@@ -2272,16 +2272,22 @@
 otected 
+final 
 void ini
@@ -2944,24 +2944,83 @@
 );%0A%09%7D%0A%0A%09/**%0A
+%09 * Get the corresponding long value for the given string.%0A
 %09 * %0A%09 * @pa
@@ -3085,16 +3085,31 @@
  @return
+ the long value
 %0A%09 */%0A%09p
@@ -3133,16 +3133,22 @@
 ngValue(
+final 
 String l
@@ -3221,24 +3221,25 @@
 sh a session
+.
 %0A%09 * %0A%09 * @s
@@ -3269,21 +3269,16 @@
 Session%0A
-%09 * %0A
 %09 * @ret
@@ -3310,16 +3310,22 @@
 %09public 
+final 
 Session 
@@ -3433,24 +3433,25 @@
 Neo4J Driver
+.
 %0A%09 * %0A%09 * @s
@@ -3493,16 +3493,22 @@
 %09public 
+final 
 void clo
@@ -3536,8 +3536,9 @@
 ();%0A%09%7D%0A%7D
+%0A

Check style