SSH native implemntation; test rewrite.

Author: bokysan

README.md

@@ -136,6 +136,4 @@ Download artifacts directly from CircleCI: https://circleci.com/gh/bokysan/jdbc-
 
 ## TODO
 
-- Make better use of connection pooling. If the system tries to open the same connection towards the same server, this implementation will reestablish
-another connection. 
-- SSH connection should be persistent and kept alive only until until the last connection is closed. Currently their livespan is connected to the life of JVM. 
+- Discover why Derby test for SSH native connection is failing.

pom.xml

@@ -91,11 +91,18 @@
             <version>${slf4j.version}</version>
         </dependency>
 
-        <dependency>
+        <!--dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-simple</artifactId>
             <version>${slf4j.version}</version>
 			  <scope>test</scope>
+        </dependency-->
+
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <version>${logback.version}</version>
+            <scope>test</scope>
         </dependency>
 
         <dependency>

src/main/java/com/cekrlic/jdbc/ssh/tunnel/AbstractSshJDriver.java

@@ -2,15 +2,19 @@
 
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.sql.*;
 import java.util.Enumeration;
+import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Logger;
 
 public abstract class AbstractSshJDriver implements Driver {
 	private static final org.slf4j.Logger log = LoggerFactory.getLogger(AbstractSshJDriver.class);
 
-	protected AbstractTunnel tunnel;
+	static final Map<String, AtomicReference<AbstractTunnel>> tunnelList = new ConcurrentHashMap<>();
 
 	abstract String getDriverPrefix();
 
@@ -19,11 +23,6 @@ public boolean acceptsURL(String url) throws SQLException {
 		return (url != null && url.startsWith(getDriverPrefix()));
 	}
 
-	@Override
-	public Connection connect(String url, Properties info) throws SQLException {
-		return null;
-	}
-
 	private Driver findDriver(String url) throws SQLException {
 		Driver realDriver = null;
 
@@ -47,7 +46,7 @@ private Driver findDriver(String url) throws SQLException {
 		return realDriver;
 	}
 
-	protected Connection getRealConnection(Properties info, String originalUrl, String newHost, String newPort) throws SQLException {
+	protected Connection getRealConnection(AbstractTunnel tunnel, Properties info, String originalUrl, String newHost, String newPort) throws SQLException {
 		originalUrl = originalUrl.replaceAll("\\{\\{[hH][oO][sS][tT]\\}\\}", newHost);
 		originalUrl = originalUrl.replaceAll("\\{\\{[pP][oO][rR][tT]\\}\\}", newPort);
 
@@ -100,4 +99,52 @@ public boolean jdbcCompliant() {
 	public Logger getParentLogger() throws SQLFeatureNotSupportedException {
 		throw new SQLFeatureNotSupportedException("Feature not supported");
 	}
+
+	@Override
+	public Connection connect(String url, Properties info) throws SQLException {
+		ConnectionData d = verifyConnection(url);
+		// Not our connection URL, skip further integration
+		if(d == null) {
+			return null;
+		}
+
+		AbstractTunnel tunnel = getTunnel(url, d);
+		tunnel.ensureStarted();
+		Connection c = getRealConnection(tunnel, info, d.getForwardingUrl(), tunnel.getLocalHost(), tunnel.getLocalPort());
+		return new SshConnection(tunnel, c);
+	}
+
+	@SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
+	public AbstractTunnel getTunnel(String url, ConnectionData d) throws SQLException {
+		AbstractTunnel tunnel;
+		try {
+			AtomicReference<AbstractTunnel> a = tunnelList.get(url);
+			if(a == null) {
+				tunnel = newTunnel(d);
+				log.info("No tunnel for {}, created a new tunnel: {}", url, tunnel);
+				a = new AtomicReference<>(tunnel);
+				synchronized (a) {
+					tunnelList.put(url, a);
+					tunnel.start();
+				}
+			} else {
+				synchronized (a) {
+					tunnel = a.get();
+					if(tunnel.isStopped()) {
+						tunnel = newTunnel(d);
+						log.info("Tunnel stopped for {}, created a new tunnel: {}", url, tunnel);
+						a.set(tunnel);
+					} else {
+						log.debug("Reusing connection {}:{}", tunnel.getLocalHost(), tunnel.getLocalPort());
+					}
+				}
+			}
+		} catch (IOException e) {
+			throw new SQLException(e);
+		}
+		return tunnel;
+	}
+
+	protected abstract AbstractTunnel newTunnel(ConnectionData d) throws IOException, SQLException;
+
 }

src/main/java/com/cekrlic/jdbc/ssh/tunnel/AbstractTunnel.java

@@ -24,6 +24,7 @@ public abstract class AbstractTunnel {
 	public static final String REMOTE = "remote";
 	public static final String DRIVERS = "drivers";
 
+	protected AtomicInteger connectionCount = new AtomicInteger(0);
 	protected AtomicInteger localPort = new AtomicInteger(20000 + new java.util.Random().nextInt(100));
 	protected String localHost;
 
@@ -132,5 +133,19 @@ private static boolean isPortOpen(String ip, int port) {
 	/**
 	 * Stop the tunnel
 	 */
-	abstract void stop();
+	abstract void stop(String reason);
+
+	abstract void ensureStarted() throws SQLException;
+
+	abstract boolean isStopped();
+
+	public void remove(SshConnection sshConnection) {
+		if(connectionCount.decrementAndGet()==0) {
+			this.stop("No connections remaining open.");
+		}
+	}
+
+	public void add(SshConnection sshConnection) {
+		connectionCount.incrementAndGet();
+	}
 }