RxJava: Cách nhóm số lượng lớn các mục tùy ý bằng chức năng ranh giới của riêng tôi

Question

Tôi có thể quan sát phát ra chuỗi và tôi muốn nhóm chúng theo ký tự đầu tiên. Thật dễ dàng để làm điều đó với groupBy như thế:

Observable<String> rows = Observable.just("aa", "ab", "ac", "bb", "bc", "cc"); 

Observable<List<String>> groupedRows = rows.groupBy(new Func1<String, Character>() { 
    public Character call(String row) { 
    return row.charAt(0); 
    } 
}).flatMap(new Func1<GroupedObservable<Character, String>, Observable<List<String>>>() { 
    public Observable<List<String>> call(GroupedObservable<Character, String> group) { 
    return group.toList(); 
    } 
}); 

groupedRows.toBlocking().forEach(new Action1<List<String>>() { 
    public void call(List<String> group) { 
    System.out.println(group); 
    } 
}); 

// Output: 
// [aa, ab, ac] 
// [bb, bc] 
// [cc] 

Nhưng nó không tốt cho mục đích của tôi, bởi vì groupBy chỉ hoàn thành từng nhóm khi nguồn phát ra quan sát được onComplete. Vì vậy, nếu tôi có nhiều hàng, chúng sẽ được thu thập hoàn toàn trong bộ nhớ và chỉ ở hàng cuối cùng "đỏ bừng" và được ghi vào đầu ra.

Tôi cần một cái gì đó như toán tử buffer, nhưng với chức năng của riêng tôi biểu thị ranh giới của mỗi nhóm. Tôi thực hiện nó như thế (biết rằng hàng luôn ra lệnh alphabeticaly):

Observable<String> rows = Observable.just("aa", "ab", "ac", "bb", "bc", "cc"); 

ConnectableObservable<String> connectableRows = rows.publish(); 

Observable<String> boundarySelector = connectableRows.filter(new Func1<String, Boolean>() { 
    private char lastChar = 0; 

    public Boolean call(String row) { 
    char currentChar = row.charAt(0); 
    boolean isNewGroup = lastChar != 0 && (currentChar != lastChar); 
    lastChar = currentChar; 
    return isNewGroup; 
    } 
}); 

Observable<List<String>> groupedRows = connectableRows.buffer(boundarySelector); 

connectableRows.connect(); 

groupedRows.toBlocking().forEach(new Action1<List<String>>() { 
    public void call(List<String> group) { 
    System.out.println(group); 
    } 
}); 

// Output: 
// [] 
// [] 
// [] 

Nó không làm việc, bởi vì boundarySelector được "ăn" các hàng, và tôi nghĩ rằng đó là lạ vì tôi đặc biệt sử dụng ConnectableObservable để biểu thị rằng Tôi cần hai người đăng ký (boundarySelector và groupedRows) trước khi bắt đầu phát ra rows.

Một cách tò mò nếu tôi trì hoãn rows sau 1 giây, sau đó mã này hoạt động.

Vì vậy, câu hỏi đặt ra là: làm thế nào để tôi nhóm số lượng tùy ý các hàng bằng cách sử dụng chức năng ranh giới của riêng tôi?

Answer 1

Observable<Integer> source = Observable.range(0, 100); 

source 
.groupBy(k -> k/10) 
.publish(groups -> groups 
     .map(g -> Pair.of(g.getKey(), g.takeUntil(groups))) 
     .flatMap(kv -> 
      kv.second 
      .doOnNext(v -> System.out.println(kv.first + " value " + v)) 
      .doOnCompleted(() -> System.out.println(kv.first + " done")) 
     )) 
.subscribe() 
; 

Answer 2

Tìm thấy một cách để làm điều đó bằng buffer:

Observable<String> rows = Observable.just("aa", "ab", "ac", "bb", "bc", "cc"); 

ConnectableObservable<String> connectableRows = rows.publish(); 

Observable<String> boundarySelector = connectableRows.filter(new Func1<String, Boolean>() { 
    private char lastChar = 0; 

    public Boolean call(String row) { 
     char currentChar = row.charAt(0); 
     boolean isNewGroup = lastChar != 0 && (currentChar != lastChar); 
     lastChar = currentChar; 
     return isNewGroup; 
    } 
}); 

Observable<List<String>> groupedRows = connectableRows 
     .refCount() 
     .buffer(boundarySelector); 

groupedRows.toBlocking().forEach(new Action1<List<String>>() { 
    public void call(List<String> group) { 
     System.out.println(group); 
    } 
}); 

// Output: 
// [aa, ab, ac] 
// [bb, bc] 
// [cc]