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

make other modules compatible with 1.2.0

Browse Source
This commit is contained in:
tommyskeff
2023-12-22 17:59:04 +00:00
parent 173e34810c
commit 6fc6c9f236
9 changed files with 55 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
futur-api/src/test/java/dev/tommyjs/test/Test.java
View File

@@ -1,35 +0,0 @@
package dev.tommyjs.test;
import dev.tommyjs.futur.promise.PooledPromiseFactory;
import dev.tommyjs.futur.promise.Promise;
import dev.tommyjs.futur.promise.PromiseFactory;
import dev.tommyjs.futur.scheduler.Scheduler;
import dev.tommyjs.futur.scheduler.SingleExecutorScheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class Test {
public static void main(String[] args) throws InterruptedException {
Scheduler scheduler = new SingleExecutorScheduler(Executors.newScheduledThreadPool(4));
Logger logger = LoggerFactory.getLogger(Test.class);
PromiseFactory factory = new PooledPromiseFactory(scheduler, logger);
Thread.sleep(2000);
Promise.start(factory)
.thenRunAsync(() -> {
System.out.println("HI");
})
.thenApplyDelayedAsync(_v -> {
return "ABC";
}, 1L, TimeUnit.SECONDS)
.thenConsumeSync(t -> {
System.out.println(t);
});
}
}

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

@@ -1,15 +1,22 @@
package dev.tommyjs.futur.reactivestreams;
import dev.tommyjs.futur.promise.AbstractPromise;
import dev.tommyjs.futur.promise.Promise;
import dev.tommyjs.futur.promise.PromiseFactory;
import dev.tommyjs.futur.promise.UnpooledPromiseFactory;
import org.jetbrains.annotations.NotNull;
import org.reactivestreams.Publisher;
public class ReactiveTransformer {
public static <T> @NotNull AbstractPromise<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();
}
public static <T> @NotNull Promise<T> wrapPublisher(@NotNull Publisher<T> publisher) {
return wrapPublisher(publisher, UnpooledPromiseFactory.INSTANCE);
}
}

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

@@ -1,14 +1,17 @@
package dev.tommyjs.futur.reactivestreams;
import dev.tommyjs.futur.promise.AbstractPromise;
import dev.tommyjs.futur.promise.Promise;
import dev.tommyjs.futur.promise.PromiseFactory;
import dev.tommyjs.futur.promise.UnpooledPromiseFactory;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
public class SingleAccumulatorSubscriber<T> implements Subscriber<T> {
private final AbstractPromise<T> promise;
private final Promise<T> promise;
public SingleAccumulatorSubscriber(AbstractPromise<T> promise) {
public SingleAccumulatorSubscriber(Promise<T> promise) {
this.promise = promise;
}
@@ -32,16 +35,20 @@ public class SingleAccumulatorSubscriber<T> implements Subscriber<T> {
// ignore
}
public AbstractPromise<T> getPromise() {
public Promise<T> getPromise() {
return promise;
}
public static <T> SingleAccumulatorSubscriber<T> create(AbstractPromise<T> promise) {
public static <T> SingleAccumulatorSubscriber<T> create(Promise<T> promise) {
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 AbstractPromise<>());
return create(UnpooledPromiseFactory.INSTANCE);
}
}

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

@@ -1,6 +1,9 @@
package dev.tommyjs.futur.reactor;
import dev.tommyjs.futur.promise.AbstractPromise;
import dev.tommyjs.futur.promise.PromiseFactory;
import dev.tommyjs.futur.promise.Promise;
import dev.tommyjs.futur.promise.UnpooledPromiseFactory;
import org.jetbrains.annotations.NotNull;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -11,14 +14,18 @@ import java.util.concurrent.atomic.AtomicReference;
public class ReactorTransformer {
public static <T> @NotNull AbstractPromise<T> wrapMono(@NotNull Mono<T> mono) {
AbstractPromise<T> promise = new AbstractPromise<>();
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 AbstractPromise<@NotNull List<T>> wrapFlux(@NotNull Flux<T> flux) {
AbstractPromise<List<T>> promise = new AbstractPromise<>();
public static <T> @NotNull Promise<T> wrapMono(@NotNull Mono<T> mono) {
return wrapMono(mono, UnpooledPromiseFactory.INSTANCE);
}
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();
@@ -28,4 +35,8 @@ public class ReactorTransformer {
return promise;
}
public static <T> @NotNull Promise<@NotNull List<T>> wrapFlux(@NotNull Flux<T> flux) {
return wrapFlux(flux, UnpooledPromiseFactory.INSTANCE);
}
}

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

@@ -14,6 +14,7 @@ repositories {
dependencies {
implementation("org.jetbrains:annotations:24.1.0")
implementation("org.slf4j:slf4j-api:2.0.9")
compileOnly(project(mapOf("path" to ":futur-api")))
testImplementation(platform("org.junit:junit-bom:5.9.1"))
testImplementation("org.junit.jupiter:junit-jupiter")

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

@@ -33,17 +33,17 @@ public class ExclusiveThreadPoolScheduler implements Scheduler {
@Override
public void runAsync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
executor.submit(wrapExceptions(task, trace));
executor.submit(Scheduler.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);
executor.schedule(Scheduler.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);
executor.scheduleAtFixedRate(Scheduler.wrapExceptions(task, trace), 0L, interval, unit);
}
public @NotNull ScheduledExecutorService getExecutor() {

4
futur-api/src/main/java/dev/tommyjs/futur/promise/PooledPromise.java → futur-standalone/src/main/java/dev/tommyjs/futur/standalone/PooledPromise.java
View File

@@ -1,5 +1,7 @@
package dev.tommyjs.futur.promise;
package dev.tommyjs.futur.standalone;
import dev.tommyjs.futur.promise.AbstractPromise;
import dev.tommyjs.futur.promise.PromiseFactory;
import dev.tommyjs.futur.scheduler.Scheduler;
import org.slf4j.Logger;

8
futur-api/src/main/java/dev/tommyjs/futur/promise/PooledPromiseFactory.java → futur-standalone/src/main/java/dev/tommyjs/futur/standalone/PooledPromiseFactory.java
View File

@@ -1,5 +1,9 @@
package dev.tommyjs.futur.promise;
package dev.tommyjs.futur.standalone;
import dev.tommyjs.futur.promise.AbstractPromise;
import dev.tommyjs.futur.promise.Promise;
import dev.tommyjs.futur.promise.PromiseCompletion;
import dev.tommyjs.futur.promise.PromiseFactory;
import dev.tommyjs.futur.scheduler.Scheduler;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
@@ -17,7 +21,7 @@ public class PooledPromiseFactory implements PromiseFactory {
@Override
public @NotNull <T> Promise<T> resolve(T value) {
AbstractPromise<T> promise = new PooledPromise<>(scheduler, logger, this);
promise.setCompletion(new PromiseCompletion<>(value));
promise.complete(value);
return promise;
}

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

@@ -20,32 +20,32 @@ public class ThreadPoolScheduler implements Scheduler {
@Override
public void runSync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
syncExecutor.submit(wrapExceptions(task, trace));
syncExecutor.submit(Scheduler.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);
syncExecutor.schedule(Scheduler.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);
syncExecutor.scheduleAtFixedRate(Scheduler.wrapExceptions(task, trace), 0L, interval, unit);
}
@Override
public void runAsync(@NotNull Runnable task, @NotNull ExecutorTrace trace) {
asyncExecutor.submit(wrapExceptions(task, trace));
asyncExecutor.submit(Scheduler.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);
asyncExecutor.schedule(Scheduler.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);
asyncExecutor.scheduleAtFixedRate(Scheduler.wrapExceptions(task, trace), 0L, interval, unit);
}
public @NotNull ScheduledExecutorService getSyncExecutor() {