RxJava 2.X में एक बूलियन गेट के आधार पर एक अवलोकन योग्य को फिर से शुरू करें और फिर से शुरू करें?

Question

मान लें कि मेरे पास एक प्रोसेसर है जो एक बटन दबाए जाने पर एक बूलियन मान उत्सर्जित करता है, इसे टॉगल के रूप में सोचें।

boolean gateValue = true; 
    PublishProcessor<Boolean> gate = PublishProcessor.create(); 
    view.onButtonClicked() 
      .subscribe(new Action1<Void>() { 
       @Override 
       public void call(final Void aVoid) { 
        gate.onNext(gateValue = !gateValue); 
       } 
      })); 

मैं करना चाहते हैं क्या रोक सकते हैं और एक नमूदार अनुक्रम फिर से शुरू करने, उत्सर्जित मूल्यों बफरिंग जबकि रुका हुआ फाटक के मूल्य का उपयोग है।

मैंने इसमें बहुत कुछ पढ़ा है और जबकि यह अन्य भाषाओं के लिए प्रतिक्रियाशील एक्सटेंशन में संभव लगता है, तो आरएक्सजेवा इसका समर्थन नहीं कर रहा है।

यहां एक उदाहरण है जो मैं प्राप्त करना चाहता हूं, यह हर सेकेंड में वृद्धिशील मूल्य को आउटपुट करता है। जब मैं बटन दबाएँ जब तक मैं इसे फिर से प्रेस उत्पादन बंद करना चाहते हैं जो चाहिए उत्पादन हर आइटम दो बटन प्रेस के बीच उत्सर्जित:

Flowable.interval(1, TimeUnit.SECONDS) 
       .bufferWhile(gate) 
       .flatMapIterable(longs -> longs) 
       .subscribe(new Consumer<Long>() { 
        @Override 
        public void accept(final Long aLong) throws Exception { 
         view.displayTime(aLong); 
        } 
       }); 

किसी को भी एक तरह से कुछ इस तरह प्राप्त करने के लिए पता है?

संपादित मैं कैसे प्राप्त करने के लिए इस https://medium.com/@scottalancooper/pausing-and-resuming-a-stream-in-rxjava-988a0977b771#.gj7fsi1xk

Answer 1

अब RxJava2 एक्सटेंशन लाइब्रेरी में एक ऑपरेटर valve() है जो अनुरोधित व्यवहार करता है।

Answer 2

बस रेडीमेड Observable.window ऑपरेटर है कि एक Observable<T> पैरामीटर लेता है का उपयोग पर एक ब्लॉग पोस्ट में लिखा है।