@Test

public void testZipObservableOfObservables() {

EventStream.getEventStream("HTTP-ClusterB", 20)

.groupBy(new Func1<Event, String>() {

@Override

public String call(Event e) {

return e.instanceId;

}

// now we have streams of cluster+instanceId

}).flatMap(new Func1<GroupedObservable<String, Event>, Observable<Map<String, String>>>() {

@Override

public Observable<Map<String, String>> call(final GroupedObservable<String, Event> ge) {

return ge.scan(new HashMap<String, String>(), new Func2<Map<String, String>, Event, Map<String, String>>() {

@Override

public Map<String, String> call(Map<String, String> accum, Event perInstanceEvent) {

accum.put("instance", ge.getKey());

return accum;

}

});

}

})

.take(10)

.toBlockingObservable().forEach(new Action1<Map<String, String>>() {

@Override

public void call(Map<String, String> v) {

System.out.println(v);

}

});

System.out.println("**** finished");

}

/**

@Test

public void testCovarianceOfZip() {

Observable<HorrorMovie> horrors = Observable.from(new HorrorMovie());

Observable<CoolRating> ratings = Observable.from(new CoolRating());

Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);

Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);

Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).toBlockingObservable().forEach(extendedAction);

Observable.<Media, Rating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);

Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).toBlockingObservable().forEach(action);

Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine);

}

Func2<Media, Rating, ExtendedResult> combine = new Func2<Media, Rating, ExtendedResult>() {

@Override

public ExtendedResult call(Media m, Rating r) {

return new ExtendedResult();

}

};

Action1<Result> action = new Action1<Result>() {

@Override

public void call(Result t1) {

System.out.println("Result: " + t1);

}

};

Action1<ExtendedResult> extendedAction = new Action1<ExtendedResult>() {

@Override

public void call(ExtendedResult t1) {

System.out.println("Result: " + t1);

}

};