@Test public void testEventFilter() throws Exception { TableCache tableCache = mock(TableCache.class); BinlogEvent lastEvent = new XidEvent(0l, 0l, BINLOG_FILE_POS, 0l); BinlogFilePos nextPosition = new BinlogFilePos("test.123", 15, 100); MysqlSourceState state = new MysqlSourceState(0l, lastEvent.getOffset(), 0l, BINLOG_FILE_POS); Filter<BinlogEvent> filter = MysqlEventFilter.create(tableCache, TABLE_NAMES, new AtomicReference(state)); when(tableCache.contains(TABLE_ID)).thenReturn(true); assertTrue( filter.apply( new TableMapEvent( TABLE_ID, 0l, 0l, nextPosition, DATABASE_NAME, TABLE_NAME, new byte[1]))); assertTrue( filter.apply(new WriteEvent(TABLE_ID, 0l, 0l, nextPosition, Collections.emptyList()))); assertTrue( filter.apply(new DeleteEvent(TABLE_ID, 0l, 0l, nextPosition, Collections.emptyList()))); assertTrue( filter.apply(new UpdateEvent(TABLE_ID, 0l, 0l, nextPosition, Collections.emptyList()))); assertTrue(filter.apply(new XidEvent(0l, 0l, BINLOG_FILE_POS, 12l))); assertTrue(filter.apply(new QueryEvent(0l, 0l, BINLOG_FILE_POS, DATABASE_NAME, ""))); assertTrue(filter.apply(new StartEvent(0l, 0l, BINLOG_FILE_POS))); assertFalse( filter.apply(new TableMapEvent(TABLE_ID, 0l, 0l, BINLOG_FILE_POS, "", "", new byte[1]))); assertFalse(filter.apply(new WriteEvent(0l, 0l, 0l, BINLOG_FILE_POS, Collections.emptyList()))); assertFalse( filter.apply(new WriteEvent(TABLE_ID, 0l, 0l, BINLOG_FILE_POS, Collections.emptyList()))); assertFalse(filter.apply(mock(BinlogEvent.class))); }
public static Filter<BinlogEvent> create( @NonNull final TableCache tableCache, @NonNull final Set<String> tableNames, @NonNull final AtomicReference<MysqlSourceState> state) { return ChainedFilter.<BinlogEvent>builder() .addFilter(new EventTypeFilter()) .addFilter(new TableFilter(tableCache, tableNames)) .addFilter(new DuplicateFilter(state)) .build(); }
MysqlEventFilter implements Filter<BinlogEvent> { public static Filter<BinlogEvent> create( @NonNull final TableCache tableCache, @NonNull final Set<String> tableNames, @NonNull final AtomicReference<MysqlSourceState> state) { return ChainedFilter.<BinlogEvent>builder() .addFilter(new EventTypeFilter()) .addFilter(new TableFilter(tableCache, tableNames)) .addFilter(new DuplicateFilter(state)) .build(); } }
MysqlEventFilter implements Filter<BinlogEvent> { public static Filter<BinlogEvent> create( @NonNull final TableCache tableCache, @NonNull final Set<String> tableNames, @NonNull final AtomicReference<MysqlSourceState> state) { return ChainedFilter.<BinlogEvent>builder() .addFilter(new EventTypeFilter()) .addFilter(new TableFilter(tableCache, tableNames)) .addFilter(new DuplicateFilter(state)) .build(); }  }
MysqlEventFilter implements Filter<BinlogEvent> { public static Filter<BinlogEvent> create( @NonNull final TableCache tableCache, @NonNull final Set<String> tableNames, @NonNull final AtomicReference<MysqlSourceState> state) { return ChainedFilter.<BinlogEvent>builder() .addFilter(new EventTypeFilter()) .addFilter(new TableFilter(tableCache, tableNames)) .addFilter(new DuplicateFilter(state)) .build(); }  static Filter<BinlogEvent> create(
      @NonNull final TableCache tableCache,
      @NonNull final Set<String> tableNames,
      @NonNull final AtomicReference<MysqlSourceState> state); }
MysqlEventFilter implements Filter<BinlogEvent> { public static Filter<BinlogEvent> create( @NonNull final TableCache tableCache, @NonNull final Set<String> tableNames, @NonNull final AtomicReference<MysqlSourceState> state) { return ChainedFilter.<BinlogEvent>builder() .addFilter(new EventTypeFilter()) .addFilter(new TableFilter(tableCache, tableNames)) .addFilter(new DuplicateFilter(state)) .build(); }  static Filter<BinlogEvent> create(
      @NonNull final TableCache tableCache,
      @NonNull final Set<String> tableNames,
      @NonNull final AtomicReference<MysqlSourceState> state);  }