DataCarrier { public boolean produce(T data) { if (driver != null) { if (!driver.isRunning(channels)) { return false; } } return this.channels.save(data); } DataCarrier(int channelSize, int bufferSize);  DataCarrier(String name, int channelSize, int bufferSize);  DataCarrier(String name, String envPrefix, int channelSize, int bufferSize); DataCarrier setPartitioner(IDataPartitioner<T> dataPartitioner); DataCarrier setBufferStrategy(BufferStrategy strategy); boolean produce(T data); DataCarrier consume(Class<? extends IConsumer<T>> consumerClass, int num, long consumeCycle); DataCarrier consume(Class<? extends IConsumer<T>> consumerClass, int num); DataCarrier consume(IConsumer<T> consumer, int num, long consumeCycle); DataCarrier consume(IConsumer<T> consumer, int num); DataCarrier consume(ConsumerPool consumerPool, IConsumer<T> consumer); void shutdownConsumers();  }
@Test public void testProduce() throws IllegalAccessException { DataCarrier<SampleData> carrier = new DataCarrier<SampleData>(2, 100); Assert.assertTrue(carrier.produce(new SampleData().setName("a"))); Assert.assertTrue(carrier.produce(new SampleData().setName("b"))); Assert.assertTrue(carrier.produce(new SampleData().setName("c"))); Assert.assertTrue(carrier.produce(new SampleData().setName("d"))); Channels<SampleData> channels = (Channels<SampleData>) (MemberModifier.field(DataCarrier.class, "channels") .get(carrier)); QueueBuffer<SampleData> buffer1 = channels.getBuffer(0); List result = new ArrayList(); buffer1.obtain(result); Assert.assertEquals(2, result.size()); QueueBuffer<SampleData> buffer2 = channels.getBuffer(1); buffer2.obtain(result); Assert.assertEquals(4, result.size()); }