tommyskeff/futur4j
1
0
Fork 0
mirror of https://github.com/tommyskeff/futur4j.git synced 2026-01-18 07:16:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

respect executors that can't schedule

Browse Source
This commit is contained in:
WhatCats
2025-05-28 15:18:51 +02:00
parent fea0575392
commit 9137eed426
4 changed files with 47 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
futur-api/src/main/java/dev/tommyjs/futur/executor/ExecutorImpl.java
View File

@@ -1,6 +1,7 @@
package dev.tommyjs.futur.executor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import java.util.concurrent.Executor;
@@ -24,8 +25,8 @@ class ExecutorImpl implements PromiseExecutor<Void> {
}
@Override
public @NotNull PromiseScheduler<?> scheduler() {
return PromiseScheduler.getDefault();
public @Nullable PromiseScheduler<?> scheduler() {
return null;
}
}

5
futur-api/src/main/java/dev/tommyjs/futur/executor/ExecutorServiceImpl.java
View File

@@ -1,6 +1,7 @@
package dev.tommyjs.futur.executor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
@@ -24,8 +25,8 @@ class ExecutorServiceImpl implements PromiseExecutor<Future<?>> {
}
@Override
public @NotNull PromiseScheduler<?> scheduler() {
return PromiseScheduler.getDefault();
public @Nullable PromiseScheduler<?> scheduler() {
return null;
}
}

3
futur-api/src/main/java/dev/tommyjs/futur/executor/PromiseExecutor.java
View File

@@ -1,6 +1,7 @@
package dev.tommyjs.futur.executor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -77,6 +78,6 @@ public interface PromiseExecutor<T> {
*/
boolean cancel(@NotNull T task);
@NotNull PromiseScheduler<?> scheduler();
@Nullable PromiseScheduler<?> scheduler();
}

58
futur-api/src/main/java/dev/tommyjs/futur/promise/AbstractPromise.java
View File

@@ -215,9 +215,8 @@ public abstract class AbstractPromise<T> implements Promise<T> {
CompletablePromise<V> promise = createLinked();
addDirectListener(
res -> runCompleter(promise, () -> {
Runnable runnable = createCompleter(res, promise, task);
F future = executor.run(runnable);
promise.addDirectListener(_ -> executor.cancel(future));
Runnable completer = createCompleter(res, promise, task);
execute(promise, completer, executor);
}),
promise::completeExceptionally
);
@@ -227,14 +226,22 @@ public abstract class AbstractPromise<T> implements Promise<T> {
private <F, V> @NotNull Promise<V> thenApplyDelayed(
@NotNull ExceptionalFunction<T, V> task, long delay,
@NotNull TimeUnit unit, @NotNull PromiseScheduler<F> scheduler
@NotNull TimeUnit unit, @NotNull PromiseExecutor<F> executor
) {
CompletablePromise<V> promise = createLinked();
addDirectListener(
res -> runCompleter(promise, () -> {
Runnable runnable = createCompleter(res, promise, task);
F future = scheduler.schedule(runnable, delay, unit);
promise.addDirectListener(_ -> scheduler.cancel(future));
Runnable completer = createCompleter(res, promise, task);
PromiseScheduler<?> scheduler = executor.scheduler();
if (scheduler == null) {
schedule(
promise,
() -> runCompleter(promise, () -> execute(promise, completer, executor)),
delay, unit, PromiseScheduler.getDefault()
);
} else {
schedule(promise, completer, delay, unit, scheduler);
}
}),
promise::completeExceptionally
);
@@ -242,6 +249,19 @@ public abstract class AbstractPromise<T> implements Promise<T> {
return promise;
}
private <F> void execute(@NotNull Promise<?> promise, @NotNull Runnable task, @NotNull PromiseExecutor<F >executor) throws Exception {
F future = executor.run(task);
promise.addDirectListener(_ -> executor.cancel(future));
}
private <F> void schedule(
@NotNull Promise<?> promise, @NotNull Runnable task,
long delay, @NotNull TimeUnit unit, @NotNull PromiseScheduler<F> scheduler
) throws Exception {
F future = scheduler.schedule(task, delay, unit);
promise.addDirectListener(_ -> scheduler.cancel(future));
}
private <V> @NotNull Promise<V> thenCompose(
@NotNull ExceptionalFunction<T, Promise<V>> task,
@NotNull PromiseExecutor<?> executor
@@ -269,7 +289,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public @NotNull Promise<Void> thenRunDelayedSync(@NotNull ExceptionalRunnable task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getSyncExecutor().scheduler());
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getSyncExecutor());
}
@Override
@@ -279,7 +299,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public @NotNull Promise<Void> thenConsumeDelayedSync(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getSyncExecutor().scheduler());
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getSyncExecutor());
}
@Override
@@ -289,7 +309,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public <V> @NotNull Promise<V> thenSupplyDelayedSync(@NotNull ExceptionalSupplier<V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getSyncExecutor().scheduler());
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getSyncExecutor());
}
@Override
@@ -299,7 +319,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public <V> @NotNull Promise<V> thenApplyDelayedSync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(task, delay, unit, getFactory().getSyncExecutor().scheduler());
return thenApplyDelayed(task, delay, unit, getFactory().getSyncExecutor());
}
@Override
@@ -314,7 +334,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public @NotNull Promise<Void> thenRunDelayedAsync(@NotNull ExceptionalRunnable task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getAsyncExecutor().scheduler());
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getAsyncExecutor());
}
@Override
@@ -324,7 +344,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public @NotNull Promise<Void> thenConsumeDelayedAsync(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getAsyncExecutor().scheduler());
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getAsyncExecutor());
}
@Override
@@ -334,7 +354,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public <V> @NotNull Promise<V> thenSupplyDelayedAsync(@NotNull ExceptionalSupplier<V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getAsyncExecutor().scheduler());
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getAsyncExecutor());
}
@Override
@@ -344,7 +364,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public <V> @NotNull Promise<V> thenApplyDelayedAsync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(task, delay, unit, getFactory().getAsyncExecutor().scheduler());
return thenApplyDelayed(task, delay, unit, getFactory().getAsyncExecutor());
}
@Override
@@ -359,7 +379,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public @NotNull Promise<Void> thenRunDelayedVirtual(@NotNull ExceptionalRunnable task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getVirtualExecutor().scheduler());
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getVirtualExecutor());
}
@Override
@@ -369,7 +389,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public @NotNull Promise<Void> thenConsumeDelayedVirtual(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getVirtualExecutor().scheduler());
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getVirtualExecutor());
}
@Override
@@ -379,7 +399,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public <V> @NotNull Promise<V> thenSupplyDelayedVirtual(@NotNull ExceptionalSupplier<V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getVirtualExecutor().scheduler());
return thenApplyDelayed(FunctionAdapter.adapt(task), delay, unit, getFactory().getVirtualExecutor());
}
@Override
@@ -389,7 +409,7 @@ public abstract class AbstractPromise<T> implements Promise<T> {
@Override
public <V> @NotNull Promise<V> thenApplyDelayedVirtual(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayed(task, delay, unit, getFactory().getVirtualExecutor().scheduler());
return thenApplyDelayed(task, delay, unit, getFactory().getVirtualExecutor());
}
@Override