ChannelLZ4Decompressor extends AbstractCommunicatingProcess implements WithChannelTransformer<ChannelLZ4Decompressor, ByteBuf, ByteBuf>, WithBinaryChannelInput<ChannelLZ4Decompressor> { public static ChannelLZ4Decompressor create() { return create( LZ4Factory.fastestInstance().fastDecompressor(), XXHashFactory.fastestInstance()); } private  ChannelLZ4Decompressor(LZ4FastDecompressor decompressor, StreamingXXHash32 checksum); static ChannelLZ4Decompressor create(); static ChannelLZ4Decompressor create(LZ4FastDecompressor decompressor, XXHashFactory xxHashFactory); ChannelLZ4Decompressor withInspector(Inspector inspector); @Override BinaryChannelInput getInput(); @SuppressWarnings("ConstantConditions") //check output for clarity @Override ChannelOutput<ByteBuf> getOutput(); void processHeader(); void processBody(); static final int HEADER_LENGTH; static final ParseException STREAM_IS_CORRUPTED; }
@Test public void testTruncatedData() { ChannelLZ4Compressor compressor = ChannelLZ4Compressor.create(LZ4Factory.fastestInstance().fastCompressor()); ChannelLZ4Decompressor decompressor = ChannelLZ4Decompressor.create(); ByteBufQueue queue = new ByteBufQueue(); await(ChannelSupplier.of(ByteBufStrings.wrapAscii("TestData")).transformWith(compressor) .streamTo(ChannelConsumer.ofConsumer(queue::add))); queue.add(ByteBuf.wrapForReading(new byte[10])); Throwable e = awaitException(ChannelSupplier.of(queue.takeRemaining()) .transformWith(decompressor) .streamTo(ChannelConsumer.ofConsumer(data -> System.out.println(data.asString(UTF_8))))); assertSame(UNEXPECTED_DATA_EXCEPTION, e); }