andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Use (fromIndex, toIndex) instead of (index, length) for ByteBuf.forEachByte(...)

Browse Source 
- Related: #1378
This commit is contained in:
Trustin Lee 2013-06-27 17:30:19 +09:00
parent 792edf631c
commit b5bb36c087
7 changed files with 132 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
buffer/src/main/java/io/netty/buffer/AbstractByteBuf.java
View File

@ -1037,32 +1037,66 @@ public abstract class AbstractByteBuf implements ByteBuf {
@Override
public int forEachByte(ByteBufProcessor processor) {
int index = readerIndex;
int length = writerIndex - index;
return forEach0(index, length, processor);
return forEachByteAsc0(readerIndex, writerIndex, processor);
}
@Override
public int forEachByte(int index, int length, ByteBufProcessor processor) {
checkIndex(index, length);
return forEach0(index, length, processor);
public int forEachByte(int fromIndex, int toIndex, ByteBufProcessor processor) {
if (fromIndex < toIndex) {
return forEachByteAsc(fromIndex, toIndex, processor);
} else if (fromIndex > toIndex) {
return forEachByteDesc(fromIndex, toIndex, processor);
} else {
checkIndex(fromIndex);
return -1;
}
}
private int forEach0(int index, int length, ByteBufProcessor processor) {
private int forEachByteAsc(int fromIndex, int toIndex, ByteBufProcessor processor) {
if (fromIndex < 0 || toIndex > capacity()) {
throw new IndexOutOfBoundsException(String.format(
"fromIndex: %d, toIndex: %d (expected: 0 <= fromIndex < toIndex <= capacity(%d))",
fromIndex, toIndex, capacity()));
}
return forEachByteAsc0(fromIndex, toIndex, processor);
}
private int forEachByteAsc0(int fromIndex, int toIndex, ByteBufProcessor processor) {
if (processor == null) {
throw new NullPointerException("processor");
}
if (length == 0) {
return -1;
}
final int end = index + length;
int i = index;
int i = fromIndex;
try {
do {
i += processor.process(this, i, _getByte(i));
} while (i < end);
} while (i < toIndex);
} catch (Signal signal) {
signal.expect(ByteBufProcessor.ABORT);
return i;
} catch (Exception e) {
PlatformDependent.throwException(e);
}
return -1;
}
private int forEachByteDesc(int fromIndex, int toIndex, ByteBufProcessor processor) {
if (toIndex < -1 || fromIndex >= capacity()) {
throw new IndexOutOfBoundsException(String.format(
"fromIndex: %d, toIndex: %d (expected: -1 <= toIndex < fromIndex < capacity(%d))",
fromIndex, toIndex, capacity()));
}
if (processor == null) {
throw new NullPointerException("processor");
}
int i = fromIndex;
try {
do {
i -= processor.process(this, i, _getByte(i));
} while (i > toIndex);
} catch (Signal signal) {
signal.expect(ByteBufProcessor.ABORT);
return i;

11
buffer/src/main/java/io/netty/buffer/ByteBuf.java
View File

@ -1709,11 +1709,20 @@ public interface ByteBuf extends ReferenceCounted, Comparable<ByteBuf> {
/**
* Iterates over the specified area of this buffer with the specified {@code processor}.
* <ul>
* <li>If {@code fromIndex} is less than {@code toIndex}, the iteration is performed in an ascending order.
* ({@code [fromIndex, fromIndex&#41} i.e. {@code fromIndex}, {@code (fromIndex + 1)}, .. {@code (toIndex - 1)})
* </li>
* <li>If {@code fromIndex} is greater than {@code toIndex}, the iteration is performed in a descending order.
* ({@code [toIndex, fromIndex&#41} i.e. {@code (fromIndex)}, {@code (fromIndex - 1)}, ... {@code toIndex + 1})
* </li>
* <li>If {@code fromIndex} is equal to {@code toIndex}, {@code -1} is returned immediately.</li>
* </ul>
*
* @return {@code -1} if the processor iterated to or beyond the end of the specified area.
* If the {@code processor} raised {@link ByteBufProcessor#ABORT}, the last-visited index will be returned.
*/
int forEachByte(int index, int length, ByteBufProcessor processor);
int forEachByte(int fromIndex, int toIndex, ByteBufProcessor processor);
/**
* Returns a copy of this buffer's readable bytes. Modifying the content

6
buffer/src/main/java/io/netty/buffer/EmptyByteBuf.java
View File

@ -718,8 +718,10 @@ public final class EmptyByteBuf implements ByteBuf {
}
@Override
public int forEachByte(int index, int length, ByteBufProcessor processor) {
checkIndex(index, length);
public int forEachByte(int fromIndex, int toIndex, ByteBufProcessor processor) {
if (fromIndex != 0 && fromIndex != toIndex) {
throw new IndexOutOfBoundsException();
}
return -1;
}

4
buffer/src/main/java/io/netty/buffer/SwappedByteBuf.java
View File

@ -725,8 +725,8 @@ public final class SwappedByteBuf implements ByteBuf {
}
@Override
public int forEachByte(int index, int length, ByteBufProcessor processor) {
return buf.forEachByte(index, length, new SwappedByteBufProcessor(processor));
public int forEachByte(int fromIndex, int toIndex, ByteBufProcessor processor) {
return buf.forEachByte(fromIndex, toIndex, new SwappedByteBufProcessor(processor));
}
@Override

4
buffer/src/main/java/io/netty/buffer/UnreleasableByteBuf.java
View File

@ -731,8 +731,8 @@ final class UnreleasableByteBuf implements ByteBuf {
}
@Override
public int forEachByte(int index, int length, ByteBufProcessor processor) {
return buf.forEachByte(index, length, processor);
public int forEachByte(int fromIndex, int toIndex, ByteBufProcessor processor) {
return buf.forEachByte(fromIndex, toIndex, processor);
}
@Override

36
buffer/src/test/java/io/netty/buffer/AbstractByteBufTest.java
View File

@ -30,6 +30,7 @@ import java.util.HashSet;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import static io.netty.buffer.Unpooled.*;
import static io.netty.util.internal.EmptyArrays.*;
@ -1655,6 +1656,7 @@ public abstract class AbstractByteBufTest {
buffer.writeByte(i + 1);
}
final AtomicInteger lastIndex = new AtomicInteger();
buffer.setIndex(CAPACITY / 4, CAPACITY * 3 / 4);
assertThat(buffer.forEachByte(new ByteBufProcessor() {
int i = CAPACITY / 4;
@ -1663,10 +1665,13 @@ public abstract class AbstractByteBufTest {
public int process(ByteBuf buf, int index, byte value) throws Exception {
assertThat(value, is((byte) (i + 1)));
assertThat(index, is(i));
i++;
i ++;
lastIndex.set(index);
return 1;
}
}), is(-1));
assertThat(lastIndex.get(), is(CAPACITY * 3 / 4 - 1));
}
@Test
@ -1677,14 +1682,14 @@ public abstract class AbstractByteBufTest {
}
final int stop = CAPACITY / 2;
assertThat(buffer.forEachByte(CAPACITY / 3, CAPACITY / 3, new ByteBufProcessor() {
assertThat(buffer.forEachByte(CAPACITY / 3, CAPACITY * 2 / 3, new ByteBufProcessor() {
int i = CAPACITY / 3;
@Override
public int process(ByteBuf buf, int index, byte value) throws Exception {
assertThat(value, is((byte) (i + 1)));
assertThat(index, is(i));
i++;
i ++;
if (index == stop) {
throw ABORT;
@ -1693,4 +1698,29 @@ public abstract class AbstractByteBufTest {
}
}), is(stop));
}
@Test
public void testForEachByteDesc() {
buffer.clear();
for (int i = 0; i < CAPACITY; i ++) {
buffer.writeByte(i + 1);
}
final AtomicInteger lastIndex = new AtomicInteger();
buffer.setIndex(CAPACITY / 4, CAPACITY * 3 / 4);
assertThat(buffer.forEachByte(CAPACITY * 3 / 4, CAPACITY / 4, new ByteBufProcessor() {
int i = CAPACITY * 3 / 4;
@Override
public int process(ByteBuf buf, int index, byte value) throws Exception {
assertThat(value, is((byte) (i + 1)));
assertThat(index, is(i));
i --;
lastIndex.set(index);
return 1;
}
}), is(-1));
assertThat(lastIndex.get(), is(CAPACITY / 4 + 1));
}
}

40
codec/src/main/java/io/netty/handler/codec/ReplayingDecoderBuffer.java
View File

@ -378,7 +378,7 @@ final class ReplayingDecoderBuffer implements ByteBuf {
@Override
public int forEachByte(ByteBufProcessor processor) {
int ret = buffer.forEachByte(processor);
if (ret < 0 && !terminated) {
if (ret < 0) {
throw REPLAY;
} else {
return ret;
@ -386,18 +386,32 @@ final class ReplayingDecoderBuffer implements ByteBuf {
}
@Override
public int forEachByte(int index, int length, ByteBufProcessor processor) {
int writerIndex = buffer.writerIndex();
public int forEachByte(int fromIndex, int toIndex, ByteBufProcessor processor) {
if (fromIndex < toIndex) {
return forEachByteAsc(fromIndex, toIndex, processor);
} else if (fromIndex > toIndex) {
return forEachByteDesc(fromIndex, toIndex, processor);
} else {
checkIndex(fromIndex, 0);
return -1;
}
}
if (index >= writerIndex) {
private int forEachByteAsc(int fromIndex, int toIndex, ByteBufProcessor processor) {
if (fromIndex < 0) {
throw new IndexOutOfBoundsException("fromIndex: " + fromIndex);
}
final int writerIndex = buffer.writerIndex();
if (fromIndex >= writerIndex) {
throw REPLAY;
}
if (terminated || index + length <= writerIndex) {
return buffer.forEachByte(index, length, processor);
if (toIndex <= writerIndex) {
return buffer.forEachByte(fromIndex, toIndex, processor);
}
int ret = buffer.forEachByte(index, writerIndex - index, processor);
int ret = buffer.forEachByte(fromIndex, writerIndex, processor);
if (ret < 0) {
throw REPLAY;
} else {
@ -405,6 +419,18 @@ final class ReplayingDecoderBuffer implements ByteBuf {
}
}
private int forEachByteDesc(int fromIndex, int toIndex, ByteBufProcessor processor) {
if (toIndex < -1) {
throw new IndexOutOfBoundsException("toIndex: " + toIndex);
}
if (fromIndex >= buffer.writerIndex()) {
throw REPLAY;
}
return buffer.forEachByte(fromIndex, toIndex, processor);
}
@Override
public ByteBuf markReaderIndex() {
buffer.markReaderIndex();