andreacavalli/netty-incubator-buffer-api
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #26 from netty/update-examples

Browse Source 
Update examples
This commit is contained in:
Chris Vest 2021-01-25 13:57:35 +01:00 committed by GitHub
commit 1a741baaf0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 292 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/io/netty/buffer/api/Buf.java
View File

@ -562,7 +562,7 @@ public interface Buf extends Rc<Buf>, BufAccessors {
* {@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.
@ -601,5 +601,5 @@ public interface Buf extends Rc<Buf>, BufAccessors {
* {@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
src/main/java/io/netty/buffer/api/ComponentProcessor.java
View File

@ -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}.
@ -41,14 +41,14 @@ public interface 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}.
@ -63,7 +63,7 @@ public interface 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;
}
/**

6
src/main/java/io/netty/buffer/api/CompositeBuf.java
View File

@ -801,7 +801,8 @@ final class CompositeBuf extends RcSupport<Buf, CompositeBuf> implements Buf {
}
@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) {
@ -819,7 +820,8 @@ final class CompositeBuf extends RcSupport<Buf, CompositeBuf> implements Buf {
}
@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) {

6
src/main/java/io/netty/buffer/api/memseg/MemSegBuf.java
View File

@ -548,13 +548,15 @@ class MemSegBuf extends RcSupport<Buf, MemSegBuf> implements Buf, ReadableCompon
}
@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 {
checkWrite(writerOffset(), Math.max(1, writableBytes()));
return processor.process(initialIndex, this)? 1 : -1;
}

43
src/test/java/io/netty/buffer/api/examples/FileCopyExample.java
View File

@ -19,7 +19,6 @@ import io.netty.buffer.api.Allocator;
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;
@ -34,27 +33,23 @@ import static java.nio.file.StandardOpenOption.WRITE;
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());
}
@ -64,19 +59,17 @@ public final class FileCopyExample {
});
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);

260
src/test/java/io/netty/buffer/api/examples/SendExample.java Normal file
View File

@ -0,0 +1,260 @@
/*
* Copyright 2021 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.buffer.api.examples;
import io.netty.buffer.api.Allocator;
import io.netty.buffer.api.Buf;
import io.netty.buffer.api.Send;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import static java.util.concurrent.Executors.newFixedThreadPool;
import static java.util.concurrent.Executors.newSingleThreadExecutor;
public class SendExample {
static final class Ex1 {
public static void main(String[] args) throws Exception {
ExecutorService executor =
newSingleThreadExecutor();
Allocator allocator = Allocator.heap();
var future = beginTask(executor, allocator);
future.get();
allocator.close();
executor.shutdown();
}
private static Future<?> beginTask(
ExecutorService executor, Allocator allocator) {
try (Buf buf = allocator.allocate(32)) {
// !!! pit-fall: buffer life-time ends before task completes
return executor.submit(new Task(buf));
}
}
private static class Task implements Runnable {
private final Buf buf;
Task(Buf buf) {
this.buf = buf;
}
@Override
public void run() {
// !!! danger: access out-side owning thread.
while (buf.writableBytes() > 0) {
buf.writeByte((byte) 42);
}
}
}
}
static final class Ex2 {
public static void main(String[] args) throws Exception {
ExecutorService executor = newSingleThreadExecutor();
Allocator allocator = Allocator.heap();
var future = beginTask(executor, allocator);
future.get();
allocator.close();
executor.shutdown();
}
private static Future<?> beginTask(
ExecutorService executor, Allocator allocator) {
try (Buf buf = allocator.allocate(32)) {
// !!! pit-fall: Rc decrement in other thread.
return executor.submit(new Task(buf.acquire()));
}
}
private static class Task implements Runnable {
private final Buf buf;
Task(Buf buf) {
this.buf = buf;
}
@Override
public void run() {
try (buf) {
// !!! danger: access out-side owning thread.
while (buf.writableBytes() > 0) {
buf.writeByte((byte) 42);
}
}
}
}
}
static final class Ex3 {
public static void main(String[] args) throws Exception {
ExecutorService executor = newSingleThreadExecutor();
Allocator allocator = Allocator.heap();
var future = beginTask(executor, allocator);
future.get();
allocator.close();
executor.shutdown();
}
private static Future<?> beginTask(
ExecutorService executor, Allocator allocator) {
try (Buf buf = allocator.allocate(32)) {
return executor.submit(new Task(buf.send()));
}
}
private static class Task implements Runnable {
private final Send<Buf> send;
Task(Send<Buf> send) {
this.send = send;
}
@Override
public void run() {
try (Buf buf = send.receive()) {
while (buf.writableBytes() > 0) {
buf.writeByte((byte) 42);
}
}
}
}
}
static final class Ex4 {
public static void main(String[] args) throws Exception {
ExecutorService executor = newFixedThreadPool(4);
Allocator allocator = Allocator.heap();
try (Buf buf = allocator.allocate(4096)) {
// !!! pit-fall: Rc decrement in other thread.
var futA = executor.submit(new Task(buf.slice(0, 1024)));
var futB = executor.submit(new Task(buf.slice(1024, 1024)));
var futC = executor.submit(new Task(buf.slice(2048, 1024)));
var futD = executor.submit(new Task(buf.slice(3072, 1024)));
futA.get();
futB.get();
futC.get();
futD.get();
}
allocator.close();
executor.shutdown();
}
private static class Task implements Runnable {
private final Buf slice;
Task(Buf slice) {
this.slice = slice;
}
@Override
public void run() {
try (slice) {
while (slice.writableBytes() > 0) {
slice.writeByte((byte) 42);
}
}
}
}
}
static final class Ex5 {
public static void main(String[] args) throws Exception {
ExecutorService executor = newFixedThreadPool(4);
Allocator allocator = Allocator.heap();
try (Buf buf = allocator.allocate(4096);
Buf sliceA = buf.slice(0, 1024);
Buf sliceB = buf.slice(1024, 1024);
Buf sliceC = buf.slice(2048, 1024);
Buf sliceD = buf.slice(3072, 1024)) {
var futA = executor.submit(new Task(sliceA));
var futB = executor.submit(new Task(sliceB));
var futC = executor.submit(new Task(sliceC));
var futD = executor.submit(new Task(sliceD));
futA.get();
futB.get();
futC.get();
futD.get();
}
allocator.close();
executor.shutdown();
}
private static class Task implements Runnable {
private final Buf slice;
Task(Buf slice) {
this.slice = slice;
}
@Override
public void run() {
while (slice.writableBytes() > 0) {
slice.writeByte((byte) 42);
}
}
}
}
static final class Ex6 {
public static void main(String[] args) throws Exception {
ExecutorService executor = newFixedThreadPool(4);
Allocator allocator = Allocator.heap();
try (Buf buf = allocator.allocate(4096)) {
var futA = executor.submit(new Task(buf.writerOffset(1024).bifurcate().send()));
var futB = executor.submit(new Task(buf.writerOffset(1024).bifurcate().send()));
var futC = executor.submit(new Task(buf.writerOffset(1024).bifurcate().send()));
var futD = executor.submit(new Task(buf.send()));
futA.get();
futB.get();
futC.get();
futD.get();
}
allocator.close();
executor.shutdown();
}
private static class Task implements Runnable {
private final Send<Buf> send;
Task(Send<Buf> send) {
this.send = send;
}
@Override
public void run() {
try (Buf buf = send.receive().writerOffset(0)) {
while (buf.writableBytes() > 0) {
buf.writeByte((byte) 42);
}
}
}
}
}
}