Added test to ensure that client can be disconnected even if no data can be read from channel (shyiko#154)

Author: shyiko

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

@@ -16,8 +16,19 @@
 package com.github.shyiko.mysql.binlog;
 
 import com.github.shyiko.mysql.binlog.jmx.BinaryLogClientStatistics;
+import com.github.shyiko.mysql.binlog.network.SocketFactory;
 import org.testng.annotations.Test;
 
+import java.io.EOFException;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import static org.testng.Assert.assertEquals;
@@ -69,4 +80,69 @@ public void testNullEventDeserializerIsNotAllowed() throws Exception {
         new BinaryLogClient("localhost", 3306, "root", "mysql").setEventDeserializer(null);
     }
 
+    @Test(timeOut = 15000)
+    public void testDisconnectWhileBlockedByFBRead() throws Exception {
+        final BinaryLogClient binaryLogClient = new BinaryLogClient("localhost", 33060, "root", "mysql");
+        final CountDownLatch readAttempted = new CountDownLatch(1);
+        binaryLogClient.setSocketFactory(new SocketFactory() {
+            @Override
+            public Socket createSocket() throws SocketException {
+                return new Socket() {
+
+                    @Override
+                    public InputStream getInputStream() throws IOException {
+                        return new FilterInputStream(super.getInputStream()) {
+
+                             @Override
+                            public int read(byte[] b, int off, int len) throws IOException {
+                                readAttempted.countDown();
+                                return super.read(b, off, len);
+                            }
+                        };
+                    }
+                };
+            }
+        });
+        binaryLogClient.setKeepAlive(false);
+        final CountDownLatch socketBound = new CountDownLatch(1);
+        new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    final ServerSocket serverSocket = new ServerSocket();
+                    try {
+                        serverSocket.bind(new InetSocketAddress("localhost", 33060));
+                        socketBound.countDown();
+                        serverSocket.accept(); // accept socket but do NOT send anything
+                        assertTrue(readAttempted.await(3000, TimeUnit.MILLISECONDS));
+                        Thread thread = new Thread(new Runnable() {
+                            @Override
+                            public void run() {
+                                try {
+                                    Thread.yield();
+                                    binaryLogClient.disconnect();
+                                } catch (IOException e) {
+                                    throw new RuntimeException(e);
+                                }
+                            }
+                        });
+                        thread.start();
+                        thread.join();
+                    } finally {
+                        serverSocket.close();
+                    }
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }
+        }).start();
+        assertTrue(socketBound.await(3000, TimeUnit.MILLISECONDS));
+        try {
+            binaryLogClient.connect();
+        } catch (IOException e) {
+           assertEquals(readAttempted.getCount(), 0);
+           assertTrue(e.getCause() instanceof EOFException);
+        }
+    }
+
 }