Skip to content

Commit e6ac3e2

Browse files
authored
Add 6 Flow operators migrated from ox (#273)
## Summary - Migrate 6 Flow operators from ox to jox: `runToSet`, `runToMap`, `onErrorRecover`, `onErrorComplete`, `mapWithResource`, `mapWithCloseableResource` - Part of #266
1 parent 0655022 commit e6ac3e2

6 files changed

Lines changed: 703 additions & 0 deletions

File tree

flows/src/main/java/com/softwaremill/jox/flows/Flow.java

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,28 @@ public List<T> runToList() throws Exception {
8282
return result;
8383
}
8484

85+
/**
86+
* Accumulates all elements emitted by this flow into a set. Blocks until the flow completes.
87+
*/
88+
public Set<T> runToSet() throws Exception {
89+
Set<T> result = new HashSet<>();
90+
runForeach(result::add);
91+
return result;
92+
}
93+
94+
/**
95+
* Accumulates all elements emitted by this flow into a map, using the provided key and value
96+
* extractors. Blocks until the flow completes.
97+
*
98+
* <p>If duplicate keys are encountered, the last value wins (same as {@link Map#put}).
99+
*/
100+
public <K, V> Map<K, V> runToMap(Function<T, K> keyExtractor, Function<T, V> valueExtractor)
101+
throws Exception {
102+
Map<K, V> result = new HashMap<>();
103+
runForeach(t -> result.put(keyExtractor.apply(t), valueExtractor.apply(t)));
104+
return result;
105+
}
106+
85107
/**
86108
* The flow is run in the background, and each emitted element is sent to a newly created
87109
* channel, which is then returned as the result of this method.
@@ -442,6 +464,63 @@ public <S, U> Flow<U> mapStatefulConcat(
442464
return mapStatefulConcat(initializeState, f, _ -> Optional.empty());
443465
}
444466

467+
/**
468+
* Maps each element using a resource that is created when the flow starts and closed when the
469+
* flow completes (either successfully or with an error).
470+
*
471+
* <p>The {@code close} function may return an optional final element to emit after the flow
472+
* completes successfully. If the flow fails, any value returned by {@code close} is discarded.
473+
* If {@code close} throws and the flow also failed, the close exception is added as suppressed.
474+
*
475+
* @param create creates the resource (called once when the flow starts)
476+
* @param close closes the resource, optionally returning a final element
477+
* @param f maps each element using the resource
478+
*/
479+
public <R, U> Flow<U> mapWithResource(
480+
Callable<R> create,
481+
ThrowingFunction<R, Optional<U>> close,
482+
ThrowingBiFunction<R, T, U> f) {
483+
return usingEmit(
484+
emit -> {
485+
R resource = create.call();
486+
Throwable[] error = {null};
487+
try {
488+
last.run(t -> emit.apply(f.apply(resource, t)));
489+
} catch (Throwable e) {
490+
error[0] = e;
491+
throw e;
492+
} finally {
493+
try {
494+
Optional<U> finalElement = close.apply(resource);
495+
if (error[0] == null && finalElement.isPresent()) {
496+
emit.apply(finalElement.get());
497+
}
498+
} catch (Throwable e) {
499+
if (error[0] != null) error[0].addSuppressed(e);
500+
else throw e;
501+
}
502+
}
503+
});
504+
}
505+
506+
/**
507+
* A variant of {@link #mapWithResource} for {@link AutoCloseable} resources. The resource is
508+
* closed using {@link AutoCloseable#close()} and no final element is emitted.
509+
*
510+
* @param create creates the {@link AutoCloseable} resource
511+
* @param f maps each element using the resource
512+
*/
513+
public <R extends AutoCloseable, U> Flow<U> mapWithCloseableResource(
514+
Callable<R> create, ThrowingBiFunction<R, T, U> f) {
515+
return mapWithResource(
516+
create,
517+
r -> {
518+
r.close();
519+
return Optional.empty();
520+
},
521+
f);
522+
}
523+
445524
/**
446525
* Emits only those elements emitted by this flow, for which `filteringPredicate` returns
447526
* `true`.
@@ -1206,6 +1285,48 @@ public <U extends T> Flow<U> recover(ThrowingFunction<Throwable, Optional<U>> pf
12061285
}));
12071286
}
12081287

1288+
/**
1289+
* Recovers from upstream errors by applying {@code f} to produce a replacement value. A variant
1290+
* of {@link #recover} that always handles the exception.
1291+
*/
1292+
public <U extends T> Flow<U> onErrorRecover(ThrowingFunction<Throwable, U> f) {
1293+
return recover(e -> Optional.of(f.apply(e)));
1294+
}
1295+
1296+
/**
1297+
* Completes the flow when any {@link Exception} is thrown by the upstream, discarding the
1298+
* exception. Downstream failures are not caught.
1299+
*/
1300+
public Flow<T> onErrorComplete() {
1301+
return onErrorComplete(e -> e instanceof Exception);
1302+
}
1303+
1304+
/**
1305+
* Completes the flow when an upstream error matches the provided predicate, discarding the
1306+
* exception. Downstream failures are not caught.
1307+
*/
1308+
public Flow<T> onErrorComplete(Predicate<Throwable> shouldComplete) {
1309+
return usingEmit(
1310+
emit -> {
1311+
Throwable[] downstreamFailure = {null};
1312+
FlowEmit<T> guardedEmit =
1313+
t -> {
1314+
try {
1315+
emit.apply(t);
1316+
} catch (Throwable e) {
1317+
downstreamFailure[0] = e;
1318+
throw e;
1319+
}
1320+
};
1321+
try {
1322+
last.run(guardedEmit);
1323+
} catch (Throwable e) {
1324+
if (downstreamFailure[0] == e || !shouldComplete.test(e)) throw e;
1325+
// else: upstream exception matching predicate, complete silently
1326+
}
1327+
});
1328+
}
1329+
12091330
/**
12101331
* Intersperses elements emitted by this flow with `inject` elements. The `inject` element is
12111332
* emitted between each pair of elements.

0 commit comments

Comments
 (0)