MySQLBinlogEventPacketDecoder extends ByteToMessageDecoder { @Override protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final List<Object> out) { MySQLPacketPayload payload = new MySQLPacketPayload(in); skipSequenceId(payload); checkError(payload); MySQLBinlogEventHeader binlogEventHeader = new MySQLBinlogEventHeader(payload); removeChecksum(binlogEventHeader.getEventType(), in); switch (MySQLBinlogEventType.valueOf(binlogEventHeader.getEventType())) { case ROTATE_EVENT: decodeRotateEvent(binlogEventHeader, payload); break; case FORMAT_DESCRIPTION_EVENT: new MySQLBinlogFormatDescriptionEventPacket(binlogEventHeader, payload); break; case TABLE_MAP_EVENT: decodeTableMapEvent(binlogEventHeader, payload); break; case WRITE_ROWS_EVENTv1: case WRITE_ROWS_EVENTv2: out.add(decodeWriteRowsEventV2(binlogEventHeader, payload)); break; case UPDATE_ROWS_EVENTv1: case UPDATE_ROWS_EVENTv2: out.add(decodeUpdateRowsEventV2(binlogEventHeader, payload)); break; case DELETE_ROWS_EVENTv1: case DELETE_ROWS_EVENTv2: out.add(decodeDeleteRowsEventV2(binlogEventHeader, payload)); break; default: out.add(createPlaceholderEvent(binlogEventHeader)); payload.skipReserved(payload.getByteBuf().readableBytes()); } if (in.isReadable()) { throw new UnsupportedOperationException(String.format("Do not parse binlog event fully, eventHeader: %s, remaining packet %s", binlogEventHeader, readRemainPacket(payload))); } } MySQLBinlogEventPacketDecoder(final int checksumLength);   }
@Test(expected = RuntimeException.class) public void assertDecodeWithPacketError() { when(byteBuf.readUnsignedByte()).thenReturn((short) 255); binlogEventPacketDecoder.decode(null, byteBuf, null); }
@Test(expected = UnsupportedOperationException.class) public void assertDecodeWithReadError() { when(byteBuf.isReadable()).thenReturn(true); binlogEventPacketDecoder.decode(null, byteBuf, new ArrayList<>()); }
@Test public void assertDecodeRotateEvent() { when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) 0, (short) MySQLBinlogEventType.ROTATE_EVENT.getValue()); List<Object> decodedEvents = new ArrayList<>(); binlogEventPacketDecoder.decode(null, byteBuf, decodedEvents); assertThat(decodedEvents.size(), is(0)); assertThat(binlogContext.getFileName(), is("")); }
@Test public void assertDecodeFormatDescriptionEvent() { when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) 0, (short) MySQLBinlogEventType.FORMAT_DESCRIPTION_EVENT.getValue(), (short) 19); when(byteBuf.readUnsignedShortLE()).thenReturn(4); List<Object> decodedEvents = new ArrayList<>(); binlogEventPacketDecoder.decode(null, byteBuf, decodedEvents); assertThat(decodedEvents.size(), is(0)); assertThat(binlogContext.getChecksumLength(), is(4)); }
@Test public void assertDecodeTableMapEvent() { when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) 0, (short) MySQLBinlogEventType.TABLE_MAP_EVENT.getValue(), (short) 0); List<Object> decodedEvents = new ArrayList<>(); binlogEventPacketDecoder.decode(null, byteBuf, decodedEvents); assertThat(decodedEvents.size(), is(0)); assertThat(binlogContext.getTableMap().size(), is(1)); assertThat(binlogContext.getTableMap().get(0L), instanceOf(MySQLBinlogTableMapEventPacket.class)); }
@Test public void assertDecodeWriteRowEvent() { when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) 0, (short) MySQLBinlogEventType.WRITE_ROWS_EVENTv2.getValue(), (short) 0); when(byteBuf.readUnsignedShortLE()).thenReturn(2); binlogContext.getTableMap().put(0L, tableMapEventPacket); when(tableMapEventPacket.getColumnDefs()).thenReturn(Collections.emptyList()); List<Object> decodedEvents = new ArrayList<>(); binlogEventPacketDecoder.decode(null, byteBuf, decodedEvents); assertThat(decodedEvents.size(), is(1)); assertThat(decodedEvents.get(0), instanceOf(WriteRowsEvent.class)); }
@Test public void assertDecodeUpdateRowEvent() { when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) 0, (short) MySQLBinlogEventType.UPDATE_ROWS_EVENTv2.getValue(), (short) 0); when(byteBuf.readUnsignedShortLE()).thenReturn(2); binlogContext.getTableMap().put(0L, tableMapEventPacket); when(tableMapEventPacket.getColumnDefs()).thenReturn(Collections.emptyList()); List<Object> decodedEvents = new ArrayList<>(); binlogEventPacketDecoder.decode(null, byteBuf, decodedEvents); assertThat(decodedEvents.size(), is(1)); assertThat(decodedEvents.get(0), instanceOf(UpdateRowsEvent.class)); }
@Test public void assertDecodeDeleteRowEvent() { when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) 0, (short) MySQLBinlogEventType.DELETE_ROWS_EVENTv2.getValue(), (short) 0); when(byteBuf.readUnsignedShortLE()).thenReturn(2); binlogContext.getTableMap().put(0L, tableMapEventPacket); when(tableMapEventPacket.getColumnDefs()).thenReturn(Collections.emptyList()); List<Object> decodedEvents = new ArrayList<>(); binlogEventPacketDecoder.decode(null, byteBuf, decodedEvents); assertThat(decodedEvents.size(), is(1)); assertThat(decodedEvents.get(0), instanceOf(DeleteRowsEvent.class)); }