@Test public void testNegativeMessageSize() throws Exception { KryoMessageCodec codec = new KryoMessageCodec(1024); ByteBuf buf = newBuffer(); buf.writeInt(-1); try { List<Object> out = Lists.newArrayList(); codec.decode(null, buf, out); fail("Should have failed to decode message with negative size."); } catch (IllegalArgumentException e) { assertTrue(e.getMessage().indexOf("must be positive") > 0); } }
@Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { if (in.readableBytes() < 4) { return; } in.markReaderIndex(); int msgSize = in.readInt(); checkSize(msgSize); if (in.readableBytes() < msgSize) { in.resetReaderIndex(); return; } try { ByteBuffer nioBuffer = maybeDecrypt(in.nioBuffer(in.readerIndex(), msgSize)); Input kryoIn = new Input(new ByteBufferInputStream(nioBuffer)); Object msg = kryos.get().readClassAndObject(kryoIn); LOG.debug("Decoded message of type {} ({} bytes)", msg != null ? msg.getClass().getName() : msg, msgSize); out.add(msg); } finally { in.skipBytes(msgSize); } }
KryoMessageCodec extends ByteToMessageCodec<Object> { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { if (in.readableBytes() < 4) { return; } in.markReaderIndex(); int msgSize = in.readInt(); checkSize(msgSize); if (in.readableBytes() < msgSize) { in.resetReaderIndex(); return; } try { ByteBuffer nioBuffer = maybeDecrypt(in.nioBuffer(in.readerIndex(), msgSize)); Input kryoIn = new Input(new ByteBufferInputStream(nioBuffer)); Object msg = kryos.get().readClassAndObject(kryoIn); LOG.debug("Decoded message of type {} ({} bytes)", msg != null ? msg.getClass().getName() : msg, msgSize); out.add(msg); } finally { in.skipBytes(msgSize); } } }
KryoMessageCodec extends ByteToMessageCodec<Object> { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { if (in.readableBytes() < 4) { return; } in.markReaderIndex(); int msgSize = in.readInt(); checkSize(msgSize); if (in.readableBytes() < msgSize) { in.resetReaderIndex(); return; } try { ByteBuffer nioBuffer = maybeDecrypt(in.nioBuffer(in.readerIndex(), msgSize)); Input kryoIn = new Input(new ByteBufferInputStream(nioBuffer)); Object msg = kryos.get().readClassAndObject(kryoIn); LOG.debug("Decoded message of type {} ({} bytes)", msg != null ? msg.getClass().getName() : msg, msgSize); out.add(msg); } finally { in.skipBytes(msgSize); } } KryoMessageCodec(int maxMessageSize, Class<?>... messages); }
KryoMessageCodec extends ByteToMessageCodec<Object> { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { if (in.readableBytes() < 4) { return; } in.markReaderIndex(); int msgSize = in.readInt(); checkSize(msgSize); if (in.readableBytes() < msgSize) { in.resetReaderIndex(); return; } try { ByteBuffer nioBuffer = maybeDecrypt(in.nioBuffer(in.readerIndex(), msgSize)); Input kryoIn = new Input(new ByteBufferInputStream(nioBuffer)); Object msg = kryos.get().readClassAndObject(kryoIn); LOG.debug("Decoded message of type {} ({} bytes)", msg != null ? msg.getClass().getName() : msg, msgSize); out.add(msg); } finally { in.skipBytes(msgSize); } } KryoMessageCodec(int maxMessageSize, Class<?>... messages); @Override void channelInactive(ChannelHandlerContext ctx); }
KryoMessageCodec extends ByteToMessageCodec<Object> { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { if (in.readableBytes() < 4) { return; } in.markReaderIndex(); int msgSize = in.readInt(); checkSize(msgSize); if (in.readableBytes() < msgSize) { in.resetReaderIndex(); return; } try { ByteBuffer nioBuffer = maybeDecrypt(in.nioBuffer(in.readerIndex(), msgSize)); Input kryoIn = new Input(new ByteBufferInputStream(nioBuffer)); Object msg = kryos.get().readClassAndObject(kryoIn); LOG.debug("Decoded message of type {} ({} bytes)", msg != null ? msg.getClass().getName() : msg, msgSize); out.add(msg); } finally { in.skipBytes(msgSize); } } KryoMessageCodec(int maxMessageSize, Class<?>... messages); @Override void channelInactive(ChannelHandlerContext ctx);  }