Added support for "non blocking" mode ("mysqlbinlog --to-last-log" equivalent)

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

Author: shyiko

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

@@ -85,6 +85,7 @@ public class BinaryLogClient implements BinaryLogClientMXBean {
     private final String username;
     private final String password;
 
+    private boolean blocking = true;
     private long serverId = 65535;
     private volatile String binlogFilename;
     private volatile long binlogPosition = 4;
@@ -154,6 +155,17 @@ public BinaryLogClient(String hostname, int port, String schema, String username
         this.password = password;
     }
 
+    public boolean isBlocking() {
+        return blocking;
+    }
+
+    /**
+     * @param blocking blocking mode. If set to false - BinaryLogClient will disconnect after the last event.
+     */
+    public void setBlocking(boolean blocking) {
+        this.blocking = blocking;
+    }
+
     /**
      * @return server id (65535 by default)
      * @see #setServerId(long)
@@ -369,8 +381,12 @@ public void connect() throws IOException {
         connected = true;
         connectionId = greetingPacket.getThreadId();
         if (logger.isLoggable(Level.INFO)) {
-            logger.info("Connected to " + hostname + ":" + port + " at " + binlogFilename + "/" + binlogPosition +
-                " (sid:" + serverId + ", cid:" + connectionId + ")");
+            String position;
+            synchronized (gtidSetAccessLock) {
+                position = gtidSet != null ? gtidSet.toString() : binlogFilename + "/" + binlogPosition;
+            }
+            logger.info("Connected to " + hostname + ":" + port + " at " + position +
+                " (" + (blocking ? "sid:" + serverId + ", " : "") + "cid:" + connectionId + ")");
         }
         synchronized (lifecycleListeners) {
             for (LifecycleListener lifecycleListener : lifecycleListeners) {
@@ -401,6 +417,7 @@ private GreetingPacket receiveGreeting() throws IOException {
     }
 
     private void requestBinaryLogStream() throws IOException {
+        long serverId = blocking ? this.serverId : 0; // http://bugs.mysql.com/bug.php?id=71178
         Command dumpBinaryLogCommand;
         synchronized (gtidSetAccessLock) {
             if (gtidSet != null) {
@@ -603,6 +620,9 @@ private void listenForEventPackets() throws IOException {
                     throw new ServerException(errorPacket.getErrorMessage(), errorPacket.getErrorCode(),
                         errorPacket.getSqlState());
                 }
+                if (marker == (byte) 0xFE && !blocking) {
+                    break;
+                }
                 Event event;
                 try {
                     event = eventDeserializer.nextEvent(packetLength == MAX_PACKET_LENGTH ?