fix(BinaryLogClient): use writeBuffered exclusively & adjust packetNum for SSL switch auth

Author: glarwood

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

@@ -646,7 +646,7 @@ private GreetingPacket receiveGreeting() throws IOException {
     }
 
     private void enableHeartbeat() throws IOException {
-        channel.write(new QueryCommand("set @master_heartbeat_period=" + heartbeatInterval * 1000000));
+        channel.writeBuffered(new QueryCommand("set @master_heartbeat_period=" + heartbeatInterval * 1000000));
         byte[] statementResult = channel.read();
         if (statementResult[0] == (byte) 0xFF /* error */) {
             byte[] bytes = Arrays.copyOfRange(statementResult, 1, statementResult.length);
@@ -669,7 +669,7 @@ private void requestBinaryLogStream() throws IOException {
                 dumpBinaryLogCommand = new DumpBinaryLogCommand(serverId, binlogFilename, binlogPosition);
             }
         }
-        channel.write(dumpBinaryLogCommand);
+        channel.writeBuffered(dumpBinaryLogCommand);
     }
 
     private void ensureEventDataDeserializer(EventType eventType,
@@ -692,6 +692,8 @@ private void ensureEventDataDeserializer(EventType eventType,
     private void authenticate(GreetingPacket greetingPacket) throws IOException {
         int collation = greetingPacket.getServerCollation();
         int packetNumber = 1;
+
+        boolean usingSSLSocket = false;
         if (sslMode != SSLMode.DISABLED) {
             boolean serverSupportsSSL = (greetingPacket.getServerCapabilities() & ClientCapabilities.SSL) != 0;
             if (!serverSupportsSSL && (sslMode == SSLMode.REQUIRED || sslMode == SSLMode.VERIFY_CA ||
@@ -701,7 +703,7 @@ private void authenticate(GreetingPacket greetingPacket) throws IOException {
             if (serverSupportsSSL) {
                 SSLRequestCommand sslRequestCommand = new SSLRequestCommand();
                 sslRequestCommand.setCollation(collation);
-                channel.write(sslRequestCommand, packetNumber++);
+                channel.writeBuffered(sslRequestCommand, packetNumber++);
                 SSLSocketFactory sslSocketFactory =
                     this.sslSocketFactory != null ?
                         this.sslSocketFactory :
@@ -710,12 +712,13 @@ private void authenticate(GreetingPacket greetingPacket) throws IOException {
                             DEFAULT_VERIFY_CA_SSL_MODE_SOCKET_FACTORY;
                 channel.upgradeToSSL(sslSocketFactory,
                     sslMode == SSLMode.VERIFY_IDENTITY ? new TLSHostnameVerifier() : null);
+                usingSSLSocket = true;
             }
         }
         AuthenticateCommand authenticateCommand = new AuthenticateCommand(schema, username, password,
             greetingPacket.getScramble());
         authenticateCommand.setCollation(collation);
-        channel.write(authenticateCommand, packetNumber);
+        channel.writeBuffered(authenticateCommand, packetNumber);
         byte[] authenticationResult = channel.read();
         if (authenticationResult[0] != (byte) 0x00 /* ok */) {
             if (authenticationResult[0] == (byte) 0xFF /* error */) {
@@ -724,14 +727,14 @@ private void authenticate(GreetingPacket greetingPacket) throws IOException {
                 throw new AuthenticationException(errorPacket.getErrorMessage(), errorPacket.getErrorCode(),
                     errorPacket.getSqlState());
             } else if (authenticationResult[0] == (byte) 0xFE) {
-                switchAuthentication(authenticationResult);
+                switchAuthentication(authenticationResult, usingSSLSocket);
             } else {
                 throw new AuthenticationException("Unexpected authentication result (" + authenticationResult[0] + ")");
             }
         }
     }
 
-    private void switchAuthentication(byte[] authenticationResult) throws IOException {
+    private void switchAuthentication(byte[] authenticationResult, boolean usingSSLSocket) throws IOException {
         /*
             Azure-MySQL likes to tell us to switch authentication methods, even though
             we haven't advertised that we support any.  It uses this for some-odd
@@ -745,7 +748,7 @@ private void switchAuthentication(byte[] authenticationResult) throws IOExceptio
             String scramble = buffer.readZeroTerminatedString();
 
             Command switchCommand = new AuthenticateNativePasswordCommand(scramble, password);
-            channel.writeBuffered(switchCommand, 3);
+            channel.writeBuffered(switchCommand, (usingSSLSocket? 4 : 3));
             byte[] authResult = channel.read();
 
             if (authResult[0] != (byte) 0x00) {
@@ -787,7 +790,7 @@ public void run() {
                             connectionLost = System.currentTimeMillis() - eventLastSeen > keepAliveInterval;
                         } else {
                             try {
-                                channel.write(new PingCommand());
+                                channel.writeBuffered(new PingCommand());
                             } catch (IOException e) {
                                 connectionLost = true;
                             }
@@ -891,7 +894,7 @@ public boolean isConnected() {
     }
 
     private String fetchGtidPurged() throws IOException {
-        channel.write(new QueryCommand("show global variables like 'gtid_purged'"));
+        channel.writeBuffered(new QueryCommand("show global variables like 'gtid_purged'"));
         ResultSetRowPacket[] resultSet = readResultSet();
         if (resultSet.length != 0) {
             return resultSet[0].getValue(1).toUpperCase();
@@ -901,7 +904,7 @@ private String fetchGtidPurged() throws IOException {
 
     private void fetchBinlogFilenameAndPosition() throws IOException {
         ResultSetRowPacket[] resultSet;
-        channel.write(new QueryCommand("show master status"));
+        channel.writeBuffered(new QueryCommand("show master status"));
         resultSet = readResultSet();
         if (resultSet.length == 0) {
             throw new IOException("Failed to determine binlog filename/position");
@@ -912,7 +915,7 @@ private void fetchBinlogFilenameAndPosition() throws IOException {
     }
 
     private ChecksumType fetchBinlogChecksum() throws IOException {
-        channel.write(new QueryCommand("show global variables like 'binlog_checksum'"));
+        channel.writeBuffered(new QueryCommand("show global variables like 'binlog_checksum'"));
         ResultSetRowPacket[] resultSet = readResultSet();
         if (resultSet.length == 0) {
             return ChecksumType.NONE;
@@ -921,7 +924,7 @@ private ChecksumType fetchBinlogChecksum() throws IOException {
     }
 
     private void confirmSupportOfChecksum(ChecksumType checksumType) throws IOException {
-        channel.write(new QueryCommand("set @master_binlog_checksum= @@global.binlog_checksum"));
+        channel.writeBuffered(new QueryCommand("set @master_binlog_checksum= @@global.binlog_checksum"));
         byte[] statementResult = channel.read();
         if (statementResult[0] == (byte) 0xFF /* error */) {
             byte[] bytes = Arrays.copyOfRange(statementResult, 1, statementResult.length);

src/main/java/com/github/shyiko/mysql/binlog/network/protocol/PacketChannel.java

@@ -61,16 +61,6 @@ public byte[] read() throws IOException {
         return inputStream.read(length);
     }
 
-    public void write(Command command, int packetNumber) throws IOException {
-        byte[] body = command.toByteArray();
-        outputStream.writeInteger(body.length, 3); // packet length
-        outputStream.writeInteger(packetNumber, 1);
-        outputStream.write(body, 0, body.length);
-        // though it has no effect in case of default (underlying) output stream (SocketOutputStream),
-        // it may be necessary in case of non-default one
-        outputStream.flush();
-    }
-
     /*
        Azure's MySQL has bizarre network properties that force us to write an
        auth-response challenge in one shot, lest their hair catch on fire and
@@ -86,8 +76,8 @@ public void writeBuffered(Command command, int packetNumber) throws IOException
         socket.getOutputStream().write(buffer.toByteArray());
     }
 
-    public void write(Command command) throws IOException {
-        write(command, 0);
+    public void writeBuffered(Command command) throws IOException {
+        writeBuffered(command, 0);
     }
 
     public void upgradeToSSL(SSLSocketFactory sslSocketFactory, HostnameVerifier hostnameVerifier) throws IOException {