mirror of
https://github.com/tommyskeff/futur4j.git
synced 2026-01-18 07:16:45 +00:00
cleanup combine methods
This commit is contained in:
@@ -16,11 +16,11 @@ public class CompletionJoiner extends PromiseJoiner<Promise<?>, Void, Void, List
|
||||
public CompletionJoiner(
|
||||
@NotNull PromiseFactory factory,
|
||||
@NotNull Iterator<Promise<?>> promises,
|
||||
int expectedSize, boolean link
|
||||
int expectedSize
|
||||
) {
|
||||
super(factory);
|
||||
results = new ConcurrentResultArray<>(expectedSize);
|
||||
join(promises, link);
|
||||
join(promises);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -15,11 +15,11 @@ public class MappedResultJoiner<K, V> extends PromiseJoiner<Map.Entry<K, Promise
|
||||
public MappedResultJoiner(
|
||||
@NotNull PromiseFactory factory,
|
||||
@NotNull Iterator<Map.Entry<K, Promise<V>>> promises,
|
||||
int expectedSize, boolean link
|
||||
int expectedSize
|
||||
) {
|
||||
super(factory);
|
||||
this.results = new ConcurrentResultArray<>(expectedSize);
|
||||
join(promises, link);
|
||||
join(promises);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -10,49 +10,51 @@ import org.jetbrains.annotations.NotNull;
|
||||
import java.util.Iterator;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public abstract class PromiseJoiner<V, K, T, R> {
|
||||
public abstract class PromiseJoiner<T, Key, Value, Result> {
|
||||
|
||||
private final CompletablePromise<R> joined;
|
||||
private final CompletablePromise<Result> joined;
|
||||
|
||||
protected PromiseJoiner(@NotNull PromiseFactory factory) {
|
||||
this.joined = factory.unresolved();
|
||||
}
|
||||
|
||||
protected abstract K getChildKey(V value);
|
||||
protected abstract Key getChildKey(T value);
|
||||
|
||||
protected abstract @NotNull Promise<T> getChildPromise(V value);
|
||||
protected abstract @NotNull Promise<Value> getChildPromise(T value);
|
||||
|
||||
protected abstract void onChildComplete(int index, K key, @NotNull PromiseCompletion<T> completion);
|
||||
protected abstract void onChildComplete(int index, Key key, @NotNull PromiseCompletion<Value> completion);
|
||||
|
||||
protected abstract R getResult();
|
||||
protected abstract Result getResult();
|
||||
|
||||
protected void join(@NotNull Iterator<V> promises, boolean link) {
|
||||
protected void join(@NotNull Iterator<T> promises) {
|
||||
AtomicInteger count = new AtomicInteger();
|
||||
|
||||
int i = 0;
|
||||
do {
|
||||
V value = promises.next();
|
||||
Promise<T> p = getChildPromise(value);
|
||||
if (joined.isCompleted()) {
|
||||
promises.forEachRemaining(v -> getChildPromise(v).cancel());
|
||||
return;
|
||||
}
|
||||
|
||||
if (link) {
|
||||
T value = promises.next();
|
||||
Promise<Value> p = getChildPromise(value);
|
||||
if (!p.isCompleted()) {
|
||||
PromiseUtil.cancelOnComplete(joined, p);
|
||||
}
|
||||
|
||||
if (!joined.isCompleted()) {
|
||||
count.incrementAndGet();
|
||||
K key = getChildKey(value);
|
||||
int index = i++;
|
||||
count.incrementAndGet();
|
||||
Key key = getChildKey(value);
|
||||
int index = i++;
|
||||
|
||||
p.addAsyncListener(res -> {
|
||||
onChildComplete(index, key, res);
|
||||
if (res.isError()) {
|
||||
assert res.getException() != null;
|
||||
joined.completeExceptionally(res.getException());
|
||||
} else if (count.decrementAndGet() == -1) {
|
||||
joined.complete(getResult());
|
||||
}
|
||||
});
|
||||
}
|
||||
p.addAsyncListener(res -> {
|
||||
onChildComplete(index, key, res);
|
||||
if (res.isError()) {
|
||||
assert res.getException() != null;
|
||||
joined.completeExceptionally(res.getException());
|
||||
} else if (count.decrementAndGet() == -1) {
|
||||
joined.complete(getResult());
|
||||
}
|
||||
});
|
||||
} while (promises.hasNext());
|
||||
|
||||
if (count.decrementAndGet() == -1) {
|
||||
@@ -60,7 +62,7 @@ public abstract class PromiseJoiner<V, K, T, R> {
|
||||
}
|
||||
}
|
||||
|
||||
public @NotNull Promise<R> joined() {
|
||||
public @NotNull Promise<Result> joined() {
|
||||
return joined;
|
||||
}
|
||||
|
||||
|
||||
@@ -16,11 +16,11 @@ public class ResultJoiner<T> extends PromiseJoiner<Promise<T>, Void, T, List<T>>
|
||||
public ResultJoiner(
|
||||
@NotNull PromiseFactory factory,
|
||||
@NotNull Iterator<Promise<T>> promises,
|
||||
int expectedSize, boolean link
|
||||
int expectedSize
|
||||
) {
|
||||
super(factory);
|
||||
this.results = new ConcurrentResultArray<>(expectedSize);
|
||||
join(promises, link);
|
||||
join(promises);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -9,9 +9,9 @@ import java.util.Iterator;
|
||||
|
||||
public class VoidJoiner extends PromiseJoiner<Promise<?>, Void, Void, Void> {
|
||||
|
||||
public VoidJoiner(@NotNull PromiseFactory factory, @NotNull Iterator<Promise<?>> promises, boolean link) {
|
||||
public VoidJoiner(@NotNull PromiseFactory factory, @NotNull Iterator<Promise<?>> promises) {
|
||||
super(factory);
|
||||
join(promises, link);
|
||||
join(promises);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
Reference in New Issue
Block a user