MessageDeframer implements Closeable { @Override public void close() { try { if (unprocessed != null) { unprocessed.close(); } if (nextFrame != null) { nextFrame.close(); } } finally { unprocessed = null; nextFrame = null; } } MessageDeframer(Listener listener, Decompressor decompressor, int maxMessageSize); void setDecompressor(Decompressor decompressor); void request(int numMessages); void deframe(ReadableBuffer data, boolean endOfStream); boolean isStalled(); @Override void close(); boolean isClosed();  }
@Test public void sizeEnforcingInputStream_readByteBelowLimit() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 4); while (stream.read() != -1) {} stream.close(); }
@Test public void sizeEnforcingInputStream_readByteAtLimit() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 3); while (stream.read() != -1) {} stream.close(); }
@Test public void sizeEnforcingInputStream_readByteAboveLimit() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 2); thrown.expect(StatusRuntimeException.class); thrown.expectMessage("INTERNAL: Compressed frame exceeds"); while (stream.read() != -1) {} stream.close(); }
@Test public void sizeEnforcingInputStream_readBelowLimit() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 4); byte[] buf = new byte[10]; int read = stream.read(buf, 0, buf.length); assertEquals(3, read); stream.close(); }
@Test public void sizeEnforcingInputStream_readAtLimit() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 3); byte[] buf = new byte[10]; int read = stream.read(buf, 0, buf.length); assertEquals(3, read); stream.close(); }
@Test public void sizeEnforcingInputStream_readAboveLimit() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 2); byte[] buf = new byte[10]; thrown.expect(StatusRuntimeException.class); thrown.expectMessage("INTERNAL: Compressed frame exceeds"); stream.read(buf, 0, buf.length); stream.close(); }
@Test public void sizeEnforcingInputStream_skipBelowLimit() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 4); long skipped = stream.skip(4); assertEquals(3, skipped); stream.close(); }
@Test public void sizeEnforcingInputStream_skipAtLimit() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 3); long skipped = stream.skip(4); assertEquals(3, skipped); stream.close(); }
@Test public void sizeEnforcingInputStream_skipAboveLimit() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 2); thrown.expect(StatusRuntimeException.class); thrown.expectMessage("INTERNAL: Compressed frame exceeds"); stream.skip(4); stream.close(); }
@Test public void sizeEnforcingInputStream_markReset() throws IOException { ByteArrayInputStream in = new ByteArrayInputStream("foo".getBytes(Charsets.UTF_8)); SizeEnforcingInputStream stream = new MessageDeframer.SizeEnforcingInputStream(in, 3); stream.skip(1); stream.mark(10); stream.skip(2); stream.reset(); long skipped = stream.skip(2); assertEquals(2, skipped); stream.close(); }