Kotlin promise coroutines:

commit 973112c
Author: Tiago de Freitas Lima <[email protected]>
Date:   Mon May 23 15:15:56 2022 -0300

    DefaultPromise starts to take a Executor in order to run CompletableFuture operations; KtorHTTPClient relies on CoroutineDispatcher from client

commit 6b024b9
Author: Tiago de Freitas Lima <[email protected]>
Date:   Mon May 23 10:17:44 2022 -0300

    (wip)

Author: ljtfreitas

core/src/main/java/com/github/ljtfreitas/julian/DefaultPromise.java

@@ -25,6 +25,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -34,19 +35,25 @@
 class DefaultPromise<T> implements Promise<T> {
 
 	private final CompletableFuture<T> future;
+	private final Executor executor;
 
 	DefaultPromise(CompletableFuture<T> future) {
+		this(future, future.defaultExecutor());
+	}
+
+	DefaultPromise(CompletableFuture<T> future, Executor executor) {
 		this.future = future;
+		this.executor = executor;
 	}
 
 	@Override
 	public Promise<T> onSuccess(Consumer<? super T> fn) {
-		return new DefaultPromise<>(future.whenCompleteAsync((r, e) -> { if (e == null) fn.accept(r); }));
+		return new DefaultPromise<>(future.whenCompleteAsync((r, e) -> { if (e == null) fn.accept(r); }, executor), executor);
 	}
 
 	@Override
 	public <R> Promise<R> then(Function<? super T, R> fn) {
-		return new DefaultPromise<>(future.thenApplyAsync(fn));
+		return new DefaultPromise<>(future.thenApplyAsync(fn, executor), executor);
 	}
 
 	@Override
@@ -56,23 +63,23 @@ public <R> R fold(Function<? super T, R> success, Function<? super Exception, R>
 				return failure.apply((Exception) e);
 			else
 				return success.apply(r);
-		})).join().unsafe();
+		}, executor), executor).join().unsafe();
 
 	}
 
 	@Override
 	public <R> Promise<R> bind(Function<? super T, Promise<R>> fn) {
-		return new DefaultPromise<>(future.thenComposeAsync(t -> fn.apply(t).future()));
+		return new DefaultPromise<>(future.thenComposeAsync(t -> fn.apply(t).future(), executor), executor);
 	}
 
 	@Override
 	public <T2, R> Promise<R> zip(Promise<T2> other, BiFunction<? super T, ? super T2, R> fn) {
-		return new DefaultPromise<>(future.thenCombineAsync(other.future(), fn));
+		return new DefaultPromise<>(future.thenCombineAsync(other.future(), fn, executor), executor);
 	}
 
 	@Override
 	public Promise<T> recover(Function<? super Exception, T> fn) {
-		return new DefaultPromise<>(future.exceptionally(t -> fn.apply((Exception) t)));
+		return new DefaultPromise<>(future.exceptionally(t -> fn.apply((Exception) t)), executor);
 	}
 
 	@Override
@@ -83,7 +90,7 @@ public <Err extends Exception> Promise<T> recover(Class<? extends Err> expected,
 				return fn.apply(expected.cast(cause));
 			else
 				return r;
-		}));
+		}, executor), executor);
 	}
 
 	@Override
@@ -94,7 +101,7 @@ public Promise<T> recover(Predicate<? super Exception> p, Function<? super Excep
 				return fn.apply(cause);
 			else
 				return r;
-		}));
+		}, executor), executor);
 	}
 
 	@Override
@@ -105,7 +112,7 @@ public <Err extends Exception> Promise<T> failure(Function<? super Exception, Er
 				throw failure(fn.apply(cause));
 			else
 				return r;
-		}));
+		}, executor), executor);
 	}
 
 	@Override
@@ -114,7 +121,7 @@ public Promise<T> onFailure(Consumer<? super Exception> fn) {
 			Exception cause = deep(e);
 			if (cause != null)
 				fn.accept(cause);
-		}));
+		}, executor), executor);
 	}
 
 	private RuntimeException failure(Exception e) {
@@ -141,6 +148,6 @@ public CompletableFuture<T> future() {
 	public Promise<T> subscribe(Subscriber<? super T> subscriber) {
 		BiConsumer<T, Throwable> handle = (r, e) -> { if (e == null) subscriber.success(r); else subscriber.failure((Exception) e); };
 		BiConsumer<T, Throwable> done = (r, e) -> subscriber.done();
-		return new DefaultPromise<>(future.whenCompleteAsync(handle).whenCompleteAsync(done));
+		return new DefaultPromise<>(future.whenCompleteAsync(handle, executor).whenCompleteAsync(done, executor), executor);
 	}
 }

core/src/main/java/com/github/ljtfreitas/julian/Promise.java

@@ -83,6 +83,10 @@ static <T> Promise<T> pending(CompletableFuture<T> future) {
 		return new DefaultPromise<>(future);
 	}
 
+	static <T> Promise<T> pending(CompletableFuture<T> future, Executor executor) {
+		return new DefaultPromise<>(future, executor);
+	}
+
 	static <T> Promise<T> pending(Supplier<T> fn) {
 		return new DefaultPromise<>(supplyAsync(fn));
 	}

http-client-ktor/src/main/kotlin/com/github/ljtfreitas/julian/k/http/client/ktor/KtorHTTPClient.kt

@@ -34,7 +34,6 @@ import com.github.ljtfreitas.julian.http.MediaType
 import com.github.ljtfreitas.julian.http.client.HTTPClient
 import com.github.ljtfreitas.julian.http.client.HTTPClientRequest
 import com.github.ljtfreitas.julian.http.client.HTTPClientResponse
-import com.github.ljtfreitas.julian.k.http.client.ktor.KtorHTTPClient.Companion.invoke
 import io.ktor.client.HttpClient
 import io.ktor.client.HttpClientConfig
 import io.ktor.client.call.receive
@@ -51,12 +50,17 @@ import io.ktor.http.HttpMethod
 import io.ktor.http.content.OutgoingContent
 import io.ktor.util.toMap
 import io.ktor.utils.io.ByteWriteChannel
+import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.asExecutor
 import kotlinx.coroutines.future.future
 import kotlinx.coroutines.jdk9.asFlow
 import java.io.Closeable
 import java.util.Optional
+import java.util.concurrent.Executor
+import kotlin.coroutines.ContinuationInterceptor
+import kotlin.coroutines.CoroutineContext
 
 @DelicateCoroutinesApi
 class KtorHTTPClient private constructor(private val client: HttpClient): HTTPClient, Closeable {
@@ -67,7 +71,7 @@ class KtorHTTPClient private constructor(private val client: HttpClient): HTTPCl
 
         operator fun invoke(
             block: HttpClientConfig<*>.() -> Unit = {}
-        ) = KtorHTTPClient(HttpClient() {
+        ) = KtorHTTPClient(HttpClient {
             block()
             expectSuccess = false
         })
@@ -90,7 +94,7 @@ class KtorHTTPClient private constructor(private val client: HttpClient): HTTPCl
     }
 
     override fun request(request: HTTPRequestDefinition) = HTTPClientRequest {
-        Promise.pending(GlobalScope.future {
+        Promise.pending(GlobalScope.future(client.coroutineContext) {
             val response: HttpResponse = client.request(request.path().toURL()) {
                 method = HttpMethod.parse(request.method().name)
 
@@ -104,7 +108,8 @@ class KtorHTTPClient private constructor(private val client: HttpClient): HTTPCl
             }
 
             return@future response.asHTTPClientResponse()
-        })
+
+        }, client.coroutineContext.javaExecutor())
     }
 
     private fun HTTPHeaders.contentType() = select(HTTPHeader.CONTENT_TYPE).map { MediaType.valueOf(it.value()) }
@@ -135,4 +140,7 @@ class KtorHTTPClient private constructor(private val client: HttpClient): HTTPCl
     }
 
     override fun close() = client.close()
-}
+
+    private fun CoroutineContext.javaExecutor(): Executor = (get(ContinuationInterceptor) as CoroutineDispatcher).asExecutor()
+}
+