DistributionChannel implements Channel { @Override public void pushRecord(final Record record) throws InterruptedException { if (FinishedRecord.class.equals(record.getClass())) { for (Entry<String, MemoryChannel> entry : channels.entrySet()) { entry.getValue().pushRecord(record); } } else if (DataRecord.class.equals(record.getClass())) { toBeAcknowledgeRecords.add(record); DataRecord dataRecord = (DataRecord) record; String index = Integer.toString(Math.abs(dataRecord.hashCode()) % channelNumber); channels.get(index).pushRecord(dataRecord); } else if (PlaceholderRecord.class.equals(record.getClass())) { toBeAcknowledgeRecords.add(record); pendingAcknowledgeRecords.put(record.getPosition(), record); } else { throw new RuntimeException("Not Support Record Type"); } } DistributionChannel(final int channelNumber, final AckCallback ackCallback); @Override void pushRecord(final Record record); @Override List<Record> fetchRecords(final int batchSize, final int timeout); @Override void ack(); @Override void close();  }
@Test public void assertAckCallbackResultSortable() throws InterruptedException, NoSuchMethodException, IllegalAccessException, InvocationTargetException { distributionChannel = new DistributionChannel(2, records -> { assertThat(records.size(), is(2)); assertTrue(((IntPosition) records.get(0).getPosition()).getId() < ((IntPosition) records.get(1).getPosition()).getId()); }); distributionChannel.pushRecord(new PlaceholderRecord(new IntPosition(1))); distributionChannel.pushRecord(new PlaceholderRecord(new IntPosition(2))); fetchRecordsAndSleep(0); fetchRecordsAndSleep(1); ReflectionUtil.invokeMethod(distributionChannel, "ackRecords0"); }
@Test public void assertBroadcastFinishedRecord() throws InterruptedException { distributionChannel = new DistributionChannel(2, records -> assertThat(records.size(), is(2))); distributionChannel.pushRecord(new FinishedRecord(new NopPosition())); }