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

Merge pull request #5 from tommyskeff/release/2.0.0

Browse Source 
Release 2.0.0
This commit is contained in:
Tommy
2024-01-02 22:50:10 +00:00
committed by GitHub
parent 11ed692e75 0a05da2aeb
commit 49b701c736
26 changed files with 648 additions and 836 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
futur-api/build.gradle.kts
View File

@@ -6,7 +6,7 @@ plugins {
}
group = "dev.tommyjs"
version = "1.0.1"
version = "2.0.0"
repositories {
mavenCentral()

3
futur-api/src/main/java/dev/tommyjs/futur/function/ExceptionalConsumer.java
View File

@@ -1,7 +1,8 @@
package dev.tommyjs.futur.function;
@FunctionalInterface
public interface ExceptionalConsumer<T> {
void accept(T value) throws Exception;
void accept(T value) throws Throwable;
}

3
futur-api/src/main/java/dev/tommyjs/futur/function/ExceptionalFunction.java
View File

@@ -1,7 +1,8 @@
package dev.tommyjs.futur.function;
@FunctionalInterface
public interface ExceptionalFunction<K, V> {
V apply(K value) throws Exception;
V apply(K value) throws Throwable;
}

3
futur-api/src/main/java/dev/tommyjs/futur/function/ExceptionalRunnable.java
View File

@@ -1,7 +1,8 @@
package dev.tommyjs.futur.function;
@FunctionalInterface
public interface ExceptionalRunnable {
void run() throws Exception;
void run() throws Throwable;
}

3
futur-api/src/main/java/dev/tommyjs/futur/function/ExceptionalSupplier.java
View File

@@ -1,7 +1,8 @@
package dev.tommyjs.futur.function;
@FunctionalInterface
public interface ExceptionalSupplier<T> {
T get() throws Exception;
T get() throws Throwable;
}

36
futur-api/src/main/java/dev/tommyjs/futur/impl/SimplePromise.java Normal file
View File

@@ -0,0 +1,36 @@
package dev.tommyjs.futur.impl;
import dev.tommyjs.futur.promise.AbstractPromise;
import dev.tommyjs.futur.promise.PromiseFactory;
import org.slf4j.Logger;
import java.util.concurrent.ScheduledExecutorService;
public class SimplePromise<T> extends AbstractPromise<T> {
private final ScheduledExecutorService executor;
private final Logger logger;
private final PromiseFactory factory;
public SimplePromise(ScheduledExecutorService executor, Logger logger, PromiseFactory factory) {
this.executor = executor;
this.logger = logger;
this.factory = factory;
}
@Override
protected ScheduledExecutorService getExecutor() {
return executor;
}
@Override
protected Logger getLogger() {
return logger;
}
@Override
public PromiseFactory getFactory() {
return factory;
}
}

40
futur-api/src/main/java/dev/tommyjs/futur/impl/SimplePromiseFactory.java Normal file
View File

@@ -0,0 +1,40 @@
package dev.tommyjs.futur.impl;
import dev.tommyjs.futur.promise.AbstractPromise;
import dev.tommyjs.futur.promise.Promise;
import dev.tommyjs.futur.promise.PromiseFactory;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import java.util.concurrent.ScheduledExecutorService;
public class SimplePromiseFactory implements PromiseFactory {
private final ScheduledExecutorService executor;
private final Logger logger;
public SimplePromiseFactory(ScheduledExecutorService executor, Logger logger) {
this.executor = executor;
this.logger = logger;
}
@Override
public @NotNull <T> Promise<T> resolve(T value) {
AbstractPromise<T> promise = new SimplePromise<>(executor, logger, this);
promise.complete(value);
return promise;
}
@Override
public @NotNull <T> Promise<T> unresolved() {
return new SimplePromise<>(executor, logger, this);
}
@Override
public @NotNull <T> Promise<T> error(Throwable error) {
AbstractPromise<T> promise = new SimplePromise<>(executor, logger, this);
promise.completeExceptionally(error);
return promise;
}
}

365
futur-api/src/main/java/dev/tommyjs/futur/promise/AbstractPromise.java Normal file
View File

@@ -0,0 +1,365 @@
package dev.tommyjs.futur.promise;
import dev.tommyjs.futur.function.ExceptionalConsumer;
import dev.tommyjs.futur.function.ExceptionalFunction;
import dev.tommyjs.futur.function.ExceptionalRunnable;
import dev.tommyjs.futur.function.ExceptionalSupplier;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import java.util.Collection;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
public abstract class AbstractPromise<T> implements Promise<T> {
private final Collection<PromiseListener<T>> listeners;
private @Nullable PromiseCompletion<T> completion;
public AbstractPromise() {
this.listeners = new ConcurrentLinkedQueue<>();
this.completion = null;
}
protected abstract ScheduledExecutorService getExecutor();
protected abstract Logger getLogger();
@Override
public T join(long interval, long timeout) throws TimeoutException {
long start = System.currentTimeMillis();
while (!isCompleted()) {
if (System.currentTimeMillis() > start + timeout)
throw new TimeoutException("Promise timed out after " + timeout + "ms");
try {
Thread.sleep(interval);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
PromiseCompletion<T> completion = getCompletion();
if (completion == null) {
throw new IllegalStateException();
}
if (completion.isError()) {
throw new RuntimeException(completion.getException());
}
return completion.getResult();
}
@Override
public @NotNull Promise<Void> thenRunSync(@NotNull ExceptionalRunnable task) {
return thenApplySync(result -> {
task.run();
return null;
});
}
@Override
public @NotNull Promise<Void> thenRunDelayedSync(@NotNull ExceptionalRunnable task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedSync(result -> {
task.run();
return null;
}, delay, unit);
}
@Override
public @NotNull Promise<Void> thenConsumeSync(@NotNull ExceptionalConsumer<T> task) {
return thenApplySync(result -> {
task.accept(result);
return null;
});
}
@Override
public @NotNull Promise<Void> thenConsumeDelayedSync(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedSync(result -> {
task.accept(result);
return null;
}, delay, unit);
}
@Override
public <V> @NotNull Promise<V> thenSupplySync(@NotNull ExceptionalSupplier<V> task) {
return thenApplySync(result -> task.get());
}
@Override
public <V> @NotNull Promise<V> thenSupplyDelayedSync(@NotNull ExceptionalSupplier<V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedSync(result -> task.get(), delay, unit);
}
@Override
public <V> @NotNull Promise<V> thenApplySync(@NotNull ExceptionalFunction<T, V> task) {
Promise<V> promise = getFactory().unresolved();
addListener(ctx -> {
if (ctx.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx.getException());
return;
}
Runnable runnable = createRunnable(ctx, promise, task);
getExecutor().submit(runnable);
});
return promise;
}
@Override
public <V> @NotNull Promise<V> thenApplyDelayedSync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit) {
Promise<V> promise = getFactory().unresolved();
addListener(ctx -> {
if (ctx.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx.getException());
return;
}
Runnable runnable = createRunnable(ctx, promise, task);
getExecutor().schedule(runnable, delay, unit);
});
return promise;
}
@Override
public <V> @NotNull Promise<V> thenComposeSync(@NotNull ExceptionalFunction<T, @NotNull Promise<V>> task) {
Promise<V> promise = getFactory().unresolved();
thenApplySync(task).thenConsumeAsync(nestedPromise -> {
nestedPromise.addListener(ctx1 -> {
if (ctx1.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx1.getException());
return;
}
promise.complete(ctx1.getResult());
});
}).addListener(ctx2 -> {
if (ctx2.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx2.getException());
}
});
return promise;
}
@Override
public @NotNull Promise<Void> thenRunAsync(@NotNull ExceptionalRunnable task) {
return thenApplyAsync(result -> {
task.run();
return null;
});
}
@Override
public @NotNull Promise<Void> thenRunDelayedAsync(@NotNull ExceptionalRunnable task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedAsync(result -> {
task.run();
return null;
}, delay, unit);
}
@Override
public @NotNull Promise<Void> thenConsumeAsync(@NotNull ExceptionalConsumer<T> task) {
return thenApplyAsync(result -> {
task.accept(result);
return null;
});
}
@Override
public @NotNull Promise<Void> thenConsumeDelayedAsync(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedAsync(result -> {
task.accept(result);
return null;
}, delay, unit);
}
@Override
public <V> @NotNull Promise<V> thenSupplyAsync(@NotNull ExceptionalSupplier<V> task) {
return thenApplyAsync(result -> task.get());
}
@Override
public <V> @NotNull Promise<V> thenSupplyDelayedAsync(@NotNull ExceptionalSupplier<V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedAsync(result -> task.get(), delay, unit);
}
@Override
public @NotNull Promise<T> thenPopulateReference(@NotNull AtomicReference<T> reference) {
return thenApplyAsync((result) -> {
reference.set(result);
return result;
});
}
@Override
public <V> @NotNull Promise<V> thenApplyAsync(@NotNull ExceptionalFunction<T, V> task) {
Promise<V> promise = getFactory().unresolved();
addListener(ctx -> {
if (ctx.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx.getException());
return;
}
Runnable runnable = createRunnable(ctx, promise, task);
getExecutor().submit(runnable);
});
return promise;
}
@Override
public <V> @NotNull Promise<V> thenApplyDelayedAsync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit) {
Promise<V> promise = getFactory().unresolved();
addListener(ctx -> {
Runnable runnable = createRunnable(ctx, promise, task);
getExecutor().schedule(runnable, delay, unit);
});
return promise;
}
@Override
public <V> @NotNull Promise<V> thenComposeAsync(@NotNull ExceptionalFunction<T, Promise<V>> task) {
Promise<V> promise = getFactory().unresolved();
thenApplyAsync(task).thenConsumeAsync(nestedPromise -> {
nestedPromise.addListener(ctx1 -> {
if (ctx1.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx1.getException());
return;
}
promise.complete(ctx1.getResult());
});
}).addListener(ctx2 -> {
if (ctx2.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx2.getException());
}
});
return promise;
}
private <V> @NotNull Runnable createRunnable(@NotNull PromiseCompletion<T> ctx, @NotNull Promise<V> promise, @NotNull ExceptionalFunction<T, V> task) {
return () -> {
if (ctx.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx.getException());
return;
}
try {
V result = task.apply(ctx.getResult());
promise.complete(result);
} catch (Throwable e) {
promise.completeExceptionally(e);
}
};
}
@Override
public @NotNull Promise<T> logExceptions() {
return addListener(ctx -> {
if (ctx.isError()) {
getLogger().error("Exception caught in promise chain", ctx.getException());
}
});
}
@Override
public @NotNull Promise<T> addListener(@NotNull PromiseListener<T> listener) {
if (isCompleted()) {
getExecutor().submit(() -> {
try {
//noinspection ConstantConditions
listener.handle(getCompletion());
} catch (Exception e) {
getLogger().error("Exception caught in promise listener", e);
}
});
} else {
getListeners().add(listener);
}
return this;
}
@Override
public @NotNull Promise<T> timeout(long time, @NotNull TimeUnit unit) {
getExecutor().schedule(() -> {
if (!isCompleted()) {
completeExceptionally(new TimeoutException("Promise timed out after " + time + " " + unit));
}
}, time, unit);
return this;
}
@Override
public @NotNull Promise<T> timeout(long ms) {
return timeout(ms, TimeUnit.MILLISECONDS);
}
protected void handleCompletion(@NotNull PromiseCompletion<T> ctx) {
if (this.isCompleted()) return;
setCompletion(ctx);
getExecutor().submit(() -> {
for (PromiseListener<T> listener : getListeners()) {
if (!ctx.isActive()) return;
try {
listener.handle(ctx);
} catch (Exception e) {
e.printStackTrace();
getLogger().error("Exception caught in promise listener", e);
}
}
});
}
@Override
public void complete(@Nullable T result) {
handleCompletion(new PromiseCompletion<>(result));
}
@Override
public void completeExceptionally(@NotNull Throwable result) {
handleCompletion(new PromiseCompletion<>(result));
}
@Override
public boolean isCompleted() {
return getCompletion() != null;
}
protected Collection<PromiseListener<T>> getListeners() {
return listeners;
}
@Override
public @Nullable PromiseCompletion<T> getCompletion() {
return completion;
}
protected void setCompletion(@NotNull PromiseCompletion<T> completion) {
this.completion = completion;
}
}

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

@@ -4,406 +4,87 @@ import dev.tommyjs.futur.function.ExceptionalConsumer;
import dev.tommyjs.futur.function.ExceptionalFunction;
import dev.tommyjs.futur.function.ExceptionalRunnable;
import dev.tommyjs.futur.function.ExceptionalSupplier;
import dev.tommyjs.futur.scheduler.Schedulers;
import dev.tommyjs.futur.trace.ExecutorTrace;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;
public class Promise<T> {
public interface Promise<T> {
private static final String PACKAGE;
private static final Logger LOGGER = LoggerFactory.getLogger(Promise.class);
static {
String[] packageElements = Promise.class.getPackageName().split("\\.");
int i = 0;
StringBuilder packageBuilder = new StringBuilder();
while (i < 3) {
packageBuilder.append(packageElements[i]);
i++;
}
PACKAGE = packageBuilder.toString();
}
private final Collection<PromiseListener<T>> listeners;
private final StackTraceElement[] stackTrace;
private @Nullable PromiseCompletion<T> completion;
public Promise() {
this.listeners = new ConcurrentLinkedQueue<>();
this.completion = null;
this.stackTrace = Arrays.stream(Thread.currentThread().getStackTrace())
.filter(v -> !v.getClassName().startsWith(PACKAGE))
.toArray(StackTraceElement[]::new);
}
public T join(long interval, long timeout) throws TimeoutException {
long start = System.currentTimeMillis();
while (!isCompleted()) {
if (System.currentTimeMillis() > start + timeout)
throw new TimeoutException("Promise timed out after " + timeout + "ms");
try {
Thread.sleep(interval);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
PromiseCompletion<T> completion = getCompletion();
if (completion == null) {
throw new IllegalStateException();
}
if (completion.isError()) {
throw new RuntimeException(completion.getException());
}
return completion.getResult();
}
public @NotNull Promise<Void> thenRunSync(@NotNull ExceptionalRunnable task) {
return thenApplySync(result -> {
task.run();
return null;
}, Schedulers.getTrace(task));
}
public @NotNull Promise<Void> thenRunDelayedSync(@NotNull ExceptionalRunnable task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedSync(result -> {
task.run();
return null;
}, delay, unit, Schedulers.getTrace(task));
}
public @NotNull Promise<Void> thenConsumeSync(@NotNull ExceptionalConsumer<T> task) {
return thenApplySync(result -> {
task.accept(result);
return null;
}, Schedulers.getTrace(task));
}
public @NotNull Promise<Void> thenConsumeDelayedSync(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedSync(result -> {
task.accept(result);
return null;
}, delay, unit, Schedulers.getTrace(task));
}
public <V> @NotNull Promise<V> thenSupplySync(@NotNull ExceptionalSupplier<V> task) {
return thenApplySync(result -> task.get(), Schedulers.getTrace(task));
}
public <V> @NotNull Promise<V> thenSupplyDelayedSync(@NotNull ExceptionalSupplier<V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedSync(result -> task.get(), delay, unit, Schedulers.getTrace(task));
}
public <V> @NotNull Promise<V> thenApplySync(@NotNull ExceptionalFunction<T, V> task, @NotNull ExecutorTrace trace) {
Promise<V> promise = new Promise<>();
addListener(ctx -> {
if (ctx.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx.getException());
return;
}
Runnable runnable = createRunnable(ctx, promise, task);
Schedulers.runSync(runnable, trace);
});
return promise;
}
public <V> @NotNull Promise<V> thenApplySync(@NotNull ExceptionalFunction<T, V> task) {
return thenApplySync(task, Schedulers.getTrace(task));
}
public <V> @NotNull Promise<V> thenApplyDelayedSync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
Promise<V> promise = new Promise<>();
addListener(ctx -> {
if (ctx.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx.getException());
return;
}
Runnable runnable = createRunnable(ctx, promise, task);
Schedulers.runDelayedSync(runnable, delay, unit, trace);
});
return promise;
}
public <V> @NotNull Promise<V> thenApplyDelayedSync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedSync(task, delay, unit, Schedulers.getTrace(task));
}
public <V> @NotNull Promise<V> thenComposeSync(@NotNull ExceptionalFunction<T, @NotNull Promise<V>> task) {
Promise<V> promise = new Promise<>();
thenApplySync(task, Schedulers.getTrace(task)).thenConsumeAsync(nestedPromise -> {
nestedPromise.addListener(ctx1 -> {
if (ctx1.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx1.getException());
return;
}
promise.complete(ctx1.getResult());
});
}).addListener(ctx2 -> {
if (ctx2.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx2.getException());
}
});
return promise;
static <T> @NotNull Promise<T> resolve(T value, PromiseFactory factory) {
return factory.resolve(value);
}
public @NotNull Promise<Void> thenRunAsync(@NotNull ExceptionalRunnable task) {
return thenApplyAsync(result -> {
task.run();
return null;
}, Schedulers.getTrace(task));
static <T> @NotNull Promise<T> error(Throwable error, PromiseFactory factory) {
return factory.error(error);
}
public @NotNull Promise<Void> thenRunDelayedAsync(@NotNull ExceptionalRunnable task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedAsync(result -> {
task.run();
return null;
}, delay, unit, Schedulers.getTrace(task));
static <T> @NotNull Promise<T> unresolved(PromiseFactory factory) {
return factory.unresolved();
}
public @NotNull Promise<Void> thenConsumeAsync(@NotNull ExceptionalConsumer<T> task) {
return thenApplyAsync(result -> {
task.accept(result);
return null;
}, Schedulers.getTrace(task));
static @NotNull Promise<Void> start(PromiseFactory factory) {
return factory.resolve(null);
}
public @NotNull Promise<Void> thenConsumeDelayedAsync(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedAsync(result -> {
task.accept(result);
return null;
}, delay, unit, Schedulers.getTrace(task));
}
PromiseFactory getFactory();
@Deprecated(forRemoval = true)
public @NotNull Promise<Void> thenConsumerDelayedAsync(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit) {
return thenConsumeDelayedAsync(task, delay, unit);
}
T join(long interval, long timeout) throws TimeoutException;
public <V> @NotNull Promise<V> thenSupplyAsync(@NotNull ExceptionalSupplier<V> task) {
return thenApplyAsync(result -> task.get(), Schedulers.getTrace(task));
}
@NotNull Promise<Void> thenRunSync(@NotNull ExceptionalRunnable task);
public <V> @NotNull Promise<V> thenSupplyDelayedAsync(@NotNull ExceptionalSupplier<V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedAsync(result -> task.get(), delay, unit, Schedulers.getTrace(task));
}
@NotNull Promise<Void> thenRunDelayedSync(@NotNull ExceptionalRunnable task, long delay, @NotNull TimeUnit unit);
public @NotNull Promise<T> thenPopulateReference(@NotNull AtomicReference<T> reference) {
return thenApplyAsync((result) -> {
reference.set(result);
return result;
});
}
@NotNull Promise<Void> thenConsumeSync(@NotNull ExceptionalConsumer<T> task);
public <V> @NotNull Promise<V> thenApplyAsync(@NotNull ExceptionalFunction<T, V> task, @NotNull ExecutorTrace trace) {
Promise<V> promise = new Promise<>();
addListener(ctx -> {
createRunnable(ctx, promise, task).run();
});
@NotNull Promise<Void> thenConsumeDelayedSync(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit);
return promise;
}
<V> @NotNull Promise<V> thenSupplySync(@NotNull ExceptionalSupplier<V> task);
public <V> @NotNull Promise<V> thenApplyAsync(@NotNull ExceptionalFunction<T, V> task) {
return thenApplyAsync(task, Schedulers.getTrace(task));
}
<V> @NotNull Promise<V> thenSupplyDelayedSync(@NotNull ExceptionalSupplier<V> task, long delay, @NotNull TimeUnit unit);
public <V> @NotNull Promise<V> thenApplyDelayedAsync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
Promise<V> promise = new Promise<>();
addListener(ctx -> {
Runnable runnable = createRunnable(ctx, promise, task);
Schedulers.runDelayedAsync(runnable, delay, unit, trace);
});
<V> @NotNull Promise<V> thenApplySync(@NotNull ExceptionalFunction<T, V> task);
return promise;
}
<V> @NotNull Promise<V> thenApplyDelayedSync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit);
public <V> @NotNull Promise<V> thenApplyDelayedAsync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit) {
return thenApplyDelayedAsync(task, delay, unit, Schedulers.getTrace(task));
}
<V> @NotNull Promise<V> thenComposeSync(@NotNull ExceptionalFunction<T, @NotNull Promise<V>> task);
public <V> @NotNull Promise<V> thenCompose(@NotNull ExceptionalFunction<T, Promise<V>> task) {
return this.thenComposeAsync(task);
}
@NotNull Promise<Void> thenRunAsync(@NotNull ExceptionalRunnable task);
public <V> @NotNull Promise<V> thenComposeAsync(@NotNull ExceptionalFunction<T, Promise<V>> task) {
Promise<V> promise = new Promise<>();
thenApplyAsync(task, Schedulers.getTrace(task)).thenConsumeAsync(nestedPromise -> {
nestedPromise.addListener(ctx1 -> {
if (ctx1.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx1.getException());
return;
}
@NotNull Promise<Void> thenRunDelayedAsync(@NotNull ExceptionalRunnable task, long delay, @NotNull TimeUnit unit);
promise.complete(ctx1.getResult());
});
}).addListener(ctx2 -> {
if (ctx2.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx2.getException());
}
});
@NotNull Promise<Void> thenConsumeAsync(@NotNull ExceptionalConsumer<T> task);
return promise;
}
@NotNull Promise<Void> thenConsumeDelayedAsync(@NotNull ExceptionalConsumer<T> task, long delay, @NotNull TimeUnit unit);
private <V> @NotNull Runnable createRunnable(@NotNull PromiseCompletion<T> ctx, @NotNull Promise<V> promise, @NotNull ExceptionalFunction<T, V> task) {
return () -> {
if (ctx.isError()) {
//noinspection ConstantConditions
promise.completeExceptionally(ctx.getException());
return;
}
<V> @NotNull Promise<V> thenSupplyAsync(@NotNull ExceptionalSupplier<V> task);
try {
V result = task.apply(ctx.getResult());
promise.complete(result);
} catch (Exception e) {
promise.completeExceptionally(e, true);
}
};
}
<V> @NotNull Promise<V> thenSupplyDelayedAsync(@NotNull ExceptionalSupplier<V> task, long delay, @NotNull TimeUnit unit);
public @NotNull Promise<T> logExceptions() {
return addListener(ctx -> {
if (ctx.isError()) {
LOGGER.error("Exception caught in promise pipeline", ctx.getException());
}
});
}
@NotNull Promise<T> thenPopulateReference(@NotNull AtomicReference<T> reference);
public @NotNull Promise<T> addListener(@NotNull PromiseListener<T> listener) {
if (isCompleted()) {
Schedulers.runAsync(() -> {
try {
listener.handle(getCompletion());
} catch (Exception e) {
LOGGER.error("Exception caught in promise listener", e);
}
}, Schedulers.getTrace(listener));
} else {
getListeners().add(listener);
}
<V> @NotNull Promise<V> thenApplyAsync(@NotNull ExceptionalFunction<T, V> task);
return this;
}
<V> @NotNull Promise<V> thenApplyDelayedAsync(@NotNull ExceptionalFunction<T, V> task, long delay, @NotNull TimeUnit unit);
public @NotNull Promise<T> timeout(long time, @NotNull TimeUnit unit) {
Schedulers.runDelayedAsync(() -> {
if (!isCompleted()) {
completeExceptionally(new TimeoutException("Promise timed out after " + time + " " + unit), true);
}
}, time, unit);
<V> @NotNull Promise<V> thenComposeAsync(@NotNull ExceptionalFunction<T, Promise<V>> task);
return this;
}
@NotNull Promise<T> logExceptions();
public @NotNull Promise<T> timeout(long ms) {
return timeout(ms, TimeUnit.MILLISECONDS);
}
@NotNull Promise<T> addListener(@NotNull PromiseListener<T> listener);
protected void handleCompletion(@NotNull PromiseCompletion<T> ctx) {
if (this.isCompleted()) return;
setCompletion(ctx);
@NotNull Promise<T> timeout(long time, @NotNull TimeUnit unit);
Schedulers.runAsync(() -> {
for (PromiseListener<T> listener : getListeners()) {
if (!ctx.isActive()) return;
@NotNull Promise<T> timeout(long ms);
try {
listener.handle(ctx);
} catch (Exception e) {
LOGGER.error("Exception caught in promise listener", e);
}
}
});
}
public void complete(@Nullable T result) {
handleCompletion(new PromiseCompletion<>(result));
}
public void completeExceptionally(@NotNull Throwable result, boolean appendStacktrace) {
if (appendStacktrace && this.stackTrace != null) {
result.setStackTrace(Stream.of(result.getStackTrace(), this.stackTrace)
.flatMap(Stream::of)
.filter(v -> !v.getClassName().startsWith(PACKAGE))
.filter(v -> !v.getClassName().startsWith("java.lang.Thread"))
.filter(v -> !v.getClassName().startsWith("java.util.concurrent"))
.toArray(StackTraceElement[]::new));
}
handleCompletion(new PromiseCompletion<>(result));
}
public void completeExceptionally(@NotNull Throwable result) {
completeExceptionally(result, false);
}
public boolean isCompleted() {
return getCompletion() != null;
}
protected Collection<PromiseListener<T>> getListeners() {
return listeners;
}
public @Nullable PromiseCompletion<T> getCompletion() {
return completion;
}
protected void setCompletion(@NotNull PromiseCompletion<T> completion) {
this.completion = completion;
}
public static <T> @NotNull Promise<T> resolve(T value) {
Promise<T> promise = new Promise<>();
promise.setCompletion(new PromiseCompletion<>(value));
return promise;
}
void complete(@Nullable T result);
public static <T> @NotNull Promise<T> error(Throwable error) {
Promise<T> promise = new Promise<>();
promise.completeExceptionally(error);
return promise;
}
void completeExceptionally(@NotNull Throwable result);
public static @NotNull Promise<Void> start() {
return Promise.resolve(null);
}
boolean isCompleted();
@Deprecated // use resolve()
public static <T> @NotNull Promise<T> start(T start) {
Promise<T> promise = new Promise<>();
promise.complete(start);
return promise;
}
@Nullable PromiseCompletion<T> getCompletion();
}

35
futur-api/src/main/java/dev/tommyjs/futur/promise/PromiseFactory.java Normal file
View File

@@ -0,0 +1,35 @@
package dev.tommyjs.futur.promise;
import dev.tommyjs.futur.impl.SimplePromiseFactory;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
public interface PromiseFactory {
<T> @NotNull Promise<T> resolve(T value);
<T> @NotNull Promise<T> unresolved();
<T> @NotNull Promise<T> error(Throwable error);
static PromiseFactory create(ScheduledExecutorService executor, Logger logger) {
return new SimplePromiseFactory(executor, logger);
}
static PromiseFactory create(ScheduledExecutorService executor) {
return create(executor, LoggerFactory.getLogger(SimplePromiseFactory.class));
}
static PromiseFactory create(int threadPoolSize) {
return create(Executors.newScheduledThreadPool(threadPoolSize));
}
static PromiseFactory create() {
return create(Runtime.getRuntime().availableProcessors());
}
}

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

@@ -13,8 +13,8 @@ import java.util.stream.Collectors;
public class Promises {
public static <K, V> @NotNull Promise<Map.Entry<K, V>> combine(@NotNull Promise<K> p1, @NotNull Promise<V> p2) {
Promise<Map.Entry<K, V>> promise = new Promise<>();
public static <K, V> @NotNull Promise<Map.Entry<K, V>> combine(@NotNull Promise<K> p1, @NotNull Promise<V> p2, PromiseFactory factory) {
Promise<Map.Entry<K, V>> promise = factory.unresolved();
p1.addListener(ctx -> {
if (ctx.isError()) {
//noinspection ConstantConditions
@@ -37,13 +37,15 @@ public class Promises {
return promise;
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, long timeout, @Nullable BiConsumer<K, Throwable> exceptionHandler) {
Map<K, V> map = new HashMap<>();
if (promises.isEmpty()) return Promise.resolve(map);
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) {
Map<K, V> map = new HashMap<>();
ReentrantLock lock = new ReentrantLock();
Promise<Map<K, V>> promise = new Promise<>();
Promise<Map<K, V>> promise = factory.unresolved();
for (Map.Entry<K, Promise<V>> entry : promises.entrySet()) {
entry.getValue().addListener((ctx) -> {
lock.lock();
@@ -69,25 +71,23 @@ public class Promises {
return promise.timeout(timeout);
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Map<K, Promise<V>> promises, long timeout, boolean strict) {
return combine(promises, timeout, strict ? null : (_k, _v) -> {});
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) {
return combine(promises, timeout, true);
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) {
return combine(promises, 1500L, true);
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) {
public static <V> @NotNull Promise<List<V>> combine(@NotNull List<Promise<V>> promises, long timeout, boolean strict, PromiseFactory factory) {
AtomicInteger index = new AtomicInteger();
return combine(
promises.stream()
.collect(Collectors.toMap(s -> index.getAndIncrement(), v -> v)),
timeout,
strict
promises.stream().collect(Collectors.toMap(s -> index.getAndIncrement(), v -> v)),
timeout, strict, factory
).thenApplySync(v ->
v.entrySet().stream()
.sorted(Map.Entry.comparingByKey())
@@ -96,18 +96,16 @@ public class Promises {
);
}
public static <V> @NotNull Promise<List<V>> combine(@NotNull List<Promise<V>> promises, long timeout) {
return combine(promises, timeout, true);
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) {
return combine(promises, 1500L, true);
public static <V> @NotNull Promise<List<V>> combine(@NotNull List<Promise<V>> promises, PromiseFactory factory) {
return combine(promises, 1500L, true, factory);
}
public static @NotNull Promise<Void> all(@NotNull List<Promise<?>> promises) {
if (promises.isEmpty()) return Promise.start();
Promise<Void> promise = new Promise<>();
public static @NotNull Promise<Void> all(@NotNull List<Promise<?>> promises, PromiseFactory factory) {
Promise<Void> promise = factory.unresolved();
for (Promise<?> p : promises) {
p.addListener((ctx) -> {
if (ctx.isError()) {
@@ -121,30 +119,26 @@ public class Promises {
return promise;
}
public static @NotNull Promise<Void> all(@NotNull Promise<?>... promises) {
return all(Arrays.asList(promises));
}
public static <K, V> @NotNull Promise<Map<K, V>> combine(@NotNull Collection<K> keys, @NotNull ExceptionalFunction<K, V> mapper, long timeout, boolean strict) {
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) {
Promise<V> promise = Promise.resolve(key).thenApplyAsync(mapper);
Promise<V> promise = factory.resolve(key).thenApplyAsync(mapper);
promises.put(key, promise);
}
return combine(promises, timeout, strict);
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) {
return combine(keys, mapper, timeout, true);
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) {
return combine(keys, mapper, 1500L, true);
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) {
Promise<Void> promise = new Promise<>();
public static @NotNull Promise<Void> erase(@NotNull Promise<?> p, PromiseFactory factory) {
Promise<Void> promise = factory.unresolved();
p.addListener(ctx -> {
if (ctx.isError()) {
//noinspection ConstantConditions
@@ -157,8 +151,12 @@ public class Promises {
return promise;
}
public static <T> @NotNull Promise<T> wrap(@NotNull CompletableFuture<T> future) {
Promise<T> promise = new Promise<>();
public static @NotNull Promise<Void> erase(@NotNull Promise<?> p) {
return erase(p, p.getFactory());
}
public static <T> @NotNull Promise<T> wrap(@NotNull CompletableFuture<T> future, PromiseFactory factory) {
Promise<T> promise = factory.unresolved();
future.whenComplete((result, e) -> {
if (e != null) {
promise.completeExceptionally(e);

24
futur-api/src/main/java/dev/tommyjs/futur/promise/StaticPromise.java Normal file
View File

@@ -0,0 +1,24 @@
package dev.tommyjs.futur.promise;
import org.slf4j.Logger;
import java.util.concurrent.ScheduledExecutorService;
public class StaticPromise<T> extends AbstractPromise<T> {
@Override
protected ScheduledExecutorService getExecutor() {
return StaticPromiseFactory.EXECUTOR;
}
@Override
protected Logger getLogger() {
return StaticPromiseFactory.LOGGER;
}
@Override
public PromiseFactory getFactory() {
return StaticPromiseFactory.INSTANCE;
}
}

44
futur-api/src/main/java/dev/tommyjs/futur/promise/StaticPromiseFactory.java Normal file
View File

@@ -0,0 +1,44 @@
package dev.tommyjs.futur.promise;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
public class StaticPromiseFactory implements PromiseFactory {
public static final @NotNull PromiseFactory INSTANCE;
public static final @NotNull ScheduledExecutorService EXECUTOR;
public static final @NotNull Logger LOGGER;
static {
INSTANCE = new StaticPromiseFactory();
EXECUTOR = Executors.newSingleThreadScheduledExecutor();
LOGGER = LoggerFactory.getLogger(StaticPromiseFactory.class);
}
private StaticPromiseFactory() {
}
@Override
public @NotNull <T> Promise<T> resolve(T value) {
AbstractPromise<T> promise = new StaticPromise<>();
promise.setCompletion(new PromiseCompletion<>(value));
return promise;
}
@Override
public @NotNull <T> Promise<T> unresolved() {
return new StaticPromise<>();
}
@Override
public @NotNull <T> Promise<T> error(Throwable error) {
AbstractPromise<T> promise = new StaticPromise<>();
promise.completeExceptionally(error);
return promise;
}
}

37
futur-api/src/main/java/dev/tommyjs/futur/scheduler/Scheduler.java
View File

@@ -1,37 +0,0 @@
package dev.tommyjs.futur.scheduler;
import dev.tommyjs.futur.trace.ExecutorTrace;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
public interface Scheduler {
Logger LOGGER = LoggerFactory.getLogger(Scheduler.class);
void runSync(@NotNull Runnable task, @NotNull ExecutorTrace trace);
void runDelayedSync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace);
void runRepeatingSync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace);
void runAsync(@NotNull Runnable task, @NotNull ExecutorTrace trace);
void runDelayedAsync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace);
void runRepeatingAsync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace);
default @NotNull Runnable wrapExceptions(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
return () -> {
try {
task.run();
} catch (Exception e) {
LOGGER.error("Exception in scheduled task: {}", e.getClass().getName());
LOGGER.error(trace.toString());
}
};
}
}

105
futur-api/src/main/java/dev/tommyjs/futur/scheduler/Schedulers.java
View File

@@ -1,105 +0,0 @@
package dev.tommyjs.futur.scheduler;
import dev.tommyjs.futur.trace.ExecutorTrace;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
@SuppressWarnings("ConstantConditions")
public class Schedulers {
private static final Logger LOGGER = LoggerFactory.getLogger(Schedulers.class);
private static @Nullable Scheduler scheduler;
public static void runSync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
ensureLoaded();
getScheduler().runSync(task, trace);
}
public static void runSync(@NotNull Runnable task) {
ensureLoaded();
getScheduler().runSync(task, Schedulers.getTrace(task));
}
public static void runDelayedSync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
ensureLoaded();
getScheduler().runDelayedSync(task, delay, unit, trace);
}
public static void runDelayedSync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit) {
ensureLoaded();
getScheduler().runDelayedSync(task, delay, unit, Schedulers.getTrace(task));
}
public static void runRepeatingSync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
ensureLoaded();
getScheduler().runRepeatingSync(task, interval, unit, trace);
}
public static void runRepeatingSync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit) {
ensureLoaded();
getScheduler().runRepeatingSync(task, interval, unit, Schedulers.getTrace(task));
}
public static void runAsync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
ensureLoaded();
getScheduler().runAsync(task, trace);
}
public static void runAsync(@NotNull Runnable task) {
ensureLoaded();
getScheduler().runAsync(task, Schedulers.getTrace(task));
}
public static void runDelayedAsync(@NotNull Runnable task, long delay, TimeUnit unit, ExecutorTrace trace) {
ensureLoaded();
getScheduler().runDelayedAsync(task, delay, unit, trace);
}
public static void runDelayedAsync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit) {
ensureLoaded();
getScheduler().runDelayedAsync(task, delay, unit, Schedulers.getTrace(task));
}
public static void runRepeatingAsync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
ensureLoaded();
getScheduler().runRepeatingAsync(task, interval, unit, trace);
}
public static void runRepeatingAsync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit) {
ensureLoaded();
getScheduler().runRepeatingAsync(task, interval, unit, Schedulers.getTrace(task));
}
public static ExecutorTrace getTrace(@NotNull Object function) {
return new ExecutorTrace(function.getClass(), Thread.currentThread().getStackTrace());
}
public static void ensureLoaded() {
if (getScheduler() == null) {
LOGGER.warn("No scheduler loaded, falling back to default single threaded scheduler");
setScheduler(SingleExecutorScheduler.create());
}
}
public static void loadDefaultScheduler() {
}
public static boolean isLoaded() {
return getScheduler() != null;
}
public static @Nullable Scheduler getScheduler() {
return scheduler;
}
public static void setScheduler(@NotNull Scheduler scheduler) {
Schedulers.scheduler = scheduler;
}
}

52
futur-api/src/main/java/dev/tommyjs/futur/scheduler/SingleExecutorScheduler.java
View File

@@ -1,52 +0,0 @@
package dev.tommyjs.futur.scheduler;
import dev.tommyjs.futur.trace.ExecutorTrace;
import org.jetbrains.annotations.NotNull;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class SingleExecutorScheduler implements Scheduler {
private final ScheduledExecutorService service;
protected SingleExecutorScheduler(ScheduledExecutorService service) {
this.service = service;
}
@Override
public void runSync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
service.submit(wrapExceptions(task, trace));
}
@Override
public void runDelayedSync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
service.schedule(wrapExceptions(task, trace), delay, unit);
}
@Override
public void runRepeatingSync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
service.scheduleAtFixedRate(wrapExceptions(task, trace), 0L, interval, unit);
}
@Override
public void runAsync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
runSync(task, trace);
}
@Override
public void runDelayedAsync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
runDelayedSync(task, delay, unit, trace);
}
@Override
public void runRepeatingAsync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
runRepeatingSync(task, interval, unit, trace);
}
public static SingleExecutorScheduler create() {
return new SingleExecutorScheduler(Executors.newSingleThreadScheduledExecutor());
}
}

31
futur-api/src/main/java/dev/tommyjs/futur/trace/ExecutorTrace.java
View File

@@ -1,31 +0,0 @@
package dev.tommyjs.futur.trace;
import org.jetbrains.annotations.NotNull;
import java.util.Arrays;
import java.util.stream.Collectors;
public class ExecutorTrace {
private final @NotNull Class<?> clazz;
private final @NotNull StackTraceElement[] trace;
public ExecutorTrace(@NotNull Class<?> clazz, @NotNull StackTraceElement[] trace) {
this.clazz = clazz;
this.trace = trace;
}
public @NotNull Class<?> getClazz() {
return clazz;
}
public @NotNull StackTraceElement[] getTrace() {
return trace;
}
@Override
public String toString() {
return Arrays.stream(trace).map(StackTraceElement::toString).collect(Collectors.joining("\n"));
}
}

2
futur-reactive-streams/build.gradle.kts
View File

@@ -6,7 +6,7 @@ plugins {
}
group = "dev.tommyjs"
version = "1.0.1"
version = "2.0.0"
repositories {
mavenCentral()

6
futur-reactive-streams/src/main/java/dev/tommyjs/futur/reactivestreams/ReactiveTransformer.java
View File

@@ -1,13 +1,15 @@
package dev.tommyjs.futur.reactivestreams;
import dev.tommyjs.futur.promise.Promise;
import dev.tommyjs.futur.promise.PromiseFactory;
import dev.tommyjs.futur.promise.StaticPromiseFactory;
import org.jetbrains.annotations.NotNull;
import org.reactivestreams.Publisher;
public class ReactiveTransformer {
public static <T> @NotNull Promise<T> wrapPublisher(@NotNull Publisher<T> publisher) {
SingleAccumulatorSubscriber<T> subscriber = SingleAccumulatorSubscriber.create();
public static <T> @NotNull Promise<T> wrapPublisher(@NotNull Publisher<T> publisher, PromiseFactory factory) {
SingleAccumulatorSubscriber<T> subscriber = SingleAccumulatorSubscriber.create(factory);
publisher.subscribe(subscriber);
return subscriber.getPromise();
}

8
futur-reactive-streams/src/main/java/dev/tommyjs/futur/reactivestreams/SingleAccumulatorSubscriber.java
View File

@@ -1,6 +1,8 @@
package dev.tommyjs.futur.reactivestreams;
import dev.tommyjs.futur.promise.Promise;
import dev.tommyjs.futur.promise.PromiseFactory;
import dev.tommyjs.futur.promise.StaticPromiseFactory;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
@@ -40,8 +42,12 @@ public class SingleAccumulatorSubscriber<T> implements Subscriber<T> {
return new SingleAccumulatorSubscriber<>(promise);
}
public static <T> SingleAccumulatorSubscriber<T> create(PromiseFactory factory) {
return create(factory.unresolved());
}
public static <T> SingleAccumulatorSubscriber<T> create() {
return create(new Promise<>());
return create(StaticPromiseFactory.INSTANCE);
}
}

2
futur-reactor/build.gradle.kts
View File

@@ -4,7 +4,7 @@ plugins {
}
group = "dev.tommyjs"
version = "1.0.1"
version = "2.0.0"
repositories {
mavenCentral()

10
futur-reactor/src/main/java/dev/tommyjs/futur/reactor/ReactorTransformer.java
View File

@@ -1,6 +1,8 @@
package dev.tommyjs.futur.reactor;
import dev.tommyjs.futur.promise.Promise;
import dev.tommyjs.futur.promise.PromiseFactory;
import dev.tommyjs.futur.promise.StaticPromiseFactory;
import org.jetbrains.annotations.NotNull;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -11,14 +13,14 @@ import java.util.concurrent.atomic.AtomicReference;
public class ReactorTransformer {
public static <T> @NotNull Promise<T> wrapMono(@NotNull Mono<T> mono) {
Promise<T> promise = new Promise<>();
public static <T> @NotNull Promise<T> wrapMono(@NotNull Mono<T> mono, PromiseFactory factory) {
Promise<T> promise = factory.unresolved();
mono.doOnSuccess(promise::complete).doOnError(promise::completeExceptionally).subscribe();
return promise;
}
public static <T> @NotNull Promise<@NotNull List<T>> wrapFlux(@NotNull Flux<T> flux) {
Promise<List<T>> promise = new Promise<>();
public static <T> @NotNull Promise<@NotNull List<T>> wrapFlux(@NotNull Flux<T> flux, PromiseFactory factory) {
Promise<List<T>> promise = factory.unresolved();
AtomicReference<List<T>> out = new AtomicReference<>(new ArrayList<>());
flux.doOnNext(out.get()::add).subscribe();

42
futur-standalone/.gitignore vendored
View File

@@ -1,42 +0,0 @@
.gradle
build/
!gradle/wrapper/gradle-wrapper.jar
!**/src/main/**/build/
!**/src/test/**/build/
### IntelliJ IDEA ###
.idea/modules.xml
.idea/jarRepositories.xml
.idea/compiler.xml
.idea/libraries/
*.iws
*.iml
*.ipr
out/
!**/src/main/**/out/
!**/src/test/**/out/
### Eclipse ###
.apt_generated
.classpath
.factorypath
.project
.settings
.springBeans
.sts4-cache
bin/
!**/src/main/**/bin/
!**/src/test/**/bin/
### NetBeans ###
/nbproject/private/
/nbbuild/
/dist/
/nbdist/
/.nb-gradle/
### VS Code ###
.vscode/
### Mac OS ###
.DS_Store

34
futur-standalone/build.gradle.kts
View File

@@ -1,34 +0,0 @@
import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
plugins {
id("java")
id("com.github.johnrengelman.shadow") version "7.1.2"
}
group = "dev.tommyjs"
version = "1.0.1"
repositories {
mavenCentral()
}
dependencies {
implementation("org.jetbrains:annotations:24.1.0")
compileOnly(project(mapOf("path" to ":futur-api")))
testImplementation(platform("org.junit:junit-bom:5.9.1"))
testImplementation("org.junit.jupiter:junit-jupiter")
}
tasks {
build {
dependsOn(shadowJar)
}
withType<ShadowJar> {
exclude("META-INF/**")
}
}
tasks.test {
useJUnitPlatform()
}

61
futur-standalone/src/main/java/dev/tommyjs/futur/standalone/ExclusiveThreadPoolScheduler.java
View File

@@ -1,61 +0,0 @@
package dev.tommyjs.futur.standalone;
import dev.tommyjs.futur.scheduler.Scheduler;
import dev.tommyjs.futur.trace.ExecutorTrace;
import org.jetbrains.annotations.NotNull;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ExclusiveThreadPoolScheduler implements Scheduler {
private final ScheduledExecutorService executor;
protected ExclusiveThreadPoolScheduler(ScheduledExecutorService executor) {
this.executor = executor;
}
@Override
public void runSync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
throw new UnsupportedOperationException("Sync task invoked on asynchronous environment");
}
@Override
public void runDelayedSync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
throw new UnsupportedOperationException("Sync task invoked on asynchronous environment");
}
@Override
public void runRepeatingSync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
throw new UnsupportedOperationException("Sync task invoked on asynchronous environment");
}
@Override
public void runAsync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
executor.submit(wrapExceptions(task, trace));
}
@Override
public void runDelayedAsync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
executor.schedule(wrapExceptions(task, trace), delay, unit);
}
@Override
public void runRepeatingAsync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
executor.scheduleAtFixedRate(wrapExceptions(task, trace), 0L, interval, unit);
}
public @NotNull ScheduledExecutorService getExecutor() {
return executor;
}
public static ExclusiveThreadPoolScheduler create(ScheduledExecutorService executor) {
return new ExclusiveThreadPoolScheduler(executor);
}
public static ExclusiveThreadPoolScheduler create(int nThreads) {
return create(Executors.newScheduledThreadPool(nThreads));
}
}

63
futur-standalone/src/main/java/dev/tommyjs/futur/standalone/ThreadPoolScheduler.java
View File

@@ -1,63 +0,0 @@
package dev.tommyjs.futur.standalone;
import dev.tommyjs.futur.scheduler.Scheduler;
import dev.tommyjs.futur.trace.ExecutorTrace;
import org.jetbrains.annotations.NotNull;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ThreadPoolScheduler implements Scheduler {
private final ScheduledExecutorService syncExecutor;
private final ScheduledExecutorService asyncExecutor;
protected ThreadPoolScheduler(ScheduledExecutorService syncExecutor, ScheduledExecutorService asyncExecutor) {
this.syncExecutor = syncExecutor;
this.asyncExecutor = asyncExecutor;
}
@Override
public void runSync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
syncExecutor.submit(wrapExceptions(task, trace));
}
@Override
public void runDelayedSync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
syncExecutor.schedule(wrapExceptions(task, trace), delay, unit);
}
@Override
public void runRepeatingSync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
syncExecutor.scheduleAtFixedRate(wrapExceptions(task, trace), 0L, interval, unit);
}
@Override
public void runAsync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
asyncExecutor.submit(wrapExceptions(task, trace));
}
@Override
public void runDelayedAsync(@NotNull Runnable task, long delay, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
asyncExecutor.schedule(wrapExceptions(task, trace), delay, unit);
}
@Override
public void runRepeatingAsync(@NotNull Runnable task, long interval, @NotNull TimeUnit unit, @NotNull ExecutorTrace trace) {
asyncExecutor.scheduleAtFixedRate(wrapExceptions(task, trace), 0L, interval, unit);
}
public @NotNull ScheduledExecutorService getSyncExecutor() {
return syncExecutor;
}
public @NotNull ScheduledExecutorService getAsyncExecutor() {
return asyncExecutor;
}
public static ThreadPoolScheduler create(int nThreads) {
return new ThreadPoolScheduler(Executors.newSingleThreadScheduledExecutor(), Executors.newScheduledThreadPool(nThreads));
}
}