Switched BinaryLogClientMXBean.connect from blocking to non-blocking variant

Author: shyiko

src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java

@@ -271,7 +271,7 @@ public void run() {
                                 if (isConnected()) {
                                     disconnectChannel();
                                 }
-                                connect(keepAliveConnectTimeout, TimeUnit.MILLISECONDS);
+                                connect(keepAliveConnectTimeout);
                             } catch (Exception ce) {
                                 if (logger.isLoggable(Level.WARNING)) {
                                     logger.warning("Failed to restore connection to " + hostname + ":" + port +
@@ -306,13 +306,12 @@ protected boolean isKeepAliveThreadRunning() {
 
     /**
      * Connect to the replication stream in a separate thread.
-     * @param timeout connect timeout
-     * @param timeUnit timeout unit
+     * @param timeoutInMilliseconds timeout in milliseconds
      * @throws AuthenticationException in case of failed authentication
      * @throws IOException if anything goes wrong while trying to connect
      * @throws TimeoutException if client wasn't able to connect in the requested period of time
      */
-    public void connect(long timeout, TimeUnit timeUnit) throws IOException, TimeoutException {
+    public void connect(long timeoutInMilliseconds) throws IOException, TimeoutException {
         final CountDownLatch countDownLatch = new CountDownLatch(1);
         AbstractLifecycleListener connectListener = new AbstractLifecycleListener() {
             @Override
@@ -338,7 +337,7 @@ public void run() {
         thread.start();
         boolean started = false;
         try {
-            started = countDownLatch.await(timeout, timeUnit);
+            started = countDownLatch.await(timeoutInMilliseconds, TimeUnit.MILLISECONDS);
         } catch (InterruptedException e) {
             if (logger.isLoggable(Level.WARNING)) {
                 logger.log(Level.WARNING, e.getMessage());
@@ -349,7 +348,7 @@ public void run() {
             throw exceptionReference.get();
         }
         if (!started) {
-            throw new TimeoutException("BinaryLogClient was unable to connect in " + timeUnit.toMillis(timeout) + "ms");
+            throw new TimeoutException("BinaryLogClient was unable to connect in " + timeoutInMilliseconds + "ms");
         }
     }
 

src/main/java/com/github/shyiko/mysql/binlog/jmx/BinaryLogClientMXBean.java

@@ -16,6 +16,7 @@
 package com.github.shyiko.mysql.binlog.jmx;
 
 import java.io.IOException;
+import java.util.concurrent.TimeoutException;
 
 /**
  * @author <a href="mailto:stanley.shyiko@gmail.com">Stanley Shyiko</a>
@@ -26,7 +27,7 @@ public interface BinaryLogClientMXBean {
     void setBinlogFilename(String binlogFilename);
     long getBinlogPosition();
     void setBinlogPosition(long binlogPosition);
-    void connect() throws IOException;
+    void connect(long timeoutInMilliseconds) throws IOException, TimeoutException;
     boolean isConnected();
     void disconnect() throws IOException;
 

src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientIntegrationTest.java

@@ -93,7 +93,7 @@ public void setUp() throws Exception {
         client.registerEventListener(new TraceEventListener());
         client.registerEventListener(eventListener = new CountDownEventListener());
         client.registerLifecycleListener(new TraceLifecycleListener());
-        client.connect(3, TimeUnit.SECONDS);
+        client.connect(DEFAULT_TIMEOUT);
         master.execute(new Callback<Statement>() {
             @Override
             public void execute(Statement statement) throws SQLException {
@@ -369,7 +369,7 @@ public void execute(Statement statement) throws SQLException {
                 eventListener.reset();
             }
         } finally {
-            client.connect(3, TimeUnit.SECONDS);
+            client.connect(DEFAULT_TIMEOUT);
         }
         eventListener.waitFor(WriteRowsEventData.class, 2, DEFAULT_TIMEOUT);
     }
@@ -388,7 +388,7 @@ public void execute(Statement statement) throws SQLException {
         client.disconnect();
         client.setBinlogFilename(binlogFilename);
         client.setBinlogPosition(binlogPosition);
-        client.connect(3, TimeUnit.SECONDS);
+        client.connect(DEFAULT_TIMEOUT);
         eventListener.waitFor(WriteRowsEventData.class, 1, DEFAULT_TIMEOUT);
     }
 
@@ -405,7 +405,7 @@ public void testAutomaticFailover() throws Exception {
                 clientOverProxy.setKeepAliveConnectTimeout(TimeUnit.SECONDS.toMillis(2));
                 clientOverProxy.registerEventListener(eventListener);
                 try {
-                    clientOverProxy.connect(3, TimeUnit.SECONDS);
+                    clientOverProxy.connect(DEFAULT_TIMEOUT);
                     eventListener.waitFor(EventType.FORMAT_DESCRIPTION, 1, DEFAULT_TIMEOUT);
                     assertTrue(clientOverProxy.isKeepAliveThreadRunning());
                     BinaryLogClient.LifecycleListener lifecycleListenerMock =
@@ -431,7 +431,7 @@ public void execute(Statement statement) throws SQLException {
                 }
                 assertFalse(clientOverProxy.isKeepAliveThreadRunning());
             } finally {
-                client.connect(3, TimeUnit.SECONDS);
+                client.connect(DEFAULT_TIMEOUT);
             }
         } finally {
             tcpReverseProxy.unbind();
@@ -474,7 +474,7 @@ public void run() {
     @Test(expectedExceptions = AuthenticationException.class)
     public void testAuthenticationFailsWhenNonExistingSchemaProvided() throws Exception {
         new BinaryLogClient(slave.hostname, slave.port, "mbcj_test_non_existing", slave.username, slave.password).
-            connect(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
+            connect(DEFAULT_TIMEOUT);
     }
 
     @Test
@@ -494,7 +494,7 @@ public void execute(Statement statement) throws SQLException {
         try {
             CountDownEventListener isolatedEventListener = new CountDownEventListener();
             isolatedClient.registerEventListener(isolatedEventListener);
-            isolatedClient.connect(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
+            isolatedClient.connect(DEFAULT_TIMEOUT);
             master.execute(new Callback<Statement>() {
                 @Override
                 public void execute(Statement statement) throws SQLException {

src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientTest.java

@@ -18,7 +18,6 @@
 import com.github.shyiko.mysql.binlog.jmx.BinaryLogClientStatistics;
 import org.testng.annotations.Test;
 
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import static org.testng.Assert.assertEquals;
@@ -62,7 +61,7 @@ public void testLifecycleListenersManagement() {
 
     @Test(expectedExceptions = TimeoutException.class)
     public void testConnectionTimeout() throws Exception {
-        new BinaryLogClient("_localhost_", 3306, "root", "mysql").connect(0, TimeUnit.MILLISECONDS);
+        new BinaryLogClient("_localhost_", 3306, "root", "mysql").connect(0);
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)