BinaryLogConnectorEventMapper { public Optional<BinlogEvent> map( @NonNull final Event event, @NonNull final BinlogFilePos position) { final EventHeaderV4 header = event.getHeader(); final EventType eventType = header.getEventType(); final long serverId = header.getServerId(); final long timestamp = header.getTimestamp(); if (EventType.isWrite(eventType)) { final WriteRowsEventData data = event.getData(); return Optional.of( new WriteEvent(data.getTableId(), serverId, timestamp, position, data.getRows())); } else if (EventType.isUpdate(eventType)) { final UpdateRowsEventData data = event.getData(); return Optional.of( new UpdateEvent(data.getTableId(), serverId, timestamp, position, data.getRows())); } else if (EventType.isDelete(eventType)) { final DeleteRowsEventData data = event.getData(); return Optional.of( new DeleteEvent(data.getTableId(), serverId, timestamp, position, data.getRows())); } else { switch (eventType) { case TABLE_MAP: TableMapEventData tableMapData = event.getData(); return Optional.of( new TableMapEvent( tableMapData.getTableId(), serverId, timestamp, position, tableMapData.getDatabase(), tableMapData.getTable(), tableMapData.getColumnTypes())); case XID: final XidEventData xidData = event.getData(); return Optional.of(new XidEvent(serverId, timestamp, position, xidData.getXid())); case GTID: final GtidEventData gtidEventData = event.getData(); return Optional.of(new GTIDEvent(serverId, timestamp, position, gtidEventData.getGtid())); case QUERY: final QueryEventData queryData = event.getData(); return Optional.of( new QueryEvent( serverId, timestamp, position, queryData.getDatabase(), queryData.getSql())); case FORMAT_DESCRIPTION: return Optional.of(new StartEvent(serverId, timestamp, position)); default: return Optional.empty(); } } }  Optional<BinlogEvent> map(
      @NonNull final Event event, @NonNull final BinlogFilePos position); static final BinaryLogConnectorEventMapper INSTANCE; }
@Test public void testWriteEvent() { eventHeader.setEventType(EventType.EXT_WRITE_ROWS); WriteRowsEventData eventData = new WriteRowsEventData(); eventData.setTableId(TABLE_ID); eventData.setRows(ImmutableList.of(ROW)); Optional<BinlogEvent> binlogEvent = BinaryLogConnectorEventMapper.INSTANCE.map( new Event(eventHeader, eventData), BINLOG_FILE_POS); assertTrue(binlogEvent.isPresent()); assertTrue(binlogEvent.get() instanceof WriteEvent); WriteEvent writeEvent = (WriteEvent) (binlogEvent.get()); assertEquals(BINLOG_FILE_POS, writeEvent.getBinlogFilePos()); assertEquals(ImmutableList.of(ROW), writeEvent.getRows()); assertEquals(SERVER_ID, writeEvent.getServerId()); assertEquals(TABLE_ID, writeEvent.getTableId()); assertEquals(TIMESTAMP, writeEvent.getTimestamp()); }
@Test public void testUpdateEvent() { eventHeader.setEventType(EventType.EXT_UPDATE_ROWS); UpdateRowsEventData eventData = new UpdateRowsEventData(); eventData.setTableId(TABLE_ID); eventData.setRows(ImmutableList.of(Maps.immutableEntry(PREV_ROW, ROW))); Optional<BinlogEvent> binlogEvent = BinaryLogConnectorEventMapper.INSTANCE.map( new Event(eventHeader, eventData), BINLOG_FILE_POS); assertTrue(binlogEvent.isPresent()); assertTrue(binlogEvent.get() instanceof UpdateEvent); UpdateEvent updateEvent = (UpdateEvent) (binlogEvent.get()); assertEquals(BINLOG_FILE_POS, updateEvent.getBinlogFilePos()); assertEquals(ImmutableList.of(Maps.immutableEntry(PREV_ROW, ROW)), updateEvent.getRows()); assertEquals(SERVER_ID, updateEvent.getServerId()); assertEquals(TABLE_ID, updateEvent.getTableId()); assertEquals(TIMESTAMP, updateEvent.getTimestamp()); }
@Test public void testDeleteEvent() { eventHeader.setEventType(EventType.EXT_DELETE_ROWS); DeleteRowsEventData eventData = new DeleteRowsEventData(); eventData.setTableId(TABLE_ID); eventData.setRows(ImmutableList.of(PREV_ROW)); Optional<BinlogEvent> binlogEvent = BinaryLogConnectorEventMapper.INSTANCE.map( new Event(eventHeader, eventData), BINLOG_FILE_POS); assertTrue(binlogEvent.isPresent()); assertTrue(binlogEvent.get() instanceof DeleteEvent); DeleteEvent deleteEvent = (DeleteEvent) (binlogEvent.get()); assertEquals(BINLOG_FILE_POS, deleteEvent.getBinlogFilePos()); assertEquals(ImmutableList.of(PREV_ROW), deleteEvent.getRows()); assertEquals(SERVER_ID, deleteEvent.getServerId()); assertEquals(TABLE_ID, deleteEvent.getTableId()); assertEquals(TIMESTAMP, deleteEvent.getTimestamp()); }
@Test public void testTableMapEvent() { eventHeader.setEventType(EventType.TABLE_MAP); TableMapEventData eventData = new TableMapEventData(); eventData.setDatabase(DATABASE); eventData.setTable(TABLE); eventData.setTableId(TABLE_ID); eventData.setColumnTypes(new byte[] {(byte) 0, (byte) 1, (byte) 2}); Optional<BinlogEvent> binlogEvent = BinaryLogConnectorEventMapper.INSTANCE.map( new Event(eventHeader, eventData), BINLOG_FILE_POS); assertTrue(binlogEvent.isPresent()); assertTrue(binlogEvent.get() instanceof TableMapEvent); TableMapEvent tableMapEvent = (TableMapEvent) (binlogEvent.get()); assertEquals(BINLOG_FILE_POS, tableMapEvent.getBinlogFilePos()); assertEquals(DATABASE, tableMapEvent.getDatabase()); assertEquals(TABLE, tableMapEvent.getTable()); assertEquals(TABLE_ID, tableMapEvent.getTableId()); assertEquals( ImmutableList.of(ColumnDataType.DECIMAL, ColumnDataType.TINY, ColumnDataType.SHORT), tableMapEvent.getColumnTypes()); }
@Test public void testXidEvent() { long xid = 88888L; eventHeader.setEventType(EventType.XID); XidEventData eventData = new XidEventData(); eventData.setXid(xid); Optional<BinlogEvent> binlogEvent = BinaryLogConnectorEventMapper.INSTANCE.map( new Event(eventHeader, eventData), BINLOG_FILE_POS); assertTrue(binlogEvent.isPresent()); assertTrue(binlogEvent.get() instanceof XidEvent); XidEvent xidEvent = (XidEvent) (binlogEvent.get()); assertEquals(BINLOG_FILE_POS, xidEvent.getBinlogFilePos()); assertEquals(SERVER_ID, xidEvent.getServerId()); assertEquals(TIMESTAMP, xidEvent.getTimestamp()); assertEquals(xid, xidEvent.getXid()); }
@Test public void testQueryEvent() { String sql = "CREATE UNIQUE INDEX unique_index ON `my_db`.`my_table` (`col1`, `col2`)"; eventHeader.setEventType(EventType.QUERY); QueryEventData eventData = new QueryEventData(); eventData.setDatabase(DATABASE); eventData.setSql(sql); Optional<BinlogEvent> binlogEvent = BinaryLogConnectorEventMapper.INSTANCE.map( new Event(eventHeader, eventData), BINLOG_FILE_POS); assertTrue(binlogEvent.isPresent()); assertTrue(binlogEvent.get() instanceof QueryEvent); QueryEvent queryEvent = (QueryEvent) (binlogEvent.get()); assertEquals(BINLOG_FILE_POS, queryEvent.getBinlogFilePos()); assertEquals(DATABASE, queryEvent.getDatabase()); assertEquals(SERVER_ID, queryEvent.getServerId()); assertEquals(TIMESTAMP, queryEvent.getTimestamp()); assertEquals(sql, queryEvent.getSql()); }
@Test public void testFormatDescriptionEvent() { eventHeader.setEventType(EventType.FORMAT_DESCRIPTION); FormatDescriptionEventData eventData = new FormatDescriptionEventData(); Optional<BinlogEvent> binlogEvent = BinaryLogConnectorEventMapper.INSTANCE.map( new Event(eventHeader, eventData), BINLOG_FILE_POS); assertTrue(binlogEvent.isPresent()); assertTrue(binlogEvent.get() instanceof StartEvent); StartEvent startEvent = (StartEvent) (binlogEvent.get()); assertEquals(BINLOG_FILE_POS, startEvent.getBinlogFilePos()); assertEquals(SERVER_ID, startEvent.getServerId()); assertEquals(TIMESTAMP, startEvent.getTimestamp()); }
@Test public void testIgnoredEvents() { eventHeader.setEventType(EventType.UNKNOWN); XAPrepareEventData eventData = new XAPrepareEventData(); Optional<BinlogEvent> binlogEvent = BinaryLogConnectorEventMapper.INSTANCE.map( new Event(eventHeader, eventData), BINLOG_FILE_POS); assertFalse(binlogEvent.isPresent()); }