LimitByRowsChannelBuffer implements ChannelBuffer { @Override public boolean push(BufferableBinlogDataSet ds, long timeout, TimeUnit timeUnit) { initIfNeed(); int sizeSum = 0; if (ds.getBinlogDataSet() != null && ds.getBinlogDataSet().getRowDataMap() != null) { for (List<BinlogResultRow> binlogResultRows : ds.getBinlogDataSet().getRowDataMap() .values()) { sizeSum += binlogResultRows.size(); } } sizeSum = Math.min(sizeSum, limit); try { if (sizeSum > 0) { if (!semaphore.tryAcquire(sizeSum, timeout, timeUnit)) { return false; } } return queue.offer(new BufferableBinlogDataSetWithRowCount(ds, sizeSum), timeout, timeUnit); } catch (InterruptedException e) { return false; } } LimitByRowsChannelBuffer(int limit);  LimitByRowsChannelBuffer(); int getLimit(); void setLimit(int limit); @Override boolean push(BufferableBinlogDataSet ds, long timeout, TimeUnit timeUnit); @Override BufferableBinlogDataSet pop(long timeout, TimeUnit timeUnit); int availablePermits();  }
@Test public void testNormalCase() { LimitByRowsChannelBuffer limitByRowsChannelBuffer = new LimitByRowsChannelBuffer(2); BinlogDataSet binlogDataSet = new BinlogDataSet("channelId", "sourceHostUrl", "gtId", "binlogPos"); List<BinlogResultRow> binlogResultRows = new ArrayList<>(); binlogResultRows.add(new BinlogResultRow(null, null, null, 0)); binlogDataSet.getRowDataMap().put("tableA", binlogResultRows); BufferableBinlogDataSet bufferableBinlogDataSet = new DefaultBufferableBinlogDataSet(binlogDataSet); Assert.assertTrue( limitByRowsChannelBuffer.push(bufferableBinlogDataSet, 1, TimeUnit.NANOSECONDS)); Assert.assertTrue( limitByRowsChannelBuffer.push(bufferableBinlogDataSet, 1, TimeUnit.NANOSECONDS)); Assert.assertFalse( limitByRowsChannelBuffer.push(bufferableBinlogDataSet, 1, TimeUnit.NANOSECONDS)); }