diff --git a/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientGTIDIntegrationTest.java b/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientGTIDIntegrationTest.java index df033afa..15675961 100644 --- a/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientGTIDIntegrationTest.java +++ b/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientGTIDIntegrationTest.java @@ -15,25 +15,26 @@ */ package com.github.shyiko.mysql.binlog; -import com.github.shyiko.mysql.binlog.event.QueryEventData; -import com.github.shyiko.mysql.binlog.event.XidEventData; -import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer; -import org.testng.annotations.AfterClass; -import org.testng.annotations.BeforeClass; -import org.testng.annotations.Test; +import static org.testng.Assert.assertNotEquals; +import static org.testng.AssertJUnit.assertNotNull; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; import java.util.concurrent.TimeUnit; -import static org.testng.Assert.assertNotEquals; -import static org.testng.AssertJUnit.assertNotNull; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import com.github.shyiko.mysql.binlog.event.QueryEventData; +import com.github.shyiko.mysql.binlog.event.XidEventData; +import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer; /** * @author Ben Osheroff */ -public class BinaryLogClientGTIDIntegrationTest extends BinaryLogClientIntegrationTest { +public class BinaryLogClientGTIDIntegrationTest extends BinaryLogClientIntegrationTestBase { @BeforeClass private void enableGTID() throws SQLException { diff --git a/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientIntegrationTest.java b/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientIntegrationTest.java index 71dfb524..4a2a3dd5 100644 --- a/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientIntegrationTest.java +++ b/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientIntegrationTest.java @@ -15,35 +15,20 @@ */ package com.github.shyiko.mysql.binlog; -import com.github.shyiko.mysql.binlog.event.ByteArrayEventData; -import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; -import com.github.shyiko.mysql.binlog.event.Event; -import com.github.shyiko.mysql.binlog.event.EventData; -import com.github.shyiko.mysql.binlog.event.EventHeaderV4; -import com.github.shyiko.mysql.binlog.event.EventType; -import com.github.shyiko.mysql.binlog.event.QueryEventData; -import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData; -import com.github.shyiko.mysql.binlog.event.WriteRowsEventData; -import com.github.shyiko.mysql.binlog.event.deserialization.ByteArrayEventDataDeserializer; -import com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializationException; -import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer; -import com.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer; -import com.github.shyiko.mysql.binlog.event.deserialization.QueryEventDataDeserializer; -import com.github.shyiko.mysql.binlog.io.BufferedSocketInputStream; -import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream; -import com.github.shyiko.mysql.binlog.network.AuthenticationException; -import com.github.shyiko.mysql.binlog.network.ServerException; -import com.github.shyiko.mysql.binlog.network.SocketFactory; -import org.mockito.InOrder; -import org.testng.SkipException; -import org.testng.annotations.AfterClass; -import org.testng.annotations.AfterMethod; -import org.testng.annotations.BeforeClass; -import org.testng.annotations.BeforeMethod; -import org.testng.annotations.Test; +import static com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.CompatibilityMode; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.only; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotEquals; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; -import javax.xml.bind.DatatypeConverter; -import java.io.Closeable; import java.io.EOFException; import java.io.FilterInputStream; import java.io.FilterOutputStream; @@ -55,8 +40,6 @@ import java.math.MathContext; import java.net.Socket; import java.net.SocketException; -import java.sql.Connection; -import java.sql.DriverManager; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.SQLSyntaxErrorException; @@ -67,7 +50,6 @@ import java.util.List; import java.util.Map; import java.util.ResourceBundle; -import java.util.TimeZone; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -78,26 +60,37 @@ import java.util.logging.Level; import java.util.logging.Logger; -import static com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.CompatibilityMode; -import static org.mockito.Matchers.any; -import static org.mockito.Matchers.eq; -import static org.mockito.Mockito.inOrder; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.only; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; -import static org.testng.Assert.assertEquals; -import static org.testng.Assert.assertFalse; -import static org.testng.Assert.assertNotEquals; -import static org.testng.Assert.assertTrue; -import static org.testng.Assert.fail; +import javax.xml.bind.DatatypeConverter; + +import org.mockito.InOrder; +import org.testng.SkipException; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import com.github.shyiko.mysql.binlog.event.ByteArrayEventData; +import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData; +import com.github.shyiko.mysql.binlog.event.Event; +import com.github.shyiko.mysql.binlog.event.EventData; +import com.github.shyiko.mysql.binlog.event.EventHeaderV4; +import com.github.shyiko.mysql.binlog.event.EventType; +import com.github.shyiko.mysql.binlog.event.QueryEventData; +import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData; +import com.github.shyiko.mysql.binlog.event.WriteRowsEventData; +import com.github.shyiko.mysql.binlog.event.deserialization.ByteArrayEventDataDeserializer; +import com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializationException; +import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer; +import com.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer; +import com.github.shyiko.mysql.binlog.io.BufferedSocketInputStream; +import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream; +import com.github.shyiko.mysql.binlog.network.AuthenticationException; +import com.github.shyiko.mysql.binlog.network.ServerException; +import com.github.shyiko.mysql.binlog.network.SocketFactory; /** * @author Stanley Shyiko */ -public class BinaryLogClientIntegrationTest { - - protected static final long DEFAULT_TIMEOUT = TimeUnit.SECONDS.toMillis(3); +public class BinaryLogClientIntegrationTest extends BinaryLogClientIntegrationTestBase { private final Logger logger = Logger.getLogger(getClass().getSimpleName()); @@ -105,55 +98,43 @@ public class BinaryLogClientIntegrationTest { logger.setLevel(Level.FINEST); } - private final TimeZone timeZoneBeforeTheTest = TimeZone.getDefault(); - - protected MySQLConnection master, slave; - protected BinaryLogClient client; - protected CountDownEventListener eventListener; - - @BeforeClass - public void setUp() throws Exception { - TimeZone.setDefault(TimeZone.getTimeZone("GMT")); - ResourceBundle bundle = ResourceBundle.getBundle("jdbc"); - String prefix = "jdbc.mysql.replication."; - master = new MySQLConnection(bundle.getString(prefix + "master.hostname"), - Integer.parseInt(bundle.getString(prefix + "master.port")), - bundle.getString(prefix + "master.username"), bundle.getString(prefix + "master.password")); - slave = new MySQLConnection(bundle.getString(prefix + "slave.hostname"), - Integer.parseInt(bundle.getString(prefix + "slave.port")), - bundle.getString(prefix + "slave.superUsername"), bundle.getString(prefix + "slave.superPassword")); - client = new BinaryLogClient(slave.hostname, slave.port, slave.username, slave.password); - EventDeserializer eventDeserializer = new EventDeserializer(); - eventDeserializer.setCompatibilityMode(CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY, - CompatibilityMode.DATE_AND_TIME_AS_LONG); - client.setEventDeserializer(eventDeserializer); - client.setServerId(client.getServerId() - 1); // avoid clashes between BinaryLogClient instances - client.setKeepAlive(false); - client.registerEventListener(new TraceEventListener()); - client.registerEventListener(eventListener = new CountDownEventListener()); - client.registerLifecycleListener(new TraceLifecycleListener()); - client.connect(DEFAULT_TIMEOUT); + @BeforeMethod + public void beforeEachTest() throws Exception { master.execute(new Callback() { @Override public void execute(Statement statement) throws SQLException { - statement.execute("drop database if exists mbcj_test"); - statement.execute("create database mbcj_test"); - statement.execute("use mbcj_test"); + statement.execute("drop table if exists bikini_bottom"); + statement.execute("create table bikini_bottom (name varchar(255) primary key)"); } }); eventListener.waitFor(EventType.QUERY, 2, DEFAULT_TIMEOUT); + eventListener.reset(); } - @BeforeMethod - public void beforeEachTest() throws Exception { + @AfterMethod + public void afterEachTest() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final String markerQuery = "drop table if exists _EOS_marker"; + BinaryLogClient.EventListener markerInterceptor = new BinaryLogClient.EventListener() { + @Override + public void onEvent(Event event) { + if (event.getHeader().getEventType() == EventType.QUERY) { + EventData data = event.getData(); + if (data != null && ((QueryEventData) data).getSql().contains("_EOS_marker")) { + latch.countDown(); + } + } + } + }; + client.registerEventListener(markerInterceptor); master.execute(new Callback() { @Override public void execute(Statement statement) throws SQLException { - statement.execute("drop table if exists bikini_bottom"); - statement.execute("create table bikini_bottom (name varchar(255) primary key)"); + statement.execute(markerQuery); } }); - eventListener.waitFor(EventType.QUERY, 2, DEFAULT_TIMEOUT); + assertTrue(latch.await(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS)); + client.unregisterEventListener(markerInterceptor); eventListener.reset(); } @@ -1001,190 +982,4 @@ public void testMySQL8TableMetadata() throws Exception { eventListener.waitFor(WriteRowsEventData.class, 1, DEFAULT_TIMEOUT); } - @AfterMethod - public void afterEachTest() throws Exception { - final CountDownLatch latch = new CountDownLatch(1); - final String markerQuery = "drop table if exists _EOS_marker"; - BinaryLogClient.EventListener markerInterceptor = new BinaryLogClient.EventListener() { - @Override - public void onEvent(Event event) { - if (event.getHeader().getEventType() == EventType.QUERY) { - EventData data = event.getData(); - if (data != null && ((QueryEventData) data).getSql().contains("_EOS_marker")) { - latch.countDown(); - } - } - } - }; - client.registerEventListener(markerInterceptor); - master.execute(new Callback() { - @Override - public void execute(Statement statement) throws SQLException { - statement.execute(markerQuery); - } - }); - assertTrue(latch.await(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS)); - client.unregisterEventListener(markerInterceptor); - eventListener.reset(); - } - - @AfterClass(alwaysRun = true) - public void tearDown() throws Exception { - TimeZone.setDefault(timeZoneBeforeTheTest); - try { - if (client != null) { - client.disconnect(); - } - } finally { - if (slave != null) { - slave.close(); - } - if (master != null) { - master.execute(new Callback() { - @Override - public void execute(Statement statement) throws SQLException { - statement.execute("drop database mbcj_test"); - } - }); - master.close(); - } - } - } - - /** - * Representation of a MySQL connection. - */ - public static final class MySQLConnection implements Closeable { - - private final String hostname; - private final int port; - private final String username; - private final String password; - private Connection connection; - - public MySQLConnection(String hostname, int port, String username, String password) - throws ClassNotFoundException, SQLException { - this.hostname = hostname; - this.port = port; - this.username = username; - this.password = password; - Class.forName("com.mysql.jdbc.Driver"); - connect(); - } - - private void connect() throws SQLException { - this.connection = DriverManager.getConnection("jdbc:mysql://" + hostname + ":" + port + - "?serverTimezone=UTC", username, password); - execute(new Callback() { - - @Override - public void execute(Statement statement) throws SQLException { - statement.execute("SET time_zone = '+00:00'"); - } - }); - } - - public String hostname() { - return hostname; - } - - public int port() { - return port; - } - - public String username() { - return username; - } - - public String password() { - return password; - } - - public void execute(Callback callback, boolean autocommit) throws SQLException { - connection.setAutoCommit(autocommit); - Statement statement = connection.createStatement(); - try { - callback.execute(statement); - if (!autocommit) { - connection.commit(); - } - } finally { - statement.close(); - } - } - - public void execute(Callback callback) throws SQLException { - execute(callback, false); - } - - public void execute(final String...statements) throws SQLException { - execute(new Callback() { - @Override - public void execute(Statement statement) throws SQLException { - for (String command : statements) { - statement.execute(command); - } - } - }); - } - - public void query(String sql, Callback callback) throws SQLException { - connection.setAutoCommit(false); - Statement statement = connection.createStatement(); - try { - ResultSet rs = statement.executeQuery(sql); - try { - callback.execute(rs); - connection.commit(); - } finally { - rs.close(); - } - } finally { - statement.close(); - } - } - - @Override - public void close() throws IOException { - try { - connection.close(); - } catch (SQLException e) { - throw new IOException(e); - } - } - - public void reconnect() throws IOException, SQLException { - close(); - connect(); - } - } - - /** - * Callback used in the {@link MySQLConnection#execute(Callback)} method. - * - * @param the type of argument - */ - public interface Callback { - - void execute(T obj) throws SQLException; - } - - /** - * Used to simulate {@link SocketException} inside - * {@link QueryEventDataDeserializer#deserialize(ByteArrayInputStream)} (once). - */ - protected class QueryEventFailureSimulator extends QueryEventDataDeserializer { - private boolean failureSimulated; - - @Override - public QueryEventData deserialize(ByteArrayInputStream inputStream) throws IOException { - QueryEventData eventData = super.deserialize(inputStream); - if (!failureSimulated) { - failureSimulated = true; - throw new SocketException(); - } - return eventData; - } - } - } diff --git a/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientIntegrationTestBase.java b/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientIntegrationTestBase.java new file mode 100644 index 00000000..9fc79423 --- /dev/null +++ b/src/test/java/com/github/shyiko/mysql/binlog/BinaryLogClientIntegrationTestBase.java @@ -0,0 +1,245 @@ +/* + * Copyright 2013 Stanley Shyiko + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.shyiko.mysql.binlog; + +import static com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.CompatibilityMode; + +import java.io.Closeable; +import java.io.IOException; +import java.net.SocketException; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ResourceBundle; +import java.util.TimeZone; +import java.util.concurrent.TimeUnit; + +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; + +import com.github.shyiko.mysql.binlog.event.EventType; +import com.github.shyiko.mysql.binlog.event.QueryEventData; +import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer; +import com.github.shyiko.mysql.binlog.event.deserialization.QueryEventDataDeserializer; +import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream; + +/** + * @author Stanley Shyiko + */ +public class BinaryLogClientIntegrationTestBase { + + protected static final long DEFAULT_TIMEOUT = TimeUnit.SECONDS.toMillis(3); + protected final TimeZone timeZoneBeforeTheTest = TimeZone.getDefault(); + + protected MySQLConnection master, slave; + protected BinaryLogClient client; + protected CountDownEventListener eventListener; + + @BeforeClass + public void setUp() throws Exception { + TimeZone.setDefault(TimeZone.getTimeZone("GMT")); + ResourceBundle bundle = ResourceBundle.getBundle("jdbc"); + String prefix = "jdbc.mysql.replication."; + master = new MySQLConnection(bundle.getString(prefix + "master.hostname"), + Integer.parseInt(bundle.getString(prefix + "master.port")), + bundle.getString(prefix + "master.username"), bundle.getString(prefix + "master.password")); + slave = new MySQLConnection(bundle.getString(prefix + "slave.hostname"), + Integer.parseInt(bundle.getString(prefix + "slave.port")), + bundle.getString(prefix + "slave.superUsername"), bundle.getString(prefix + "slave.superPassword")); + client = new BinaryLogClient(slave.hostname, slave.port, slave.username, slave.password); + EventDeserializer eventDeserializer = new EventDeserializer(); + eventDeserializer.setCompatibilityMode(CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY, + CompatibilityMode.DATE_AND_TIME_AS_LONG); + client.setEventDeserializer(eventDeserializer); + client.setServerId(client.getServerId() - 1); // avoid clashes between BinaryLogClient instances + client.setKeepAlive(false); + client.registerEventListener(new TraceEventListener()); + client.registerEventListener(eventListener = new CountDownEventListener()); + client.registerLifecycleListener(new TraceLifecycleListener()); + client.connect(DEFAULT_TIMEOUT); + master.execute(new Callback() { + @Override + public void execute(Statement statement) throws SQLException { + statement.execute("drop database if exists mbcj_test"); + statement.execute("create database mbcj_test"); + statement.execute("use mbcj_test"); + } + }); + eventListener.waitFor(EventType.QUERY, 2, DEFAULT_TIMEOUT); + } + + @AfterClass(alwaysRun = true) + public void tearDown() throws Exception { + TimeZone.setDefault(timeZoneBeforeTheTest); + try { + if (client != null) { + client.disconnect(); + } + } finally { + if (slave != null) { + slave.close(); + } + if (master != null) { + master.execute(new Callback() { + @Override + public void execute(Statement statement) throws SQLException { + statement.execute("drop database mbcj_test"); + } + }); + master.close(); + } + } + } + + /** + * Representation of a MySQL connection. + */ + public static final class MySQLConnection implements Closeable { + + protected final String hostname; + protected final int port; + protected final String username; + protected final String password; + protected Connection connection; + + public MySQLConnection(String hostname, int port, String username, String password) + throws ClassNotFoundException, SQLException { + this.hostname = hostname; + this.port = port; + this.username = username; + this.password = password; + Class.forName("com.mysql.jdbc.Driver"); + connect(); + } + + private void connect() throws SQLException { + this.connection = DriverManager.getConnection("jdbc:mysql://" + hostname + ":" + port + + "?serverTimezone=UTC", username, password); + execute(new Callback() { + + @Override + public void execute(Statement statement) throws SQLException { + statement.execute("SET time_zone = '+00:00'"); + } + }); + } + + public String hostname() { + return hostname; + } + + public int port() { + return port; + } + + public String username() { + return username; + } + + public String password() { + return password; + } + + public void execute(Callback callback, boolean autocommit) throws SQLException { + connection.setAutoCommit(autocommit); + Statement statement = connection.createStatement(); + try { + callback.execute(statement); + if (!autocommit) { + connection.commit(); + } + } finally { + statement.close(); + } + } + + public void execute(Callback callback) throws SQLException { + execute(callback, false); + } + + public void execute(final String...statements) throws SQLException { + execute(new Callback() { + @Override + public void execute(Statement statement) throws SQLException { + for (String command : statements) { + statement.execute(command); + } + } + }); + } + + public void query(String sql, Callback callback) throws SQLException { + connection.setAutoCommit(false); + Statement statement = connection.createStatement(); + try { + ResultSet rs = statement.executeQuery(sql); + try { + callback.execute(rs); + connection.commit(); + } finally { + rs.close(); + } + } finally { + statement.close(); + } + } + + @Override + public void close() throws IOException { + try { + connection.close(); + } catch (SQLException e) { + throw new IOException(e); + } + } + + public void reconnect() throws IOException, SQLException { + close(); + connect(); + } + } + + /** + * Callback used in the {@link MySQLConnection#execute(Callback)} method. + * + * @param the type of argument + */ + public interface Callback { + + void execute(T obj) throws SQLException; + } + + /** + * Used to simulate {@link SocketException} inside + * {@link QueryEventDataDeserializer#deserialize(ByteArrayInputStream)} (once). + */ + protected class QueryEventFailureSimulator extends QueryEventDataDeserializer { + private boolean failureSimulated; + + @Override + public QueryEventData deserialize(ByteArrayInputStream inputStream) throws IOException { + QueryEventData eventData = super.deserialize(inputStream); + if (!failureSimulated) { + failureSimulated = true; + throw new SocketException(); + } + return eventData; + } + } + +}