{
- private static final EventType[] EVENT_TYPES = EventType.values();
-
@Override
public EventHeaderV4 deserialize(ByteArrayInputStream inputStream) throws IOException {
EventHeaderV4 header = new EventHeaderV4();
@@ -41,10 +39,8 @@ public EventHeaderV4 deserialize(ByteArrayInputStream inputStream) throws IOExce
}
private static EventType getEventType(int ordinal) {
- if (ordinal >= EVENT_TYPES.length) {
- return EventType.UNKNOWN;
- }
- return EVENT_TYPES[ordinal];
+ EventType eventType = EventType.byEventNumber(ordinal);
+ return eventType == null ? EventType.UNKNOWN : eventType;
}
}
diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidEventDataDeserializer.java b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidEventDataDeserializer.java
new file mode 100644
index 0000000..c57e255
--- /dev/null
+++ b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidEventDataDeserializer.java
@@ -0,0 +1,33 @@
+package com.github.shyiko.mysql.binlog.event.deserialization;
+
+import com.github.shyiko.mysql.binlog.event.MariadbGtidEventData;
+import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
+
+import java.io.IOException;
+
+/**
+ * Mariadb GTID_EVENT Fields
+ *
+ * uint<8> GTID sequence
+ * uint<4> Replication Domain ID
+ * uint<1> Flags
+ *
+ * if flag & FL_GROUP_COMMIT_ID
+ * uint<8> commit_id
+ * else
+ * uint<6> 0
+ *
+ *
+ * @author Winger
+ * @see GTID_EVENT for the original doc
+ */
+public class MariadbGtidEventDataDeserializer implements EventDataDeserializer {
+ @Override
+ public MariadbGtidEventData deserialize(ByteArrayInputStream inputStream) throws IOException {
+ MariadbGtidEventData event = new MariadbGtidEventData();
+ event.setSequence(inputStream.readLong(8));
+ event.setDomainId(inputStream.readInteger(4));
+ // Flags ignore
+ return event;
+ }
+}
diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidListEventDataDeserializer.java b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidListEventDataDeserializer.java
new file mode 100644
index 0000000..8a42953
--- /dev/null
+++ b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidListEventDataDeserializer.java
@@ -0,0 +1,39 @@
+package com.github.shyiko.mysql.binlog.event.deserialization;
+
+
+import com.github.shyiko.mysql.binlog.MariadbGtidSet;
+import com.github.shyiko.mysql.binlog.event.MariadbGtidListEventData;
+import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
+
+import java.io.IOException;
+
+/**
+ * Mariadb GTID_LIST_EVENT Fields
+ *
+ * uint<4> Number of GTIDs
+ * GTID[0]
+ * uint<4> Replication Domain ID
+ * uint<4> Server_ID
+ * uint<8> GTID sequence ...
+ * GTID[n]
+ *
+ *
+ * @author Winger
+ * @see GTID_EVENT for the original doc
+ */
+public class MariadbGtidListEventDataDeserializer implements EventDataDeserializer {
+ @Override
+ public MariadbGtidListEventData deserialize(ByteArrayInputStream inputStream) throws IOException {
+ MariadbGtidListEventData eventData = new MariadbGtidListEventData();
+ long gtidLength = inputStream.readInteger(4);
+ MariadbGtidSet mariaGTIDSet = new MariadbGtidSet();
+ for (int i = 0; i < gtidLength; i++) {
+ long domainId = inputStream.readInteger(4);
+ long serverId = inputStream.readInteger(4);
+ long sequence = inputStream.readLong(8);
+ mariaGTIDSet.add(new MariadbGtidSet.MariaGtid(domainId, serverId, sequence));
+ }
+ eventData.setMariaGTIDSet(mariaGTIDSet);
+ return eventData;
+ }
+}
diff --git a/src/main/java/com/github/shyiko/mysql/binlog/network/protocol/command/DumpBinaryLogCommand.java b/src/main/java/com/github/shyiko/mysql/binlog/network/protocol/command/DumpBinaryLogCommand.java
index 36216c7..b7436aa 100644
--- a/src/main/java/com/github/shyiko/mysql/binlog/network/protocol/command/DumpBinaryLogCommand.java
+++ b/src/main/java/com/github/shyiko/mysql/binlog/network/protocol/command/DumpBinaryLogCommand.java
@@ -24,9 +24,11 @@
*/
public class DumpBinaryLogCommand implements Command {
+ public static final int BINLOG_SEND_ANNOTATE_ROWS_EVENT = 2;
private long serverId;
private String binlogFilename;
private long binlogPosition;
+ private boolean sendAnnotateRowsEvent;
public DumpBinaryLogCommand(long serverId, String binlogFilename, long binlogPosition) {
this.serverId = serverId;
@@ -34,12 +36,21 @@ public DumpBinaryLogCommand(long serverId, String binlogFilename, long binlogPos
this.binlogPosition = binlogPosition;
}
+ public DumpBinaryLogCommand(long serverId, String binlogFilename, long binlogPosition, boolean sendAnnotateRowsEvent) {
+ this(serverId, binlogFilename, binlogPosition);
+ this.sendAnnotateRowsEvent = sendAnnotateRowsEvent;
+ }
+
@Override
public byte[] toByteArray() throws IOException {
ByteArrayOutputStream buffer = new ByteArrayOutputStream();
buffer.writeInteger(CommandType.BINLOG_DUMP.ordinal(), 1);
buffer.writeLong(this.binlogPosition, 4);
- buffer.writeInteger(0, 2); // flag
+ int binlogFlags = 0;
+ if (sendAnnotateRowsEvent) {
+ binlogFlags |= BINLOG_SEND_ANNOTATE_ROWS_EVENT;
+ }
+ buffer.writeInteger(binlogFlags, 2); // flag
buffer.writeLong(this.serverId, 4);
buffer.writeString(this.binlogFilename);
return buffer.toByteArray();
diff --git a/src/test/java/com/github/shyiko/mysql/binlog/MariadbBinaryLogClientIntegrationTest.java b/src/test/java/com/github/shyiko/mysql/binlog/MariadbBinaryLogClientIntegrationTest.java
new file mode 100644
index 0000000..fac8b57
--- /dev/null
+++ b/src/test/java/com/github/shyiko/mysql/binlog/MariadbBinaryLogClientIntegrationTest.java
@@ -0,0 +1,84 @@
+package com.github.shyiko.mysql.binlog;
+
+import com.github.shyiko.mysql.binlog.event.AnnotateRowsEventData;
+import com.github.shyiko.mysql.binlog.event.MariadbGtidEventData;
+import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
+import org.testng.annotations.Test;
+
+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;
+
+/**
+ * @author Winger
+ */
+public class MariadbBinaryLogClientIntegrationTest {
+
+ protected BinaryLogClientIntegrationTest.MySQLConnection master;
+
+ @Test
+ public void testMariadbUseGTIDAndAnnotateRowsEvent() throws Exception {
+ master = new BinaryLogClientIntegrationTest.MySQLConnection("127.0.0.1", 3306, "root", "");
+ master.execute(new BinaryLogClientIntegrationTest.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("CREATE TABLE if not exists foo (i int)");
+ statement.execute("CREATE TABLE if not exists bar (i int)");
+ }
+ });
+ // get current gtid
+ final String[] currentGtidPos = new String[1];
+ master.query("show global variables like 'gtid_current_pos%'", new BinaryLogClientIntegrationTest.Callback() {
+
+ @Override
+ public void execute(ResultSet rs) throws SQLException {
+ rs.next();
+ currentGtidPos[0] = rs.getString(2);
+ }
+ });
+
+ CountDownEventListener eventListener;
+ MariadbBinaryLogClient client = new MariadbBinaryLogClient("127.0.0.1", 3306, "root", "123456");
+ client.setGtidSet(currentGtidPos[0]);
+ client.setUseSendAnnotateRowsEvent(true);
+
+ EventDeserializer eventDeserializer = new EventDeserializer();
+ eventDeserializer.setCompatibilityMode(EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY,
+ EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG);
+ client.setEventDeserializer(eventDeserializer);
+ client.registerEventListener(new TraceEventListener());
+ client.registerLifecycleListener(new TraceLifecycleListener());
+ client.registerEventListener(eventListener = new CountDownEventListener());
+
+ master.execute(new BinaryLogClientIntegrationTest.Callback() {
+ @Override
+ public void execute(Statement statement) throws SQLException {
+ statement.execute("INSERT INTO foo set i = 2");
+ statement.execute("DROP TABLE IF EXISTS bar");
+ }
+ });
+
+ try {
+ eventListener.reset();
+ client.connect();
+
+ eventListener.waitFor(MariadbGtidEventData.class, 1, TimeUnit.SECONDS.toMillis(4));
+ String gtidSet = client.getGtidSet();
+ assertNotNull(gtidSet);
+
+ eventListener.reset();
+ eventListener.waitFor(AnnotateRowsEventData.class, 1, TimeUnit.SECONDS.toMillis(4));
+ gtidSet = client.getGtidSet();
+ assertNotEquals(currentGtidPos[0], gtidSet);
+ } finally {
+ client.disconnect();
+ }
+ }
+}
diff --git a/src/test/java/com/github/shyiko/mysql/binlog/MariadbGtidSetTest.java b/src/test/java/com/github/shyiko/mysql/binlog/MariadbGtidSetTest.java
new file mode 100644
index 0000000..71d5753
--- /dev/null
+++ b/src/test/java/com/github/shyiko/mysql/binlog/MariadbGtidSetTest.java
@@ -0,0 +1,36 @@
+package com.github.shyiko.mysql.binlog;
+
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+
+/**
+ * @author Winger
+ */
+public class MariadbGtidSetTest {
+
+ @Test
+ public void testAdd() {
+ MariadbGtidSet gtidSet = new MariadbGtidSet("0-102-7255");
+ gtidSet.add("0-102-7256");
+ gtidSet.add("0-102-7257");
+ gtidSet.add("0-102-7259");
+ gtidSet.add("1-102-7300");
+ assertNotEquals(gtidSet.toString(), "1-102-7300");
+ assertNotEquals(gtidSet.toString(), "0-102-7259");
+ assertEquals(gtidSet.toString(), "0-102-7259,1-102-7300");
+ }
+
+ @Test
+ public void testEmptySet() {
+ assertEquals(new MariadbGtidSet("").toString(), "");
+ }
+
+ @Test
+ public void testEquals() {
+ assertEquals(new MariadbGtidSet(""), new MariadbGtidSet(null));
+ assertEquals(new MariadbGtidSet(""), new MariadbGtidSet(""));
+ assertEquals(new MariadbGtidSet("0-0-7404"), new MariadbGtidSet("0-0-7404"));
+ }
+}
diff --git a/src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/AnnotateRowsEventDataDeserializerTest.java b/src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/AnnotateRowsEventDataDeserializerTest.java
new file mode 100644
index 0000000..e117d32
--- /dev/null
+++ b/src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/AnnotateRowsEventDataDeserializerTest.java
@@ -0,0 +1,27 @@
+package com.github.shyiko.mysql.binlog.event.deserialization;
+
+import com.github.shyiko.mysql.binlog.event.AnnotateRowsEventData;
+import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static junit.framework.Assert.assertEquals;
+
+/**
+ * @author Winger
+ */
+public class AnnotateRowsEventDataDeserializerTest {
+
+ private static final byte[] DATA = {73, 78, 83, 69, 82, 84, 32, 73, 78, 84, 79, 32, 102, 111, 111, 32, 115, 101, 116, 32, 105, 32, 61, 32, 50};
+
+ private static final String sql = "INSERT INTO foo set i = 2";
+
+ @Test
+ public void deserialize() throws IOException {
+ AnnotateRowsEventDataDeserializer deserializer = new AnnotateRowsEventDataDeserializer();
+ AnnotateRowsEventData eventData = deserializer.deserialize(new ByteArrayInputStream(DATA));
+
+ assertEquals(sql, eventData.getRowsQuery());
+ }
+}
diff --git a/src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidEventDataDeserializerTest.java b/src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidEventDataDeserializerTest.java
new file mode 100644
index 0000000..f6e703b
--- /dev/null
+++ b/src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidEventDataDeserializerTest.java
@@ -0,0 +1,26 @@
+package com.github.shyiko.mysql.binlog.event.deserialization;
+
+import com.github.shyiko.mysql.binlog.event.MariadbGtidEventData;
+import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static junit.framework.Assert.assertEquals;
+
+/**
+ * @author Winger
+ */
+public class MariadbGtidEventDataDeserializerTest {
+
+ private static final byte[] DATA = {-20, 28, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 12, 0, 0, 0, 0, 0, 0, 121};
+
+ private static final String GTID_SET = "0-0-7404";
+
+ @Test
+ public void deserialize() throws IOException {
+ MariadbGtidEventDataDeserializer deserializer = new MariadbGtidEventDataDeserializer();
+ MariadbGtidEventData eventData = deserializer.deserialize(new ByteArrayInputStream(DATA));
+ assertEquals(GTID_SET, eventData.toString());
+ }
+}
diff --git a/src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidListEventDataDeserializerTest.java b/src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidListEventDataDeserializerTest.java
new file mode 100644
index 0000000..3a6f33f
--- /dev/null
+++ b/src/test/java/com/github/shyiko/mysql/binlog/event/deserialization/MariadbGtidListEventDataDeserializerTest.java
@@ -0,0 +1,26 @@
+package com.github.shyiko.mysql.binlog.event.deserialization;
+
+import com.github.shyiko.mysql.binlog.event.MariadbGtidListEventData;
+import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static junit.framework.Assert.assertEquals;
+
+/**
+ * @author Winger
+ */
+public class MariadbGtidListEventDataDeserializerTest {
+
+ private static final byte[] DATA = {1, 0, 0, 0, 0, 0, 0, 0, 102, 0, 0, 0, 87, 28, 0, 0, 0, 0, 0, 0, 77};
+
+ private static final String GTID_SET_LIST = "MariadbGtidListEventData{mariaGTIDSet=0-102-7255}";
+
+ @Test
+ public void deserialize() throws IOException {
+ MariadbGtidListEventDataDeserializer deserializer = new MariadbGtidListEventDataDeserializer();
+ MariadbGtidListEventData eventData = deserializer.deserialize(new ByteArrayInputStream(DATA));
+ assertEquals(GTID_SET_LIST, eventData.toString());
+ }
+}