java - RxJava 中的 n 元笛卡尔积

Question

现在我拿着一个Observable<Observable<Integer> , 我怎样才能把它转移到 Observable<int[]>包含 n 元笛卡尔积？

例如:

Observable<Observable<Integer> ob = Observable.just(
  Observable.just(0,1),
  Observable.just(2,3),
  Observable.just(4,5)
  );
ob...... ->   (0,2,4), (0,3,4), (0,2,5), (0,3,5), (1,2,4), (1,3,4), (1,2,5), (1,3,5)

Answer 1

首先，您需要固定数量的输入Observable。其次，不需要阻塞，但可能需要缓存，因为第二个、第三个等 Observable 需要多次使用。

import java.util.*;

import io.reactivex.Observable;

public class Cartesian {

    static Observable<int[]> cartesian(Observable<Observable<Integer>> sources) {
        return sources.toList().flatMapObservable(list -> cartesian(list));
    }

    static Observable<int[]> cartesian(List<Observable<Integer>> sources) {
        if (sources.size() == 0) {
            return Observable.<int[]>empty();
        }
        Observable<int[]> main = sources.get(0).map(v -> new int[] { v });

        for (int i = 1; i < sources.size(); i++) {
            int j = i;
            Observable<Integer> o = sources.get(i).cache();
            main = main.flatMap(v -> {
                return o.map(w -> {
                    int[] arr = Arrays.copyOf(v, j + 1);
                    arr[j] = w;
                    return arr;
                });
            });
        }

        return main;
    }

    public static void main(String[] args) {
        cartesian(Observable.just(
            Observable.just(0, 1), 
            Observable.just(2, 3), 
            Observable.just(4, 5)
        ))
        .subscribe(v -> System.out.println(Arrays.toString(v)));
    }
}