यह संभव समय

Question

के अलावा कुछ पर Observable.Buffer को मैं कैसे rx में Observable.Buffer का उपयोग करने पर उदाहरण के लिए तलाश कर दिया गया है, लेकिन बायलर प्लेट समय बफ़र सामान की तुलना में अधिक पर्याप्त कुछ भी नहीं मिल सकता है।

वहाँ एक "bufferClosingSelector" निर्दिष्ट करने के लिए एक अधिभार होने लगते है, लेकिन मैं इसे चारों ओर मेरे मन लपेटो नहीं कर सकते।

मुझे क्या करना कोशिश कर रहा हूँ एक दृश्य उस समय तक या एक "संचय" द्वारा बफ़र्स पैदा करते हैं। एक अनुरोध धारा पर विचार करें जहां प्रत्येक अनुरोध के लिए कुछ प्रकार का वजन होता है और मैं एक समय में एक्स से अधिक भारित वजन को संसाधित नहीं करना चाहता, या यदि पर्याप्त मात्रा में जमा नहीं हुआ है तो बस मुझे अंतिम समय सीमा में क्या आया है (नियमित बफर कार्यक्षमता)

Answer 1

bufferClosingSelector एक पर्यवेक्षक प्राप्त करने के लिए हर बार एक समारोह कहा जाता है जो बफर को बंद होने की अपेक्षा होने पर एक मूल्य उत्पन्न करेगा।

उदाहरण के लिए

, नियमित रूप से Buffer(time) अधिभार की तरह

source.Buffer(() => Observable.Timer(TimeSpan.FromSeconds(1))) काम करता है।

आप एक अनुक्रम वजन करना चाहते हैं, तो आप अनुक्रम पर Scan लागू कर सकते हैं और फिर अपनी समेकित स्थिति पर निर्णय ले सकते हैं।

:

जैसे, source.Scan((a,c) => a + c).SkipWhile(a => a < 100) आप एक दृश्य जो जब स्रोत अनुक्रम आप Amb उपयोग कर सकते हैं इन दोनों समापन की स्थिति को देखने के लिए जो दौड़ के लिए और अधिक से अधिक 100

अप करने के लिए जोड़ा गया है एक मूल्य का उत्पादन पहले से प्रतिक्रिया करता है देता है

 .Buffer(() => Observable.Amb 
        (
          Observable.Timer(TimeSpan.FromSeconds(1)), 
          source.Scan((a,c) => a + c).SkipWhile(a => a < 100) 
        ) 
       ) 

आप जिसके लिए बफर उस बिंदु पर बंद किया जा करने के लिए कोई मूल्य पैदा करता है combinators के किसी भी श्रृंखला का उपयोग कर सकते हैं।

नोट: समापन चयनकर्ता को दिया कोई फर्क नहीं पड़ता मूल्य - यह सूचना है कि मायने रखती है। इसलिए Amb के साथ विभिन्न प्रकार के स्रोतों को गठबंधन करने के लिए इसे बस System.Reactive.Unit पर बदलें।

Observable.Amb(stream1.Select(_ => new Unit()), stream2.Select(_ => new Unit())