Skip to content

Instantly share code, notes, and snippets.

@torgeir
Last active April 13, 2018 01:48
Show Gist options
  • Save torgeir/bdfffbdfdc8ed98056b8 to your computer and use it in GitHub Desktop.
Save torgeir/bdfffbdfdc8ed98056b8 to your computer and use it in GitHub Desktop.
Combining observable results of multiple async http requests with rxjava's Observable.zip.
public class Statistics {
public static void main(String[] args) {
List<Observable<ObservableHttpResponse>> observableRequests = Arrays.asList(
Http.getAsync("http://localhost:3001/stream"),
Http.getAsync("http://localhost:3002/stream"),
Http.getAsync("http://localhost:3003/stream"),
Http.getAsync("http://localhost:3004/stream"));
List<Observable<Stats>> observableStats = observableRequests.stream()
.map(observableRequest ->
observableRequest.flatMap(response ->
response.getContent()
.map(new EventStreamJsonMapper<>(Stats.class))))
.collect(toList());
Observable<List<Stats>> joinedObservables = Observable.zip(
observableStats.get(0),
observableStats.get(1),
observableStats.get(2),
observableStats.get(3),
Arrays::asList);
// This does not work, as FuncN accepts (Object...) https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/functions/FuncN.java#L19
// Observable<List<Stats>> joinedObservables = Observable.zip(observableStats, Arrays::asList);
joinedObservables
.take(10)
.subscribe(
(List<Stats> statslist) -> {
System.out.println(statslist);
double average = statslist.stream()
.mapToInt(stats -> stats.ongoingRequests)
.average()
.getAsDouble();
System.out.println("avg: " + average);
},
System.err::println,
Http::shutdown);
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment