clazz);
+
/**
* Creates and initializes a new {@link Node} instance using the provided configuration.
*
diff --git a/api/src/main/java/io/bosonnetwork/utils/Functional.java b/api/src/main/java/io/bosonnetwork/utils/Functional.java
index 5917ca2b..a91f4db0 100644
--- a/api/src/main/java/io/bosonnetwork/utils/Functional.java
+++ b/api/src/main/java/io/bosonnetwork/utils/Functional.java
@@ -129,4 +129,24 @@ private static void throwAsUnchecked(Throwable t) {
private static void asUnchecked(Throwable t) throws T {
throw (T) t;
}
+
+
+ /**
+ * Rethrows the {@code e} without being declared.
+ *
+ * This uses the "sneaky-throw" idiom — the original exception is rethrown
+ * as-is using a generic-erasure trick, not wrapped in a {@code RuntimeException}. A caller's
+ * {@code catch (IOException e)} clause will still match an {@code IOException} thrown from
+ * this method, even though this method's signature does not declare it. Use this only where
+ * declaring the checked type is impossible (e.g. inside a {@code Function} / {@code Supplier})
+ * and the caller is prepared for the actual exception type to surface.
+ *
+ * @param the exception type
+ * @param e the exception to throw
+ * @throws E the exception
+ */
+ @SuppressWarnings("unchecked")
+ public static void sneakyThrow(Throwable e) throws E {
+ throw (E) e;
+ }
}
\ No newline at end of file
diff --git a/api/src/main/java/io/bosonnetwork/vertx/AsyncInputStream.java b/api/src/main/java/io/bosonnetwork/vertx/AsyncInputStream.java
new file mode 100644
index 00000000..33994c25
--- /dev/null
+++ b/api/src/main/java/io/bosonnetwork/vertx/AsyncInputStream.java
@@ -0,0 +1,256 @@
+/*
+ * Copyright (c) 2023 - bosonnetwork.io
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.bosonnetwork.vertx;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Objects;
+
+import io.vertx.core.Context;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.streams.ReadStream;
+
+/**
+ * A Vert.x {@link ReadStream} that adapts a blocking {@link InputStream}.
+ *
+ * Each chunk is read in a short {@link Context#executeBlocking(java.util.concurrent.Callable, boolean)
+ * executeBlocking} task and the next read is scheduled only when there is outstanding demand, so —
+ * unlike a naive adapter that loops on a worker thread for the stream's lifetime — no worker thread is
+ * held while the stream is idle or paused. Back-pressure is honored per chunk: {@link #pause()},
+ * {@link #resume()} and {@link #fetch(long)} control how many further chunks are read and delivered.
+ *
+ *
Threading
+ * The stream binds to a Vert.x {@link Context} the first time a data handler is set, and from then on
+ * all reads complete, and all handlers are invoked, on that context. As with the built-in Vert.x
+ * streams, its methods are expected to be called from that same context.
+ *
+ * Resource ownership
+ * The wrapped {@link InputStream} is closed when the stream ends, fails, or {@link #close()} is
+ * called — but only if {@code closeInput} was set ({@code true} for the two-argument constructor).
+ */
+public class AsyncInputStream implements ReadStream {
+ private static final int DEFAULT_CHUNK_SIZE = 8192;
+
+ private final Vertx vertx;
+ private final InputStream input;
+ private final int chunkSize;
+ private final boolean closeInput;
+
+ // All mutable state below is confined to {@link #context} once bound.
+ private Context context;
+
+ private Handler dataHandler;
+ private Handler endHandler;
+ private Handler exceptionHandler;
+
+ // Outstanding number of chunks to deliver; Long.MAX_VALUE means "flowing" (unbounded).
+ private long demand = Long.MAX_VALUE;
+ private boolean readInProgress;
+ private boolean closed;
+ private boolean inputClosed;
+
+ /**
+ * Creates a new {@code AsyncInputStream} with the default chunk size (8192 bytes) that closes the
+ * wrapped {@link InputStream} when finished.
+ *
+ * @param vertx the Vert.x instance
+ * @param input the input stream to read from
+ */
+ public AsyncInputStream(Vertx vertx, InputStream input) {
+ this(vertx, input, DEFAULT_CHUNK_SIZE, true);
+ }
+
+ /**
+ * Creates a new {@code AsyncInputStream}.
+ *
+ * @param vertx the Vert.x instance
+ * @param input the input stream to read from
+ * @param chunkSize the size of the buffer used for each read (must be {@code > 0})
+ * @param closeInput whether to close the input stream when the stream ends, fails or is closed
+ */
+ public AsyncInputStream(Vertx vertx, InputStream input, int chunkSize, boolean closeInput) {
+ this.vertx = Objects.requireNonNull(vertx, "vertx");
+ this.input = Objects.requireNonNull(input, "input");
+ if (chunkSize <= 0)
+ throw new IllegalArgumentException("chunkSize must be > 0");
+
+ this.chunkSize = chunkSize;
+ this.closeInput = closeInput;
+ }
+
+ @Override
+ public AsyncInputStream exceptionHandler(Handler handler) {
+ this.exceptionHandler = handler;
+ return this;
+ }
+
+ @Override
+ public AsyncInputStream handler(Handler handler) {
+ this.dataHandler = handler;
+ if (handler != null && !closed) {
+ if (context == null)
+ context = vertx.getOrCreateContext();
+ doRead();
+ }
+ return this;
+ }
+
+ @Override
+ public AsyncInputStream pause() {
+ demand = 0L;
+ return this;
+ }
+
+ @Override
+ public AsyncInputStream resume() {
+ return fetch(Long.MAX_VALUE);
+ }
+
+ @Override
+ public AsyncInputStream fetch(long amount) {
+ if (amount < 0)
+ throw new IllegalArgumentException("amount must be >= 0");
+ if (closed)
+ return this;
+
+ // saturating add
+ demand += amount;
+ if (demand < 0)
+ demand = Long.MAX_VALUE;
+
+ if (dataHandler != null) {
+ if (context == null)
+ context = vertx.getOrCreateContext();
+ doRead();
+ }
+ return this;
+ }
+
+ @Override
+ public AsyncInputStream endHandler(Handler handler) {
+ this.endHandler = handler;
+ return this;
+ }
+
+ /**
+ * Closes the stream, stopping any further reads and (if configured) closing the wrapped
+ * {@link InputStream}. Idempotent. No more data, end or exception events are delivered afterwards.
+ */
+ public void close() {
+ Context ctx = context;
+ if (ctx != null && Vertx.currentContext() != ctx)
+ ctx.runOnContext(v -> doClose());
+ else
+ doClose();
+ }
+
+ private void doClose() {
+ if (closed)
+ return;
+ closed = true;
+ dataHandler = null;
+ // If a blocking read is in flight, defer closing the input to its completion to avoid closing
+ // the stream concurrently with a read on the worker thread.
+ if (!readInProgress)
+ closeInputQuietly();
+ }
+
+ private void doRead() {
+ if (closed || readInProgress || demand == 0L || dataHandler == null)
+ return;
+
+ readInProgress = true;
+ byte[] buf = new byte[chunkSize];
+ context.executeBlocking(() -> input.read(buf), false).onComplete(ar -> {
+ readInProgress = false;
+
+ if (closed) {
+ closeInputQuietly();
+ return;
+ }
+ if (ar.failed()) {
+ handleException(ar.cause());
+ return;
+ }
+
+ int len = ar.result();
+ if (len < 0) {
+ handleEnd();
+ return;
+ }
+ if (len > 0) {
+ if (demand != Long.MAX_VALUE)
+ demand--;
+ Handler handler = dataHandler;
+ if (handler != null) {
+ try {
+ handler.handle(Buffer.buffer(len).appendBytes(buf, 0, len));
+ } catch (Throwable t) {
+ handleException(t);
+ return;
+ }
+ }
+ }
+
+ if (!closed && demand > 0L)
+ context.runOnContext(v -> doRead());
+ });
+ }
+
+ private void handleEnd() {
+ if (closed)
+ return;
+ closed = true;
+ dataHandler = null;
+ closeInputQuietly();
+ Handler handler = endHandler;
+ if (handler != null)
+ handler.handle(null);
+ }
+
+ private void handleException(Throwable cause) {
+ if (closed)
+ return;
+ closed = true;
+ dataHandler = null;
+ closeInputQuietly();
+ Handler handler = exceptionHandler;
+ if (handler != null)
+ handler.handle(cause);
+ }
+
+ private void closeInputQuietly() {
+ if (inputClosed)
+ return;
+ inputClosed = true;
+ if (closeInput) {
+ try {
+ input.close();
+ } catch (IOException ignore) {
+ // best-effort close
+ }
+ }
+ }
+}
diff --git a/api/src/main/java/io/bosonnetwork/vertx/AsyncOutputStream.java b/api/src/main/java/io/bosonnetwork/vertx/AsyncOutputStream.java
new file mode 100644
index 00000000..0238fe49
--- /dev/null
+++ b/api/src/main/java/io/bosonnetwork/vertx/AsyncOutputStream.java
@@ -0,0 +1,253 @@
+/*
+ * Copyright (c) 2023 - bosonnetwork.io
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.bosonnetwork.vertx;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.Objects;
+
+import io.vertx.core.Context;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.streams.WriteStream;
+
+/**
+ * A Vert.x {@link WriteStream} that adapts a blocking {@link OutputStream}.
+ *
+ * Buffers are written one at a time, each in a short
+ * {@link Context#executeBlocking(java.util.concurrent.Callable, boolean) executeBlocking} task, so —
+ * unlike a naive adapter — no worker thread is held while the stream is idle. Write ordering is
+ * preserved (only one write is in flight at a time) and back-pressure is reported through
+ * {@link #writeQueueFull()} / {@link #drainHandler(Handler)}: the queue is measured in buffered bytes
+ * and bounded by {@link #setWriteQueueMaxSize(int)}.
+ *
+ *
Threading
+ * The stream binds to a Vert.x {@link Context} on its first write, and from then on all writes
+ * complete, and all handlers are invoked, on that context.
+ *
+ * Resource ownership
+ * On {@link #end()} the wrapped {@link OutputStream} is flushed and, if {@code closeOutput} was set
+ * ({@code true} for the two-argument constructor), closed.
+ */
+public class AsyncOutputStream implements WriteStream {
+ private static final int DEFAULT_MAX_QUEUE_BYTES = 64 * 1024;
+
+ private final Vertx vertx;
+ private final OutputStream output;
+ private final boolean closeOutput;
+
+ // All mutable state below is confined to {@link #context} once bound.
+ private Context context;
+ private Handler exceptionHandler;
+ private Handler drainHandler;
+
+ private final Deque pending = new ArrayDeque<>();
+ private long pendingBytes;
+ private int maxQueueBytes = DEFAULT_MAX_QUEUE_BYTES;
+ private boolean wasFull;
+ private boolean writeInProgress;
+ private boolean ending;
+ private boolean closed;
+ private Promise endPromise;
+
+ private record PendingWrite(Buffer buffer, Promise promise) {}
+
+ /**
+ * Creates a new {@code AsyncOutputStream} that flushes and closes the wrapped {@link OutputStream}
+ * on {@link #end()}.
+ *
+ * @param vertx the Vert.x instance
+ * @param output the output stream to write to
+ */
+ public AsyncOutputStream(Vertx vertx, OutputStream output) {
+ this(vertx, output, true);
+ }
+
+ /**
+ * Creates a new {@code AsyncOutputStream}.
+ *
+ * @param vertx the Vert.x instance
+ * @param output the output stream to write to
+ * @param closeOutput whether to close the output stream on {@link #end()} (it is always flushed)
+ */
+ public AsyncOutputStream(Vertx vertx, OutputStream output, boolean closeOutput) {
+ this.vertx = Objects.requireNonNull(vertx, "vertx");
+ this.output = Objects.requireNonNull(output, "output");
+ this.closeOutput = closeOutput;
+ }
+
+ @Override
+ public AsyncOutputStream exceptionHandler(Handler handler) {
+ this.exceptionHandler = handler;
+ return this;
+ }
+
+ @Override
+ public Future write(Buffer data) {
+ Objects.requireNonNull(data, "data");
+ Promise promise = Promise.promise();
+ execute(() -> {
+ if (closed || ending) {
+ promise.fail(new IllegalStateException("Stream is closed"));
+ return;
+ }
+ pending.add(new PendingWrite(data, promise));
+ pendingBytes += data.length();
+ pump();
+ });
+ return promise.future();
+ }
+
+ @Override
+ public Future end() {
+ Promise promise = Promise.promise();
+ execute(() -> {
+ if (closed) {
+ promise.complete();
+ return;
+ }
+ if (ending) {
+ promise.fail(new IllegalStateException("Stream is already ending"));
+ return;
+ }
+ ending = true;
+ endPromise = promise;
+ if (!writeInProgress && pending.isEmpty())
+ finish();
+ });
+ return promise.future();
+ }
+
+ @Override
+ public AsyncOutputStream setWriteQueueMaxSize(int maxSize) {
+ if (maxSize < 1)
+ throw new IllegalArgumentException("maxSize must be >= 1");
+ this.maxQueueBytes = maxSize;
+ return this;
+ }
+
+ @Override
+ public boolean writeQueueFull() {
+ boolean full = pendingBytes >= maxQueueBytes;
+ if (full)
+ wasFull = true;
+ return full;
+ }
+
+ @Override
+ public AsyncOutputStream drainHandler(Handler handler) {
+ this.drainHandler = handler;
+ return this;
+ }
+
+ private void execute(Runnable action) {
+ if (context == null)
+ context = vertx.getOrCreateContext();
+ if (Vertx.currentContext() == context)
+ action.run();
+ else
+ context.runOnContext(v -> action.run());
+ }
+
+ private void pump() {
+ if (writeInProgress || pending.isEmpty())
+ return;
+
+ PendingWrite w = pending.poll();
+ pendingBytes -= w.buffer().length();
+ writeInProgress = true;
+ byte[] bytes = w.buffer().getBytes();
+ context.executeBlocking(() -> {
+ output.write(bytes);
+ return null;
+ }, false).onComplete(ar -> {
+ writeInProgress = false;
+ if (ar.failed()) {
+ w.promise().fail(ar.cause());
+ fail(ar.cause());
+ return;
+ }
+
+ w.promise().complete();
+ callDrainIfNeeded();
+
+ if (!pending.isEmpty())
+ pump();
+ else if (ending)
+ finish();
+ });
+ }
+
+ private void callDrainIfNeeded() {
+ if (wasFull && drainHandler != null && pendingBytes <= maxQueueBytes / 2) {
+ wasFull = false;
+ drainHandler.handle(null);
+ }
+ }
+
+ private void finish() {
+ context.executeBlocking(() -> {
+ output.flush();
+ if (closeOutput)
+ output.close();
+ return null;
+ }, false).onComplete(ar -> {
+ closed = true;
+ if (ar.failed())
+ endPromise.fail(ar.cause());
+ else
+ endPromise.complete();
+ });
+ }
+
+ private void fail(Throwable cause) {
+ if (closed)
+ return;
+ closed = true;
+
+ PendingWrite w;
+ while ((w = pending.poll()) != null)
+ w.promise().fail(cause);
+ pendingBytes = 0;
+
+ if (closeOutput) {
+ try {
+ output.close();
+ } catch (IOException ignore) {
+ // best-effort close
+ }
+ }
+
+ if (ending && endPromise != null)
+ endPromise.fail(cause);
+
+ Handler handler = exceptionHandler;
+ if (handler != null)
+ handler.handle(cause);
+ }
+}
diff --git a/api/src/main/java/io/bosonnetwork/web/PaginatedResult.java b/api/src/main/java/io/bosonnetwork/web/PaginatedResult.java
new file mode 100644
index 00000000..96b71057
--- /dev/null
+++ b/api/src/main/java/io/bosonnetwork/web/PaginatedResult.java
@@ -0,0 +1,136 @@
+/*
+ * Copyright (c) 2023 - bosonnetwork.io
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.bosonnetwork.web;
+
+import java.util.Collections;
+import java.util.List;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Represents a paginated result set.
+ *
+ * @param the type of items in the result set
+ */
+public class PaginatedResult {
+ @JsonProperty("page")
+ private final long page;
+ @JsonProperty("pageSize")
+ private final long pageSize;
+ @JsonProperty("totalPages")
+ private final long totalPages;
+ @JsonProperty("totalItems")
+ private final long totalItems;
+ @JsonProperty("items")
+ private final List items;
+
+ /**
+ * Creates a new paginated result.
+ *
+ * @param page the current page number
+ * @param pageSize the number of items per page
+ * @param totalPages the total number of pages
+ * @param totalItems the total number of items
+ * @param items the items in the current page
+ */
+ @JsonCreator
+ protected PaginatedResult(@JsonProperty(value = "page", required = true) long page,
+ @JsonProperty(value = "pageSize", required = true) long pageSize,
+ @JsonProperty(value = "totalPages", required = true) long totalPages,
+ @JsonProperty(value = "totalItems", required = true) long totalItems,
+ @JsonProperty(value = "items") List items) {
+ this.page = page;
+ this.pageSize = pageSize;
+ this.totalPages = totalPages;
+ this.totalItems = totalItems;
+ this.items = items == null || items.isEmpty() ? Collections.emptyList() : Collections.unmodifiableList(items);
+ }
+
+ private PaginatedResult(long page, long pageSize, long totalItems, List items) {
+ this.page = page;
+ this.pageSize = pageSize;
+ this.totalPages = pageSize > 0 ? (totalItems + pageSize - 1) / pageSize : 0;
+ this.totalItems = totalItems;
+ this.items = items == null || items.isEmpty() ? Collections.emptyList() : Collections.unmodifiableList(items);
+ }
+
+ /**
+ * Creates a new paginated result.
+ *
+ * @param the type of items
+ * @param page the current page number
+ * @param pageSize the number of items per page
+ * @param totalItems the total number of items
+ * @param items the items in the current page
+ * @return a new paginated result
+ */
+ public static PaginatedResult of(long page, long pageSize, long totalItems, List items) {
+ return new PaginatedResult<>(page, pageSize, totalItems, items);
+ }
+
+ /**
+ * Returns the current page number.
+ *
+ * @return the page number
+ */
+ public long page() {
+ return page;
+ }
+
+ /**
+ * Returns the number of items per page.
+ *
+ * @return the page size
+ */
+ public long pageSize() {
+ return pageSize;
+ }
+
+ /**
+ * Returns the total number of pages.
+ *
+ * @return the total pages
+ */
+ public long totalPages() {
+ return totalPages;
+ }
+
+ /**
+ * Returns the total number of items.
+ *
+ * @return the total items
+ */
+ public long totalItems() {
+ return totalItems;
+ }
+
+ /**
+ * Returns the items in the current page.
+ *
+ * @return the items
+ */
+ public List items() {
+ return items;
+ }
+}
\ No newline at end of file
diff --git a/api/src/test/java/io/bosonnetwork/identifier/DHTRegistryTest.java b/api/src/test/java/io/bosonnetwork/identifier/DHTRegistryTest.java
index bc21d8d5..75a0c481 100644
--- a/api/src/test/java/io/bosonnetwork/identifier/DHTRegistryTest.java
+++ b/api/src/test/java/io/bosonnetwork/identifier/DHTRegistryTest.java
@@ -130,6 +130,11 @@ public CryptoContext createCryptoContext(Id id) {
return null;
}
+ @Override
+ public T unwrap(Class clazz) {
+ return null;
+ }
+
@Override
public CompletableFuture> findNode(Id id, LookupOption option) {
return null;
diff --git a/api/src/test/java/io/bosonnetwork/vertx/AsyncInputStreamTest.java b/api/src/test/java/io/bosonnetwork/vertx/AsyncInputStreamTest.java
new file mode 100644
index 00000000..81e58218
--- /dev/null
+++ b/api/src/test/java/io/bosonnetwork/vertx/AsyncInputStreamTest.java
@@ -0,0 +1,110 @@
+package io.bosonnetwork.vertx;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+
+@ExtendWith(VertxExtension.class)
+public class AsyncInputStreamTest {
+ // A ByteArrayInputStream that records whether it was closed.
+ private static class TrackingInputStream extends ByteArrayInputStream {
+ final AtomicBoolean closed = new AtomicBoolean();
+
+ TrackingInputStream(byte[] buf) {
+ super(buf);
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed.set(true);
+ super.close();
+ }
+ }
+
+ private static byte[] randomBytes(int n) {
+ byte[] b = new byte[n];
+ new Random(42).nextBytes(b);
+ return b;
+ }
+
+ @Test
+ void readsAllContentInOrderAndClosesInput(Vertx vertx, VertxTestContext tc) {
+ byte[] data = randomBytes(4096);
+ TrackingInputStream in = new TrackingInputStream(data);
+ Buffer acc = Buffer.buffer();
+
+ // chunkSize smaller than the payload to force multiple chunks
+ vertx.runOnContext(v -> {
+ AsyncInputStream stream = new AsyncInputStream(vertx, in, 256, true);
+ stream.exceptionHandler(tc::failNow);
+ stream.endHandler(x -> tc.verify(() -> {
+ assertArrayEquals(data, acc.getBytes());
+ assertTrue(in.closed.get(), "input should be closed on end");
+ tc.completeNow();
+ }));
+ stream.handler(acc::appendBuffer);
+ });
+ }
+
+ @Test
+ void doesNotCloseInputWhenConfigured(Vertx vertx, VertxTestContext tc) {
+ byte[] data = randomBytes(512);
+ TrackingInputStream in = new TrackingInputStream(data);
+
+ vertx.runOnContext(v -> {
+ AsyncInputStream stream = new AsyncInputStream(vertx, in, 256, false);
+ stream.exceptionHandler(tc::failNow);
+ stream.endHandler(x -> tc.verify(() -> {
+ assertFalse(in.closed.get(), "input should not be closed when closeInput=false");
+ tc.completeNow();
+ }));
+ stream.handler(b -> { });
+ });
+ }
+
+ @Test
+ void fetchHonorsDemand(Vertx vertx, VertxTestContext tc) {
+ byte[] data = randomBytes(1024); // 4 chunks at chunkSize 256
+ TrackingInputStream in = new TrackingInputStream(data);
+ AtomicInteger chunks = new AtomicInteger();
+ Buffer acc = Buffer.buffer();
+
+ vertx.runOnContext(v -> {
+ AsyncInputStream stream = new AsyncInputStream(vertx, in, 256, true);
+ stream.exceptionHandler(tc::failNow);
+ stream.endHandler(x -> tc.verify(() -> {
+ assertArrayEquals(data, acc.getBytes());
+ tc.completeNow();
+ }));
+ stream.pause();
+ stream.handler(b -> {
+ chunks.incrementAndGet();
+ acc.appendBuffer(b);
+ });
+ // request exactly one chunk
+ stream.fetch(1);
+
+ // after the single chunk has had time to arrive, only one should have been delivered
+ vertx.setTimer(300, t -> tc.verify(() -> {
+ assertEquals(1, chunks.get(), "fetch(1) must deliver exactly one chunk");
+ assertFalse(in.closed.get(), "stream must not have ended yet");
+ stream.resume(); // drain the remaining chunks and trigger end
+ }));
+ });
+ }
+}
\ No newline at end of file
diff --git a/api/src/test/java/io/bosonnetwork/vertx/AsyncOutputStreamTest.java b/api/src/test/java/io/bosonnetwork/vertx/AsyncOutputStreamTest.java
new file mode 100644
index 00000000..bef88030
--- /dev/null
+++ b/api/src/test/java/io/bosonnetwork/vertx/AsyncOutputStreamTest.java
@@ -0,0 +1,90 @@
+package io.bosonnetwork.vertx;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(VertxExtension.class)
+public class AsyncOutputStreamTest {
+ // A ByteArrayOutputStream that records whether it was closed.
+ private static class TrackingOutputStream extends ByteArrayOutputStream {
+ final AtomicBoolean closed = new AtomicBoolean();
+
+ @Override
+ public void close() {
+ closed.set(true);
+ }
+ }
+
+ private static byte[] randomBytes(int n) {
+ byte[] b = new byte[n];
+ new Random(42).nextBytes(b);
+ return b;
+ }
+
+ @Test
+ void writesAllDataInOrderAndCloses(Vertx vertx, VertxTestContext tc) {
+ TrackingOutputStream out = new TrackingOutputStream();
+
+ vertx.runOnContext(v -> {
+ AsyncOutputStream stream = new AsyncOutputStream(vertx, out, true);
+ stream.exceptionHandler(tc::failNow);
+ stream.write(Buffer.buffer("Hello, "))
+ .compose(x -> stream.write(Buffer.buffer("Ion ")))
+ .compose(x -> stream.write(Buffer.buffer("Store!")))
+ .compose(x -> stream.end())
+ .onComplete(tc.succeeding(x -> tc.verify(() -> {
+ assertEquals("Hello, Ion Store!", out.toString(StandardCharsets.UTF_8));
+ assertTrue(out.closed.get(), "output should be closed on end");
+ tc.completeNow();
+ })));
+ });
+ }
+
+ @Test
+ void doesNotCloseOutputWhenConfigured(Vertx vertx, VertxTestContext tc) {
+ TrackingOutputStream out = new TrackingOutputStream();
+
+ vertx.runOnContext(v -> {
+ AsyncOutputStream stream = new AsyncOutputStream(vertx, out, false);
+ stream.exceptionHandler(tc::failNow);
+ stream.write(Buffer.buffer("data"))
+ .compose(x -> stream.end())
+ .onComplete(tc.succeeding(x -> tc.verify(() -> {
+ assertEquals("data", out.toString(StandardCharsets.UTF_8));
+ assertFalse(out.closed.get(), "output should not be closed when closeOutput=false");
+ tc.completeNow();
+ })));
+ });
+ }
+
+ @Test
+ void pipeFromAsyncInputStreamRoundTrips(Vertx vertx, VertxTestContext tc) {
+ byte[] data = randomBytes(4096);
+ TrackingOutputStream out = new TrackingOutputStream();
+
+ vertx.runOnContext(v -> {
+ AsyncInputStream in = new AsyncInputStream(vertx, new ByteArrayInputStream(data), 256, true);
+ AsyncOutputStream sink = new AsyncOutputStream(vertx, out, true);
+ in.pipeTo(sink).onComplete(tc.succeeding(x -> tc.verify(() -> {
+ assertArrayEquals(data, out.toByteArray());
+ assertTrue(out.closed.get(), "output should be closed once the pipe completes");
+ tc.completeNow();
+ })));
+ });
+ }
+}
diff --git a/dht/src/main/java/io/bosonnetwork/kademlia/KadNode.java b/dht/src/main/java/io/bosonnetwork/kademlia/KadNode.java
index ef8255d1..ecf0f61f 100644
--- a/dht/src/main/java/io/bosonnetwork/kademlia/KadNode.java
+++ b/dht/src/main/java/io/bosonnetwork/kademlia/KadNode.java
@@ -838,6 +838,14 @@ public CryptoContext createCryptoContext(Id id) throws CryptoException {
return identity.createCryptoContext(id);
}
+ @Override
+ public T unwrap(Class clazz) {
+ if (clazz.isInstance(vertx))
+ return clazz.cast(vertx);
+
+ return null;
+ }
+
@Override
public String toString() {
return "Kademlia node: " + identity.getId().toString();