Added BinaryLogClient::gtidSetFallbackToPurged

Author: shyiko

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

@@ -132,6 +132,7 @@ public X509Certificate[] getAcceptedIssuers() {
 
     private GtidSet gtidSet;
     private final Object gtidSetAccessLock = new Object();
+    private boolean gtidSetFallbackToPurged;
 
     private EventDeserializer eventDeserializer = new EventDeserializer();
 
@@ -298,12 +299,15 @@ public String getGtidSet() {
 
     /**
      * @param gtidSet GTID set (can be an empty string).
-     * <p>NOTE #1: Any value but null will switch BinaryLogClient into a GTID mode (in which case GTID set will be
-     * updated with each incoming GTID event) as well as set binlogFilename to "" (empty string) (meaning
-     * BinaryLogClient will request events "outside of the set" <u>starting from the oldest known binlog</u>).
+     * <p>NOTE #1: Any value but null will switch BinaryLogClient into a GTID mode (this will also set binlogFilename
+     * to "" (provided it's null) forcing MySQL to send events starting from the oldest known binlog (keep in mind
+     * that connection will fail if gtid_purged is anything but empty (unless
+     * {@link #setGtidSetFallbackToPurged(boolean)} is set to true))).
      * <p>NOTE #2: {@link #setBinlogFilename(String)} and {@link #setBinlogPosition(long)} can be used to specify the
      * exact position from which MySQL server should start streaming events (taking into account GTID set).
+     * <p>NOTE #3: GTID set is automatically updated with each incoming GTID event (provided GTID mode is on).
      * @see #getGtidSet()
+     * @see #setGtidSetFallbackToPurged(boolean)
      */
     public void setGtidSet(String gtidSet) {
         if (gtidSet != null && this.binlogFilename == null) {
@@ -314,6 +318,21 @@ public void setGtidSet(String gtidSet) {
         }
     }
 
+    /**
+     * @see #setGtidSetFallbackToPurged(boolean)
+     */
+    public boolean isGtidSetFallbackToPurged() {
+        return gtidSetFallbackToPurged;
+    }
+
+    /**
+     * @param gtidSetFallbackToPurged true if gtid_purged should be used as a fallback when gtidSet is set to "" and
+     * MySQL server has purged some of the binary logs, false otherwise (default).
+     */
+    public void setGtidSetFallbackToPurged(boolean gtidSetFallbackToPurged) {
+        this.gtidSetFallbackToPurged = gtidSetFallbackToPurged;
+    }
+
     /**
      * @return true if "keep alive" thread should be automatically started (default), false otherwise.
      * @see #setKeepAlive(boolean)
@@ -474,6 +493,13 @@ public void connect() throws IOException {
                 GreetingPacket greetingPacket = receiveGreeting();
                 authenticate(greetingPacket);
                 connectionId = greetingPacket.getThreadId();
+                if ("".equals(binlogFilename)) {
+                    synchronized (gtidSetAccessLock) {
+                        if (gtidSet != null && "".equals(gtidSet.toString()) && gtidSetFallbackToPurged) {
+                            gtidSet = new GtidSet(fetchGtidPurged());
+                        }
+                    }
+                }
                 if (binlogFilename == null) {
                     fetchBinlogFilenameAndPosition();
                 }
@@ -800,6 +826,15 @@ public boolean isConnected() {
         return connected;
     }
 
+    private String fetchGtidPurged() throws IOException {
+        channel.write(new QueryCommand("show global variables like 'gtid_purged'"));
+        ResultSetRowPacket[] resultSet = readResultSet();
+        if (resultSet.length != 0) {
+            return resultSet[0].getValue(1).toUpperCase();
+        }
+        return "";
+    }
+
     private void fetchBinlogFilenameAndPosition() throws IOException {
         ResultSetRowPacket[] resultSet;
         channel.write(new QueryCommand("show master status"));