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

promise util with propagate cancellation options

Browse Source
This commit is contained in:
WhatCats
2024-04-06 13:55:50 +02:00
parent 5bbcfdc9b3
commit 8ba023c04a
10 changed files with 330 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
futur-api/src/main/java/dev/tommyjs/futur/executor/DualPoolExecutor.java
View File

@@ -3,11 +3,11 @@ package dev.tommyjs.futur.executor;
import org.jetbrains.annotations.NotNull;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
public class DualPoolExecutor implements PromiseExecutor<ScheduledFuture<?>> {
public class DualPoolExecutor implements PromiseExecutor<Future<?>> {
private final @NotNull ScheduledExecutorService syncSvc;
private final @NotNull ScheduledExecutorService asyncSvc;
@@ -22,17 +22,17 @@ public class DualPoolExecutor implements PromiseExecutor<ScheduledFuture<?>> {
}
@Override
public ScheduledFuture<?> runSync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit) {
public Future<?> runSync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit) {
return syncSvc.schedule(task, delay, unit);
}
@Override
public ScheduledFuture<?> runAsync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit) {
public Future<?> runAsync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit) {
return asyncSvc.schedule(task, delay, unit);
}
@Override
public void cancel(ScheduledFuture<?> task) {
public void cancel(Future<?> task) {
task.cancel(true);
}

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

@@ -27,6 +27,27 @@ public abstract class AbstractPromise<T, F> implements Promise<T> {
this.completion = new AtomicReference<>();
}
protected static <V> void propagateResult(Promise<V> from, Promise<V> to) {
from.addListener(to::complete, to::completeExceptionally);
}
protected static void propagateCancel(Promise<?> from, Promise<?> to) {
from.onCancel(to::completeExceptionally);
}
private <V> @NotNull Runnable createRunnable(T result, @NotNull Promise<V> promise, @NotNull ExceptionalFunction<T, V> task) {
return () -> {
if (promise.isCompleted()) return;
try {
V nextResult = task.apply(result);
promise.complete(nextResult);
} catch (Throwable e) {
promise.completeExceptionally(e);
}
};
}
public abstract @NotNull AbstractPromiseFactory<F> getFactory();
protected @NotNull PromiseExecutor<F> getExecutor() {
@@ -124,7 +145,7 @@ public abstract class AbstractPromise<T, F> implements Promise<T> {
promise::completeExceptionally
);
addChild(promise);
propagateCancel(promise, this);
return promise;
}
@@ -140,7 +161,7 @@ public abstract class AbstractPromise<T, F> implements Promise<T> {
promise::completeExceptionally
);
addChild(promise);
propagateCancel(promise, this);
return promise;
}
@@ -149,13 +170,13 @@ public abstract class AbstractPromise<T, F> implements Promise<T> {
Promise<V> promise = getFactory().unresolved();
thenApplySync(task).addListener(
nestedPromise -> {
nestedPromise.propagateResult(promise);
nestedPromise.addChild(promise);
propagateResult(nestedPromise, promise);
propagateCancel(promise, nestedPromise);
},
promise::completeExceptionally
);
addChild(promise);
propagateCancel(promise, this);
return promise;
}
@@ -221,7 +242,7 @@ public abstract class AbstractPromise<T, F> implements Promise<T> {
promise::completeExceptionally
);
addChild(promise);
propagateCancel(promise, this);
return promise;
}
@@ -237,7 +258,7 @@ public abstract class AbstractPromise<T, F> implements Promise<T> {
promise::completeExceptionally
);
addChild(promise);
propagateCancel(promise, this);
return promise;
}
@@ -246,29 +267,22 @@ public abstract class AbstractPromise<T, F> implements Promise<T> {
Promise<V> promise = getFactory().unresolved();
thenApplyAsync(task).addListener(
nestedPromise -> {
nestedPromise.propagateResult(promise);
nestedPromise.addChild(promise);
propagateResult(nestedPromise, promise);
propagateCancel(promise, nestedPromise);
},
promise::completeExceptionally
);
addChild(promise);
propagateCancel(promise, this);
return promise;
}
private <V> @NotNull Runnable createRunnable(T result, @NotNull Promise<V> promise, @NotNull ExceptionalFunction<T, V> task) {
return () -> {
if (promise.isCompleted()) return;
try {
V nextResult = task.apply(result);
promise.complete(nextResult);
} catch (Throwable e) {
promise.completeExceptionally(e);
}
};
@Override
public @NotNull Promise<Void> erase() {
return thenSupplyAsync(() -> null);
}
@Override
public @NotNull Promise<T> logExceptions(@NotNull String message) {
return onError(e -> getLogger().error(message, e));
@@ -365,16 +379,6 @@ public abstract class AbstractPromise<T, F> implements Promise<T> {
return this.completion.compareAndSet(null, completion);
}
@Override
public void addChild(@NotNull Promise<?> child) {
child.onCancel((e) -> this.cancel(e.getMessage()));
}
@Override
public void propagateResult(@NotNull Promise<T> target) {
addListener(target::complete, target::completeExceptionally);
}
@Override
public void cancel() {
completeExceptionally(new CancellationException());

72
futur-api/src/main/java/dev/tommyjs/futur/promise/AbstractPromiseFactory.java
View File

@@ -3,14 +3,12 @@ package dev.tommyjs.futur.promise;
import dev.tommyjs.futur.executor.PromiseExecutor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
@@ -19,9 +17,9 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
public abstract @NotNull PromiseExecutor<F> getExecutor();
@Override
public <K, V> @NotNull Promise<Map.Entry<K, V>> combine(@NotNull Promise<K> p1, @NotNull Promise<V> p2) {
public <K, V> @NotNull Promise<Map.Entry<K, V>> combine(boolean propagateCancel, @NotNull Promise<K> p1, @NotNull Promise<V> p2) {
List<Promise<?>> promises = List.of(p1, p2);
return all(promises)
return all(propagateCancel, promises)
.thenApplyAsync((res) -> new AbstractMap.SimpleImmutableEntry<>(
Objects.requireNonNull(p1.getCompletion()).getResult(),
Objects.requireNonNull(p2.getCompletion()).getResult()
@@ -29,12 +27,16 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
}
@Override
public <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, @Nullable BiConsumer<K, Throwable> exceptionHandler) {
public <K, V> @NotNull Promise<Map<K, V>> combine(boolean propagateCancel, @NotNull Map<K, Promise<V>> promises, @Nullable BiConsumer<K, Throwable> exceptionHandler) {
if (promises.isEmpty()) return resolve(Collections.emptyMap());
Map<K, V> map = new HashMap<>();
Promise<Map<K, V>> promise = unresolved();
for (Map.Entry<K, Promise<V>> entry : promises.entrySet()) {
if (propagateCancel) {
AbstractPromise.propagateCancel(promise, entry.getValue());
}
entry.getValue().addListener((ctx) -> {
synchronized (map) {
if (ctx.getException() != null) {
@@ -59,12 +61,13 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
}
@Override
public <V> @NotNull Promise<List<V>> combine(@NotNull Iterable<Promise<V>> promises, @Nullable Consumer<Throwable> exceptionHandler) {
public <V> @NotNull Promise<List<V>> combine(boolean propagateCancel, @NotNull Iterable<Promise<V>> promises, @Nullable BiConsumer<Integer, Throwable> exceptionHandler) {
AtomicInteger index = new AtomicInteger();
return this.combine(
propagateCancel,
StreamSupport.stream(promises.spliterator(), false)
.collect(Collectors.toMap(k -> index.getAndIncrement(), v -> v)),
exceptionHandler != null ? (i, e) -> exceptionHandler.accept(e) : null
exceptionHandler
).thenApplyAsync(v ->
v.entrySet().stream()
.sorted(Map.Entry.comparingByKey())
@@ -74,12 +77,7 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
}
@Override
public <V> @NotNull Promise<List<V>> combine(@NotNull Iterable<Promise<V>> promises) {
return combine(promises, null);
}
@Override
public @NotNull Promise<List<PromiseCompletion<?>>> allSettled(@NotNull Iterable<Promise<?>> promiseIterable) {
public @NotNull Promise<List<PromiseCompletion<?>>> allSettled(boolean propagateCancel, @NotNull Iterable<Promise<?>> promiseIterable) {
List<Promise<?>> promises = new ArrayList<>();
promiseIterable.iterator().forEachRemaining(promises::add);
@@ -91,7 +89,13 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
while (iter.hasNext()) {
int index = iter.nextIndex();
iter.next().addListener((res) -> {
var p = iter.next();
if (propagateCancel) {
AbstractPromise.propagateCancel(promise, p);
}
p.addListener((res) -> {
synchronized (results) {
results[index] = res;
if (Arrays.stream(results).allMatch(Objects::nonNull))
@@ -104,7 +108,7 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
}
@Override
public @NotNull Promise<Void> all(@NotNull Iterable<Promise<?>> promiseIterable) {
public @NotNull Promise<Void> all(boolean propagateCancel, @NotNull Iterable<Promise<?>> promiseIterable) {
List<Promise<?>> promises = new ArrayList<>();
promiseIterable.iterator().forEachRemaining(promises::add);
@@ -113,12 +117,14 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
Promise<Void> promise = unresolved();
for (Promise<?> p : promises) {
if (propagateCancel) {
AbstractPromise.propagateCancel(promise, p);
}
p.addListener((res) -> {
if (res.getException() != null) {
promise.completeExceptionally(res.getException());
}
if (completed.incrementAndGet() == promises.size()) {
} else if (completed.incrementAndGet() == promises.size()) {
promise.complete(null);
}
});
@@ -127,6 +133,18 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
return promise;
}
@Override
public <V> @NotNull Promise<V> race(boolean cancelRaceLosers, @NotNull Iterable<Promise<V>> promises) {
Promise<V> promise = unresolved();
for (Promise<V> p : promises) {
if (cancelRaceLosers) {
promise.addListener((res) -> p.cancel());
}
AbstractPromise.propagateResult(p, promise);
}
return promise;
}
@Override
public <T> @NotNull Promise<T> wrap(@NotNull CompletableFuture<T> future) {
Promise<T> promise = unresolved();
@@ -145,10 +163,7 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
@Override
public <T> @NotNull Promise<T> wrap(@NotNull Mono<T> mono) {
Promise<T> promise = this.unresolved();
Disposable disposable = mono.subscribe(promise::complete, promise::completeExceptionally);
promise.onCancel((e) -> disposable.dispose());
return promise;
return wrap(mono.toFuture());
}
@Override
@@ -165,17 +180,4 @@ public abstract class AbstractPromiseFactory<F> implements PromiseFactory {
return promise;
}
@Override
public @NotNull Promise<Void> erase(@NotNull Promise<?> p) {
Promise<Void> promise = unresolved();
p.addListener(ctx -> {
if (ctx.getException() != null) {
promise.completeExceptionally(ctx.getException());
} else {
promise.complete(null);
}
});
return promise;
}
}

18
futur-api/src/main/java/dev/tommyjs/futur/promise/Promise.java
View File

@@ -55,12 +55,14 @@ public interface Promise<T> {
<V> @NotNull Promise<V> thenComposeAsync(@NotNull ExceptionalFunction<T, Promise<V>> task);
@NotNull Promise<T> logExceptions(@NotNull String message);
@NotNull Promise<Void> erase();
default @NotNull Promise<T> logExceptions() {
return logExceptions("Exception caught in promise chain");
}
@NotNull Promise<T> logExceptions(@NotNull String message);
@NotNull Promise<T> addListener(@NotNull PromiseListener<T> listener);
@NotNull Promise<T> addListener(@Nullable Consumer<T> successHandler, @Nullable Consumer<Throwable> errorHandler);
@@ -73,30 +75,26 @@ public interface Promise<T> {
@NotNull Promise<T> onCancel(@NotNull Consumer<CancellationException> listener);
@Deprecated
@NotNull Promise<T> timeout(long time, @NotNull TimeUnit unit);
@Deprecated
default @NotNull Promise<T> timeout(long ms) {
return timeout(ms, TimeUnit.MILLISECONDS);
}
@NotNull Promise<T> maxWaitTime(long time, @NotNull TimeUnit unit);
@Deprecated
@NotNull Promise<T> timeout(long time, @NotNull TimeUnit unit);
default @NotNull Promise<T> maxWaitTime(long ms) {
return maxWaitTime(ms, TimeUnit.MILLISECONDS);
}
void addChild(@NotNull Promise<?> child);
void propagateResult(@NotNull Promise<T> target);
void cancel(@Nullable String reason);
@NotNull Promise<T> maxWaitTime(long time, @NotNull TimeUnit unit);
default void cancel() {
cancel(null);
}
void cancel(@Nullable String reason);
void complete(@Nullable T result);
void completeExceptionally(@NotNull Throwable result);

65
futur-api/src/main/java/dev/tommyjs/futur/promise/PromiseFactory.java
View File

@@ -10,7 +10,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
public interface PromiseFactory {
@@ -18,44 +17,84 @@ public interface PromiseFactory {
<T> @NotNull Promise<T> unresolved();
<K, V> @NotNull Promise<Map.Entry<K, V>> combine(@NotNull Promise<K> p1, @NotNull Promise<V> p2);
default <K, V> @NotNull Promise<Map.Entry<K, V>> combine(@NotNull Promise<K> p1, @NotNull Promise<V> p2) {
return combine(false, p1, p2);
}
<K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, @Nullable BiConsumer<K, Throwable> exceptionHandler);
<K, V> @NotNull Promise<Map.Entry<K, V>> combine(boolean propagateCancel, @NotNull Promise<K> p1, @NotNull Promise<V> p2);
default <K, V> @NotNull Promise<Map<K, V>> combine(boolean propagateCancel, @NotNull Map<K, Promise<V>> promises) {
return combine(propagateCancel, promises, null);
}
<K, V> @NotNull Promise<Map<K, V>> combine(boolean propagateCancel, @NotNull Map<K, Promise<V>> promises, @Nullable BiConsumer<K, Throwable> exceptionHandler);
default <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises) {
return combine(promises, null);
}
<V> @NotNull Promise<List<V>> combine(@NotNull Iterable<Promise<V>> promises, @Nullable Consumer<Throwable> exceptionHandler);
default <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, @Nullable BiConsumer<K, Throwable> exceptionHandler) {
return combine(false, promises, exceptionHandler);
}
default <V> @NotNull Promise<List<V>> combine(boolean propagateCancel, @NotNull Iterable<Promise<V>> promises) {
return combine(propagateCancel, promises, null);
}
<V> @NotNull Promise<List<V>> combine(boolean propagateCancel, @NotNull Iterable<Promise<V>> promises, @Nullable BiConsumer<Integer, Throwable> exceptionHandler);
default <V> @NotNull Promise<List<V>> combine(@NotNull Iterable<Promise<V>> promises) {
return combine(promises, null);
}
@NotNull Promise<List<PromiseCompletion<?>>> allSettled(@NotNull Iterable<Promise<?>> promiseIterable);
default <V> @NotNull Promise<List<V>> combine(@NotNull Iterable<Promise<V>> promises, @Nullable BiConsumer<Integer, Throwable> exceptionHandler) {
return combine(false, promises, exceptionHandler);
}
default @NotNull Promise<List<PromiseCompletion<?>>> allSettled(@NotNull Iterable<Promise<?>> promiseIterable) {
return allSettled(false, promiseIterable);
}
@NotNull Promise<List<PromiseCompletion<?>>> allSettled(boolean propagateCancel, @NotNull Iterable<Promise<?>> promiseIterable);
default @NotNull Promise<List<PromiseCompletion<?>>> allSettled(@NotNull Promise<?>... promiseArray) {
return allSettled(Arrays.asList(promiseArray));
return allSettled(false, promiseArray);
}
@NotNull Promise<Void> all(@NotNull Iterable<Promise<?>> promiseIterable);
default @NotNull Promise<List<PromiseCompletion<?>>> allSettled(boolean propagateCancel, @NotNull Promise<?>... promiseArray) {
return allSettled(propagateCancel, Arrays.asList(promiseArray));
}
default @NotNull Promise<Void> all(@NotNull Iterable<Promise<?>> promiseIterable) {
return all(false, promiseIterable);
}
@NotNull Promise<Void> all(boolean propagateCancel, @NotNull Iterable<Promise<?>> promiseIterable);
default @NotNull Promise<Void> all(@NotNull Promise<?>... promiseArray) {
return all(Arrays.asList(promiseArray));
return all(false, promiseArray);
}
default @NotNull Promise<Void> all(boolean propagateCancel, @NotNull Promise<?>... promiseArray) {
return all(propagateCancel, Arrays.asList(promiseArray));
}
default <V> @NotNull Promise<V> race(@NotNull Iterable<Promise<V>> promises) {
return race(false, promises);
}
<V> @NotNull Promise<V> race(boolean cancelRaceLosers, @NotNull Iterable<Promise<V>> promises);
<T> @NotNull Promise<T> wrap(@NotNull CompletableFuture<T> future);
<T> @NotNull Promise<T> wrap(@NotNull Mono<T> mono);
<T> @NotNull Promise<T> resolve(T value);
default @NotNull Promise<Void> start() {
return resolve(null);
}
<T> @NotNull Promise<T> resolve(T value);
<T> @NotNull Promise<T> error(@NotNull Throwable error);
@NotNull Promise<Void> erase(@NotNull Promise<?> p);
}

44
futur-api/src/main/java/dev/tommyjs/futur/promise/Promises.java
View File

@@ -17,38 +17,38 @@ import java.util.function.BiConsumer;
@Deprecated
public class Promises {
public static <K, V> @NotNull Promise<Map.Entry<K, V>> combine(@NotNull Promise<K> p1, @NotNull Promise<V> p2, PromiseFactory factory) {
return factory.combine(p1, p2);
}
public static <K, V> @NotNull Promise<Map.Entry<K, V>> combine(@NotNull Promise<K> p1, @NotNull Promise<V> p2) {
return combine(p1, p2, p1.getFactory());
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, long timeout, @Nullable BiConsumer<K, Throwable> exceptionHandler, PromiseFactory factory) {
return factory.combine(promises, exceptionHandler).timeout(timeout);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, long timeout, boolean strict, PromiseFactory factory) {
return combine(promises, timeout, strict ? null : (_k, _v) -> {}, factory);
public static <K, V> @NotNull Promise<Map.Entry<K, V>> combine(@NotNull Promise<K> p1, @NotNull Promise<V> p2, PromiseFactory factory) {
return factory.combine(p1, p2);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, long timeout, PromiseFactory factory) {
return combine(promises, timeout, true, factory);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, long timeout, boolean strict, PromiseFactory factory) {
return combine(promises, timeout, strict ? null : (_k, _v) -> {}, factory);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, long timeout, @Nullable BiConsumer<K, Throwable> exceptionHandler, PromiseFactory factory) {
return factory.combine(promises, exceptionHandler).timeout(timeout);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, PromiseFactory factory) {
return combine(promises, 1500L, true, factory);
}
public static <V> @NotNull Promise<List<V>> combine(@NotNull List<Promise<V>> promises, long timeout, boolean strict, PromiseFactory factory) {
return factory.combine(promises, strict ? null : (_v) -> {}).timeout(timeout);
}
public static <V> @NotNull Promise<List<V>> combine(@NotNull List<Promise<V>> promises, long timeout, PromiseFactory factory) {
return combine(promises, timeout, true, factory);
}
public static <V> @NotNull Promise<List<V>> combine(@NotNull List<Promise<V>> promises, long timeout, boolean strict, PromiseFactory factory) {
return factory.combine(promises, strict ? null : (_i, _v) -> {}).timeout(timeout);
}
public static <V> @NotNull Promise<List<V>> combine(@NotNull List<Promise<V>> promises, PromiseFactory factory) {
return combine(promises, 1500L, true, factory);
}
@@ -57,6 +57,10 @@ public class Promises {
return factory.all(promises);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Collection<K> keys, @NotNull ExceptionalFunction<K, V> mapper, long timeout, PromiseFactory factory) {
return combine(keys, mapper, timeout, true, factory);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Collection<K> keys, @NotNull ExceptionalFunction<K, V> mapper, long timeout, boolean strict, PromiseFactory factory) {
Map<K, Promise<V>> promises = new HashMap<>();
for (K key : keys) {
@@ -67,22 +71,18 @@ public class Promises {
return combine(promises, timeout, strict, factory);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Collection<K> keys, @NotNull ExceptionalFunction<K, V> mapper, long timeout, PromiseFactory factory) {
return combine(keys, mapper, timeout, true, factory);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Collection<K> keys, @NotNull ExceptionalFunction<K, V> mapper, PromiseFactory factory) {
return combine(keys, mapper, 1500L, true, factory);
}
public static @NotNull Promise<Void> erase(@NotNull Promise<?> p, PromiseFactory factory) {
return factory.erase(p);
}
public static @NotNull Promise<Void> erase(@NotNull Promise<?> p) {
return erase(p, p.getFactory());
}
public static @NotNull Promise<Void> erase(@NotNull Promise<?> p, PromiseFactory factory) {
return p.erase();
}
public static <T> @NotNull Promise<T> wrap(@NotNull CompletableFuture<T> future, PromiseFactory factory) {
return factory.wrap(future);
}

120
futur-api/src/test/java/dev/tommyjs/futur/PromiseTests.java
View File

@@ -9,19 +9,22 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
public final class PromiseTests {
private final Logger logger = LoggerFactory.getLogger(PromiseTests.class);
private final PromiseExecutor<ScheduledFuture<?>> executor = SinglePoolExecutor.create(1);
private final PromiseExecutor<Future<?>> executor = SinglePoolExecutor.create(5);
private final PromiseFactory pfac = new SimplePromiseFactory<>(executor, logger);
@Test
void testMono() {
public void testMono() {
Exception value = new Exception("Test Error");
var error = pfac.wrap(Mono.error(value));
@@ -34,15 +37,118 @@ public final class PromiseTests {
}
@Test
void testErrorCancellation() throws InterruptedException {
var finish = new AtomicBoolean();
public void testErrorCancellation() throws InterruptedException {
var finished = new AtomicBoolean();
pfac.start()
.thenRunDelayedAsync(() -> finish.set(true), 50, TimeUnit.MILLISECONDS)
.thenRunDelayedAsync(() -> finished.set(true), 50, TimeUnit.MILLISECONDS)
.thenRunAsync(() -> {})
.cancel();
Thread.sleep(100L);
assert !finish.get();
assert !finished.get();
}
@Test
public void testCombineUtil() throws TimeoutException {
pfac.all(
pfac.start().thenRunDelayedAsync(() -> {}, 50, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> {}, 50, TimeUnit.MILLISECONDS)
)
.join(100L);
pfac.allSettled(
pfac.start().thenRunDelayedAsync(() -> {}, 50, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> {}, 50, TimeUnit.MILLISECONDS)
)
.join(100L);
pfac.combine(
pfac.start().thenRunDelayedAsync(() -> {}, 50, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> {}, 50, TimeUnit.MILLISECONDS)
)
.join(100L);
pfac.combine(
List.of(
pfac.start().thenRunDelayedAsync(() -> {}, 49, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> {}, 50, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> {}, 51, TimeUnit.MILLISECONDS)
)
)
.join(100L);
pfac.combine(
Map.of(
"a", pfac.start().thenRunDelayedAsync(() -> {}, 50, TimeUnit.MILLISECONDS),
"b", pfac.start().thenRunDelayedAsync(() -> {}, 50, TimeUnit.MILLISECONDS)
)
)
.join(100L);
}
@Test
public void testCombineUtilPropagation() throws InterruptedException {
var finished1 = new AtomicBoolean();
pfac.all(
true,
pfac.start().thenRunDelayedAsync(() -> finished1.set(true), 50, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> finished1.set(true), 50, TimeUnit.MILLISECONDS)
)
.cancel();
var finished2 = new AtomicBoolean();
pfac.allSettled(
true,
pfac.start().thenRunDelayedAsync(() -> finished2.set(true), 50, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> finished2.set(true), 50, TimeUnit.MILLISECONDS)
)
.cancel();
var finished3 = new AtomicBoolean();
pfac.combine(
true,
pfac.start().thenRunDelayedAsync(() -> finished3.set(true), 50, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> finished3.set(true), 50, TimeUnit.MILLISECONDS)
)
.cancel();
var finished4 = new AtomicBoolean();
pfac.combine(
true,
List.of(
pfac.start().thenRunDelayedAsync(() -> finished4.set(true), 50, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> finished4.set(true), 50, TimeUnit.MILLISECONDS),
pfac.start().thenRunDelayedAsync(() -> finished4.set(true), 50, TimeUnit.MILLISECONDS)
)
)
.cancel();
var finished5 = new AtomicBoolean();
pfac.combine(
true,
Map.of(
"a", pfac.start().thenRunDelayedAsync(() -> finished5.set(true), 50, TimeUnit.MILLISECONDS),
"b", pfac.start().thenRunDelayedAsync(() -> finished5.set(true), 50, TimeUnit.MILLISECONDS)
)
)
.cancel();
Thread.sleep(100L);
assert !finished1.get();
assert !finished2.get();
assert !finished3.get();
assert !finished4.get();
assert !finished5.get();
}
@Test
public void testRace() throws TimeoutException {
assert pfac.race(
List.of(
pfac.start().thenSupplyDelayedAsync(() -> true, 50, TimeUnit.MILLISECONDS),
pfac.start().thenSupplyDelayedAsync(() -> false, 150, TimeUnit.MILLISECONDS)
)
).join(100L);
}
}