@Test public void testRead() throws IOException { Path srcPath = Paths.get("test_data/test.txt"); FileChannel channel = FileChannel.open(srcPath, set(READ)); byte[] result = new byte[20]; await(Promises.all(IntStream.range(0, 100) .mapToObj(i -> service.read(channel, 0, result, 0, result.length) .whenComplete((res, e) -> { if (e != null) { e.printStackTrace(); fail(); } try { assertEquals(Files.readAllBytes(srcPath).length, res.intValue()); } catch (IOException e1) { e1.printStackTrace(); } })).collect(Collectors.toList()))); }
@Override public Promise<Integer> read(FileChannel channel, long position, byte[] array, int offset, int size) { return ofBlockingCallable(executor, () -> { ByteBuffer buffer = ByteBuffer.wrap(array, offset, size); long pos = position; do { try { int readBytes = channel.read(buffer, pos); if (readBytes == -1) { break; } pos += readBytes; } catch (IOException e) { throw new UncheckedException(e); } } while (buffer.position() < buffer.limit()); return Math.toIntExact(pos - position); }); }
ExecutorAsyncFileService implements AsyncFileService { @Override public Promise<Integer> read(FileChannel channel, long position, byte[] array, int offset, int size) { return ofBlockingCallable(executor, () -> { ByteBuffer buffer = ByteBuffer.wrap(array, offset, size); long pos = position; do { try { int readBytes = channel.read(buffer, pos); if (readBytes == -1) { break; } pos += readBytes; } catch (IOException e) { throw new UncheckedException(e); } } while (buffer.position() < buffer.limit()); return Math.toIntExact(pos - position); }); } }
ExecutorAsyncFileService implements AsyncFileService { @Override public Promise<Integer> read(FileChannel channel, long position, byte[] array, int offset, int size) { return ofBlockingCallable(executor, () -> { ByteBuffer buffer = ByteBuffer.wrap(array, offset, size); long pos = position; do { try { int readBytes = channel.read(buffer, pos); if (readBytes == -1) { break; } pos += readBytes; } catch (IOException e) { throw new UncheckedException(e); } } while (buffer.position() < buffer.limit()); return Math.toIntExact(pos - position); }); } ExecutorAsyncFileService(@NotNull Executor executor); }
ExecutorAsyncFileService implements AsyncFileService { @Override public Promise<Integer> read(FileChannel channel, long position, byte[] array, int offset, int size) { return ofBlockingCallable(executor, () -> { ByteBuffer buffer = ByteBuffer.wrap(array, offset, size); long pos = position; do { try { int readBytes = channel.read(buffer, pos); if (readBytes == -1) { break; } pos += readBytes; } catch (IOException e) { throw new UncheckedException(e); } } while (buffer.position() < buffer.limit()); return Math.toIntExact(pos - position); }); } ExecutorAsyncFileService(@NotNull Executor executor); @Override Promise<Integer> read(FileChannel channel, long position, byte[] array, int offset, int size); @Override Promise<Integer> write(FileChannel channel, long position, byte[] array, int offset, int size); }
ExecutorAsyncFileService implements AsyncFileService { @Override public Promise<Integer> read(FileChannel channel, long position, byte[] array, int offset, int size) { return ofBlockingCallable(executor, () -> { ByteBuffer buffer = ByteBuffer.wrap(array, offset, size); long pos = position; do { try { int readBytes = channel.read(buffer, pos); if (readBytes == -1) { break; } pos += readBytes; } catch (IOException e) { throw new UncheckedException(e); } } while (buffer.position() < buffer.limit()); return Math.toIntExact(pos - position); }); } ExecutorAsyncFileService(@NotNull Executor executor); @Override Promise<Integer> read(FileChannel channel, long position, byte[] array, int offset, int size); @Override Promise<Integer> write(FileChannel channel, long position, byte[] array, int offset, int size);  }