Author: trustin Date: 2009-11-20 09:35:17 -0500 (Fri, 20 Nov 2009) New Revision: 8338 Modified: branches/20-optimisation/src/main/org/hornetq/integration/transports/netty/HornetQFrameDecoder2.java Log: Fixed a bug in HornetQFrameDecoder2 where the first four bytes length header is omitted Modified: branches/20-optimisation/src/main/org/hornetq/integration/transports/netty/HornetQFrameDecoder2.java =================================================================== --- branches/20-optimisation/src/main/org/hornetq/integration/transports/netty/HornetQFrameDecoder2.java 2009-11-20 14:20:03 UTC (rev 8337) +++ branches/20-optimisation/src/main/org/hornetq/integration/transports/netty/HornetQFrameDecoder2.java 2009-11-20 14:35:17 UTC (rev 8338) @@ -64,7 +64,6 @@ // treatment because it is the only message that spans over the two // buffers. final int length; - final ChannelBuffer frame; switch (previousData.readableBytes()) { case 1: length = (previousData.getUnsignedByte(previousData.readerIndex()) << 24) | @@ -72,9 +71,6 @@ if (in.readableBytes() - 3 < length) { append(in, length); return; - } else { - frame = in.slice(in.readerIndex() + 3, length); - in.skipBytes(length + 3); } break; case 2: @@ -83,9 +79,6 @@ if (in.readableBytes() - 2 < length) { append(in, length); return; - } else { - frame = in.slice(in.readerIndex() + 2, length); - in.skipBytes(length + 2); } break; case 3: @@ -94,9 +87,6 @@ if (in.readableBytes() - 1 < length) { append(in, length); return; - } else { - frame = in.slice(in.readerIndex() + 1, length); - in.skipBytes(length + 1); } break; case 4: @@ -104,9 +94,6 @@ if (in.readableBytes() - 4 < length) { append(in, length); return; - } else { - frame = in.slice(in.readerIndex(), length); - in.skipBytes(length); } break; default: @@ -114,22 +101,23 @@ if (in.readableBytes() + previousData.readableBytes() - 4 < length) { append(in, length); return; - } else { - if (previousData instanceof DynamicChannelBuffer) { - // It's safe to reuse the current dynamic buffer - // because previousData will be reassigned to - // EMPTY_BUFFER or 'in' later. - previousData.skipBytes(4); - previousData.writeBytes(in, length - previousData.readableBytes()); - frame = previousData.slice(); - } else { - frame = ChannelBuffers.buffer(length); - frame.writeBytes(previousData, previousData.readerIndex() + 4, previousData.readableBytes() - 4); - frame.writeBytes(in, length - frame.writerIndex()); - } } } + final ChannelBuffer frame; + if (previousData instanceof DynamicChannelBuffer) { + // It's safe to reuse the current dynamic buffer + // because previousData will be reassigned to + // EMPTY_BUFFER or 'in' later. + previousData.writeBytes(in, length + 4 - previousData.readableBytes()); + frame = previousData.slice(); + } else { + frame = ChannelBuffers.buffer(length + 4); + frame.writeBytes(previousData, previousData.readerIndex(), previousData.readableBytes()); + frame.writeBytes(in, length + 4 - frame.writerIndex()); + } + + frame.skipBytes(4); Channels.fireMessageReceived(ctx, frame); if (!in.readable()) { @@ -167,8 +155,9 @@ break; } - ChannelBuffer frame = in.slice(in.readerIndex() + SIZE_INT, length); + ChannelBuffer frame = in.slice(in.readerIndex(), length + SIZE_INT); in.skipBytes(SIZE_INT + length); + frame.skipBytes(SIZE_INT); Channels.fireMessageReceived(ctx, frame); } } @@ -179,6 +168,7 @@ // a client is very slow. (e.g.sending each byte one by one) if (previousData instanceof DynamicChannelBuffer) { + previousData.discardReadBytes(); previousData.writeBytes(in); } else @@ -206,27 +196,32 @@ } }); } + + final int MSG_LEN = 40; final DecoderEmbedder<ChannelBuffer> decoder = new DecoderEmbedder<ChannelBuffer>(handler); - ChannelBuffer src = ChannelBuffers.buffer(30000 * 1004); + ChannelBuffer src = ChannelBuffers.buffer(30000 * (MSG_LEN + 4)); while (src.writerIndex() < src.capacity()) { - src.writeInt(1000); - src.writeZero(1000); + src.writeInt(MSG_LEN); + for (int i = 0; i < MSG_LEN; i ++) { + src.writeByte((byte) i); + } + //src.writeZero(1000); } Random rand = new Random(); List<ChannelBuffer> packets = new ArrayList<ChannelBuffer>(); for (int i = 0; i < src.capacity();) { - int length = Math.min(rand.nextInt(3000), src.capacity() - i); + int length = Math.min(rand.nextInt(10000), src.capacity() - i); packets.add(src.copy(i, length)); i += length; } long startTime = System.nanoTime(); - for (int i = 0; i < 100; i ++) { + for (int i = 0; i < 10; i ++) { int cnt = 0; for (ChannelBuffer p: packets) { decoder.offer(p.duplicate()); @@ -235,9 +230,14 @@ if (frame == null) { break; } - if (frame.readableBytes() != 1000) { + if (frame.readableBytes() != MSG_LEN) { System.out.println("ARGH 1: " + frame.readableBytes()); } + for (int j = 0; j < MSG_LEN; j ++) { + if (frame.readByte() != (byte) j) { + System.out.println("ARGH 3"); + } + } cnt ++; } }