Skip to content
This repository was archived by the owner on Dec 12, 2022. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/main/java/io/netty/buffer/api/Buf.java
Original file line number Diff line number Diff line change
Expand Up @@ -562,7 +562,7 @@ default void ensureWritable(int size) {
* {@link ReadableComponentProcessor#process(int, ReadableComponent)} returned {@code false}.
* In any case, the number of components processed may be less than {@link #countComponents()}.
*/
int forEachReadable(int initialIndex, ReadableComponentProcessor processor);
<E extends Exception> int forEachReadable(int initialIndex, ReadableComponentProcessor<E> processor) throws E;

/**
* Process all writable components of this buffer, and return the number of components processed.
Expand Down Expand Up @@ -601,5 +601,5 @@ default void ensureWritable(int size) {
* {@link WritableComponentProcessor#process(int, WritableComponent)} returned {@code false}.
* In any case, the number of components processed may be less than {@link #countComponents()}.
*/
int forEachWritable(int initialIndex, WritableComponentProcessor processor);
<E extends Exception> int forEachWritable(int initialIndex, WritableComponentProcessor<E> processor) throws E;
}
8 changes: 4 additions & 4 deletions src/main/java/io/netty/buffer/api/ComponentProcessor.java
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ public interface ComponentProcessor {
* A processor of {@linkplain ReadableComponent readable components}.
*/
@FunctionalInterface
interface ReadableComponentProcessor extends ComponentProcessor {
interface ReadableComponentProcessor<E extends Exception> extends ComponentProcessor {
/**
* Process the given component at the given index in the
* {@link Buf#forEachReadable(int, ReadableComponentProcessor) iteration}.
Expand All @@ -41,14 +41,14 @@ interface ReadableComponentProcessor extends ComponentProcessor {
* @return {@code true} if the iteration should continue and more components should be processed, otherwise
* {@code false} to stop the iteration early.
*/
boolean process(int index, ReadableComponent component);
boolean process(int index, ReadableComponent component) throws E;
}

/**
* A processor of {@linkplain WritableComponent writable components}.
*/
@FunctionalInterface
interface WritableComponentProcessor extends ComponentProcessor {
interface WritableComponentProcessor<E extends Exception> extends ComponentProcessor {
/**
* Process the given component at the given index in the
* {@link Buf#forEachWritable(int, WritableComponentProcessor)} iteration}.
Expand All @@ -63,7 +63,7 @@ interface WritableComponentProcessor extends ComponentProcessor {
* @return {@code true} if the iteration should continue and more components should be processed, otherwise
* {@code false} to stop the iteration early.
*/
boolean process(int index, WritableComponent component);
boolean process(int index, WritableComponent component) throws E;
}

/**
Expand Down
6 changes: 4 additions & 2 deletions src/main/java/io/netty/buffer/api/CompositeBuf.java
Original file line number Diff line number Diff line change
Expand Up @@ -801,7 +801,8 @@ public int countWritableComponents() {
}

@Override
public int forEachReadable(int initialIndex, ReadableComponentProcessor processor) {
public <E extends Exception> int forEachReadable(int initialIndex, ReadableComponentProcessor<E> processor)
throws E {
checkReadBounds(readerOffset(), Math.max(1, readableBytes()));
int visited = 0;
for (Buf buf : bufs) {
Expand All @@ -819,7 +820,8 @@ public int forEachReadable(int initialIndex, ReadableComponentProcessor processo
}

@Override
public int forEachWritable(int initialIndex, WritableComponentProcessor processor) {
public <E extends Exception> int forEachWritable(int initialIndex, WritableComponentProcessor<E> processor)
throws E {
checkWriteBounds(writerOffset(), Math.max(1, writableBytes()));
int visited = 0;
for (Buf buf : bufs) {
Expand Down
6 changes: 4 additions & 2 deletions src/main/java/io/netty/buffer/api/memseg/MemSegBuf.java
Original file line number Diff line number Diff line change
Expand Up @@ -548,13 +548,15 @@ public int countWritableComponents() {
}

@Override
public int forEachReadable(int initialIndex, ReadableComponentProcessor processor) {
public <E extends Exception> int forEachReadable(int initialIndex, ReadableComponentProcessor<E> processor)
throws E {
checkRead(readerOffset(), Math.max(1, readableBytes()));
return processor.process(initialIndex, this)? 1 : -1;
}

@Override
public int forEachWritable(int initialIndex, WritableComponentProcessor processor) {
public <E extends Exception> int forEachWritable(int initialIndex, WritableComponentProcessor<E> processor)
throws E {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thats neat!

checkWrite(writerOffset(), Math.max(1, writableBytes()));
return processor.process(initialIndex, this)? 1 : -1;
}
Expand Down
43 changes: 18 additions & 25 deletions src/test/java/io/netty/buffer/api/examples/FileCopyExample.java
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import io.netty.buffer.api.Buf;
import io.netty.buffer.api.Send;

import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.util.concurrent.ArrayBlockingQueue;
Expand All @@ -34,27 +33,23 @@
public final class FileCopyExample {
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(2);
ArrayBlockingQueue<Object> queue = new ArrayBlockingQueue<>(8);
Object done = new Object();
ArrayBlockingQueue<Send<Buf>> queue = new ArrayBlockingQueue<>(8);
try (Allocator allocator = Allocator.pooledDirect();
var input = FileChannel.open(Path.of("/dev/urandom"), READ);
var output = FileChannel.open(Path.of("random.bin"), CREATE, TRUNCATE_EXISTING, WRITE)) {
Send<Buf> done = allocator.compose().send();

var reader = executor.submit(() -> {
var buf = ByteBuffer.allocateDirect(1024);
for (int i = 0; i < 1024; i++) {
buf.clear();
while (buf.hasRemaining()) {
int r = input.read(buf);
System.out.println("r = " + r);
System.out.println("buf = " + buf);
}
buf.flip();
try (Buf in = allocator.allocate(1024)) {
System.out.println("in = " + in);
while (buf.hasRemaining()) {
in.writeByte(buf.get());
}
in.forEachWritable(0, (index, component) -> {
var bb = component.writableBuffer();
while (bb.hasRemaining()) {
input.read(bb);
}
return true;
});
System.out.println("Sending " + in.readableBytes() + " bytes.");
queue.put(in.send());
}
Expand All @@ -64,19 +59,17 @@ public static void main(String[] args) throws Exception {
});

var writer = executor.submit(() -> {
var buf = ByteBuffer.allocateDirect(1024);
Object msg;
while ((msg = queue.take()) != done) {
buf.clear();
@SuppressWarnings("unchecked")
Send<Buf> send = (Send<Buf>) msg;
Send<Buf> send;
while ((send = queue.take()) != done) {
try (Buf out = send.receive()) {
System.out.println("Received " + out.readableBytes() + " bytes.");
out.copyInto(0, buf, 0, out.readableBytes());
buf.position(0).limit(out.readableBytes());
}
while (buf.hasRemaining()) {
output.write(buf);
out.forEachReadable(0, (index, component) -> {
var bb = component.readableBuffer();
while (bb.hasRemaining()) {
output.write(bb);
}
return true;
});
}
}
output.force(true);
Expand Down
Loading