rxjavaで境界線が開いていて、clossingしているバッファ

Question

rxjavaで境界線が開いていて、clossingしているバッファで作業しようとしていますが、うまく動作しません。 0,1,2,3,0,1,3,0,3私は{0,1,2,3}、{0,1}で終わりたいと思っています。 、3}、{0,3}である。

これは私がこれまでしているコードです：

PublishSubject openning = PublishSubject.create(); 
     openning.doOnNext(new Consumer() { 
      @Override 
      public void accept(@NonNull Object o) throws Exception { 
       if(o.equals("0")) 
        openning.onNext(o); 
      } 
     }); 

     Observable<String> observableA = Observable.interval(1, TimeUnit.SECONDS).map(value -> String.valueOf(value % 10)); 

     observableA.subscribe(openning); 
     // TODO: Buffer by boundary 
     observableA = observableA.buffer(openning, new Function<String, Observable<List<String>>>() { 
      @Override 
      public Observable<List<String>> apply(@NonNull String o) throws Exception { 
       list.add(o); 
       if (o.equals("0")) { 
        return Observable.just(list); 
       } else { 
        list.add(o); 
        sb.append(o); 
        return Observable.never(); 
       } 
      } 
     }, new Callable() { 
      @Override 
      public Object call() throws Exception { 
       return list; 
      } 
     }); 

すべてのヘルプは、あなたはバッファが3に遭遇した後に再起動する必要があることを意味している場合、のextensions projectでbufferUntilオペレータを見

Answer 1

を理解されるであろうRxJava 1.1 2.

compile "com.github.akarnokd:rxjava2-extensions:0.17.5" 

Flowable.just("1", "2", "#", "3", "#", "4", "#") 
.compose(FlowableTransformers.bufferUntil(v -> "#".equals(v))) 
.test() 
.assertResult(
    Arrays.asList("1", "2", "#"), 
    Arrays.asList("3", "#"), 
    Arrays.asList("4", "#") 
); 

は、Transformers.toListUntil()とrxjava-extrasがあります。