Refactoring customizers for contextual promises & blocking IO; fixing NIO classes

Author: vsilaev

.settings/org.eclipse.jdt.core.prefs

@@ -3,11 +3,14 @@ org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
 org.eclipse.jdt.core.compiler.codegen.methodParameters=do not generate
 org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
 org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve
-org.eclipse.jdt.core.compiler.compliance=9
+org.eclipse.jdt.core.compiler.compliance=1.8
 org.eclipse.jdt.core.compiler.debug.lineNumber=generate
 org.eclipse.jdt.core.compiler.debug.localVariable=generate
 org.eclipse.jdt.core.compiler.debug.sourceFile=generate
 org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
+org.eclipse.jdt.core.compiler.problem.enablePreviewFeatures=disabled
 org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
 org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
+org.eclipse.jdt.core.compiler.problem.reportPreviewFeatures=ignore
+org.eclipse.jdt.core.compiler.release=disabled
 org.eclipse.jdt.core.compiler.source=1.8

README.md

@@ -1,4 +1,4 @@
-[![Maven Central](https://img.shields.io/maven-central/v/net.tascalate/net.tascalate.concurrent.svg)](https://search.maven.org/artifact/net.tascalate/net.tascalate.concurrent/0.9.5/jar) [![GitHub release](https://img.shields.io/github/release/vsilaev/tascalate-concurrent.svg)](https://github.com/vsilaev/tascalate-concurrent/releases/tag/0.9.5) [![license](https://img.shields.io/github/license/vsilaev/tascalate-concurrent.svg)](http://www.apache.org/licenses/LICENSE-2.0.txt)
+[![Maven Central](https://img.shields.io/maven-central/v/net.tascalate/net.tascalate.concurrent.svg)](https://search.maven.org/artifact/net.tascalate/net.tascalate.concurrent/0.9.6/jar) [![GitHub release](https://img.shields.io/github/release/vsilaev/tascalate-concurrent.svg)](https://github.com/vsilaev/tascalate-concurrent/releases/tag/0.9.6) [![license](https://img.shields.io/github/license/vsilaev/tascalate-concurrent.svg)](http://www.apache.org/licenses/LICENSE-2.0.txt)
 # tascalate-concurrent
 The library provides an implementation of the [CompletionStage](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletionStage.html) interface and related classes these are designed to support long-running blocking tasks (typically, I/O bound). This functionality augments the sole Java 8 built-in implementation, [CompletableFuture](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html), that is primarily supports computational tasks. Also, the library helps with numerous asynchronous programing challenges like handling timeouts, retry/poll functionality, orchestrating results of multiple concurrent computations and similar.
 
@@ -12,7 +12,7 @@ New name:
 <dependency>
     <groupId>net.tascalate</groupId>
     <artifactId>net.tascalate.concurrent</artifactId>
-    <version>0.9.5</version> <!-- Any version above 0.8.0, the latest one is recommended -->
+    <version>0.9.6</version> <!-- Any version above 0.8.0, the latest one is recommended -->
 </dependency>
 ```
 Old Name
@@ -38,7 +38,7 @@ To use a library you have to add a single Maven dependency
 <dependency>
     <groupId>net.tascalate</groupId>
     <artifactId>net.tascalate.concurrent</artifactId>
-    <version>0.9.5</version>
+    <version>0.9.6</version>
 </dependency>
 ```
 # What is inside?

pom.xml

@@ -6,7 +6,7 @@
 
 	<groupId>net.tascalate</groupId>
 	<artifactId>net.tascalate.concurrent</artifactId>
-	<version>0.9.5</version>
+	<version>0.9.6</version>
 	<packaging>jar</packaging>  
 
 	<name>Tascalate Concurrent</name>

src/main/java/net/tascalate/concurrent/decorators/CompletableFutureDecorator.java

@@ -21,7 +21,6 @@
 import java.util.concurrent.CompletionStage;
 
 import net.tascalate.concurrent.Promise;
-import net.tascalate.concurrent.decorators.AbstractFutureDecorator;
 
 public class CompletableFutureDecorator<T> 
     extends AbstractFutureDecorator<T, CompletableFuture<T>> 

src/main/java/net/tascalate/concurrent/io/AbstractAsyncFileChannel.java

@@ -62,22 +62,25 @@ public void force(boolean metaData) throws IOException {
     }
 
     public Promise<FileLock> lock(boolean shared) {
-        return lock(0, Long.MAX_VALUE, shared);
+        return doLock(0, Long.MAX_VALUE, shared);
     }
 
     @Override
     public Promise<FileLock> lock(long position, long size, boolean shared) {
-        Promise<FileLock> promise = (Promise<FileLock>)delegate.lock(position, size, shared);
-        return promise.dependent()
-                      .thenApply(this::upgradeLock, true)
-                      .unwrap();
+        return doLock(position, size, shared);
     }
 
     public final <A> void lock(boolean shared, A attachment,
                                CompletionHandler<FileLock, ? super A> handler) {
         lock(0, Long.MAX_VALUE, shared, attachment, handler);
     }
 
+    protected Promise<FileLock> doLock(long position, long size, boolean shared) {
+        AsyncResult<FileLock> asyncResult = new AsyncResult<>();
+        lock(position, size, shared, null, asyncResult.handler);
+        return asyncResult;
+    }
+    
     @Override
     public <A> void lock(long position, long size, boolean shared, A attachment,
                          CompletionHandler<FileLock, ? super A> handler) {
@@ -114,7 +117,9 @@ public <A> void read(ByteBuffer dst, long position, A attachment, CompletionHand
 
     @Override
     public Promise<Integer> read(ByteBuffer dst, long position) {
-        return (Promise<Integer>)delegate.read(dst, position);
+        AsyncResult<Integer> asyncResult = new AsyncResult<>();
+        read(dst, position, null, asyncResult.handler);
+        return asyncResult;
     }
 
     @Override
@@ -124,7 +129,9 @@ public <A> void write(ByteBuffer src, long position, A attachment, CompletionHan
 
     @Override
     public Promise<Integer> write(ByteBuffer src, long position) {
-        return (Promise<Integer>)delegate.write(src, position);
+        AsyncResult<Integer> asyncResult = new AsyncResult<>();
+        write(src, position, null, asyncResult.handler);
+        return asyncResult;
     }
 
     protected FileLock upgradeLock(FileLock delegate) {

src/main/java/net/tascalate/concurrent/io/AbstractAsyncServerSocketChannel.java

@@ -88,10 +88,9 @@ public Promise<AsynchronousSocketChannel> accept() {
     */
 
     public Promise<C> accept() {
-        Promise<AsynchronousSocketChannel> promise = (Promise<AsynchronousSocketChannel>)delegate.accept();
-        return promise.dependent()
-                      .thenApply(this::wrap, true)
-                      .unwrap();
+        AsyncResult<C> asyncResult = new AsyncResult<>();
+        accept(null, asyncResult.handler);
+        return asyncResult;
     }
 
     /*

src/main/java/net/tascalate/concurrent/io/AbstractAsyncSocketChannel.java

@@ -93,12 +93,16 @@ public <A> void connect(SocketAddress remote, A attachment, CompletionHandler<Vo
 
     @Override
     public Promise<Void> connect(SocketAddress remote) {
-        return (Promise<Void>)delegate.connect(remote);
+        AsyncResult<Void> asyncResult = new AsyncResult<>();
+        connect(remote, null, asyncResult.handler);
+        return asyncResult;
     }
 
     @Override
     public Promise<Integer> read(ByteBuffer dst) {
-        return (Promise<Integer>)delegate.read(dst);
+        AsyncResult<Integer> asyncResult = new AsyncResult<>();
+        read(dst, null, asyncResult.handler);
+        return asyncResult;
     }
 
     @Override
@@ -115,7 +119,9 @@ public <A> void read(ByteBuffer[] dsts, int offset, int length, long timeout, Ti
 
     @Override
     public Promise<Integer> write(ByteBuffer src) {
-        return (Promise<Integer>)delegate.write(src);
+        AsyncResult<Integer> asyncResult = new AsyncResult<>();
+        write(src, null, asyncResult.handler);
+        return asyncResult;
     }
 
     @Override

src/main/java/net/tascalate/concurrent/io/AsyncChannelGroup.java

@@ -24,16 +24,15 @@
 import java.util.HashSet;
 import java.util.Objects;
 import java.util.Set;
-
-import net.tascalate.concurrent.TaskExecutorService;
+import java.util.concurrent.ExecutorService;
 
 public class AsyncFileChannel extends AbstractAsyncFileChannel<AsyncFileChannel> {
 
     protected AsyncFileChannel(AsynchronousFileChannel delegate) {
         super(delegate);
     }
 
-    public static AsyncFileChannel open(Path file, TaskExecutorService executor, OpenOption... options) throws IOException {
+    public static AsyncFileChannel open(Path file, ExecutorService executor, OpenOption... options) throws IOException {
         Set<OpenOption> set;
         if (options.length == 0) {
             set = Collections.emptySet();
@@ -45,7 +44,7 @@ public static AsyncFileChannel open(Path file, TaskExecutorService executor, Ope
     }
 
     public static AsyncFileChannel open(Path file,
-                                        TaskExecutorService executor,
+                                        ExecutorService executor,
                                         Set<? extends OpenOption> options,
                                         FileAttribute<?>... attrs) throws IOException {
         Objects.requireNonNull(executor, "Executor should be specified");

src/main/java/net/tascalate/concurrent/io/AsyncFileChannel.java

@@ -0,0 +1,39 @@
+/**
+ * Copyright 2015-2020 Valery Silaev (http://vsilaev.com)
+ *
+ * Licensed 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
+ *
+ *     http://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 net.tascalate.concurrent.io;
+
+import java.nio.channels.CompletionHandler;
+
+import net.tascalate.concurrent.CompletablePromise;
+
+class AsyncResult<V> extends CompletablePromise<V> {
+    
+    final CompletionHandler<V, Object> handler = new CompletionHandler<V, Object>() {
+        @Override
+        public void completed(final V result, final Object attachment) {
+            success(result);
+        }
+
+        @Override
+        public void failed(Throwable exc, Object attachment) {
+            failure(exc);     
+        }
+    };
+
+    AsyncResult() {
+    }
+    
+}