थ्रॉटल Rx.Observable मूल्यों को छोड़ दिए बिना

Question

Throttle विधि एक अवलोकन अनुक्रम से मूल्य छोड़ देता है अगर अन्य बहुत जल्दी अनुसरण करते हैं। लेकिन मुझे बस देरी करने के लिए एक विधि की जरूरत है। यही है, मुझे छोड़ने के बिना आइटमों के बीच न्यूनतम देरी सेट करने की आवश्यकता है।

प्रैक्टिकल उदाहरण: एक वेब सेवा है जो अनुरोधों को दूसरे से एक बार तेज नहीं कर सकती है; एक ऐसा उपयोगकर्ता है जो अनुरोध, एकल या बैचों में जोड़ सकता है। आरएक्स के बिना, मैं एक सूची और एक टाइमर बनाउंगा। जब उपयोगकर्ता अनुरोध जोड़ते हैं, तो मैं उन्हें सूची में जोड़ दूंगा। टाइमर घटना में, मैं गीलेर की जांच करूंगा कि सूची खाली है। यदि ऐसा नहीं है, तो मैं एक अनुरोध भेजूंगा और संबंधित आइटम को हटा दूंगा। ताले और सभी चीजों के साथ। अब, आरएक्स के साथ, मैं Subject बना सकता हूं, जब उपयोगकर्ता अनुरोध जोड़ते हैं तो आइटम जोड़ें। लेकिन मुझे यह सुनिश्चित करने का एक तरीका चाहिए कि देरी लागू करके वेब सेवा बाढ़ न हो।

मैं आरएक्स के लिए नया हूं, तो शायद मुझे कुछ स्पष्ट याद आ रही है।

Answer 1

वहाँ एक काफी आसान तरीका क्या आप एक EventLoopScheduler का उपयोग कर करना चाहते हैं नहीं है।

मैंने एक अवलोकन के साथ शुरुआत की जो हर 0 से 3 सेकंड में यादृच्छिक रूप से मूल्यों का उत्पादन करेगी।

var rnd = new Random(); 

var xs = 
    Observable 
     .Generate(
      0, 
      x => x < 20, 
      x => x + 1, 
      x => x, 
      x => TimeSpan.FromSeconds(rnd.NextDouble() * 3.0)); 

अब

, इस उत्पादन मूल्यों बनाने के लिए तुरंत जब तक अंतिम मान के भीतर दूसरी पहले मैं ऐसा किया था:

var ys = 
    Observable.Create<int>(o => 
    { 
     var els = new EventLoopScheduler(); 
     return xs 
      .ObserveOn(els) 
      .Do(x => els.Schedule(() => Thread.Sleep(1000))) 
      .Subscribe(o); 
    }); 

यह प्रभावी रूप से EventLoopScheduler पर स्रोत का अवलोकन करता है और फिर कहते हैं के लिए सोने के लिए प्रत्येक OnNext के बाद 1 सेकंड ताकि यह जागने के बाद ही अगले OnNext शुरू कर सके।

मैं परीक्षण किया है कि यह इस कोड के साथ काम किया:

ys 
    .Timestamp() 
    .Select(x => x.Timestamp.Second + (double)x.Timestamp.Millisecond/1000.0) 
    .Subscribe(x => Console.WriteLine(x)); 

मुझे आशा है कि इस मदद करता है।

Answer 2

अवरुद्ध कतार से लेने के लिए एक अवलोकन करने योग्य टाइमर का उपयोग करने के बारे में कैसे? नीचे दिए गए कोड अपरीक्षित है, लेकिन आप मुझे क्या मतलब है की एक विचार देना चाहिए ...

//assuming somewhere there is 
BlockingCollection<MyWebServiceRequestData> workQueue = ... 

Observable 
    .Timer(new TimeSpan(0,0,1), new EventLoopScheduler()) 
    .Do(i => myWebService.Send(workQueue.Take())); 

// Then just add items to the queue using workQueue.Add(...) 

Answer 3

एक बफर का उपयोग करने पर विचार करें।

http://msdn.microsoft.com/en-us/library/hh212130(v=vs.103).aspx

नाम इंगित करता है, यह "गिर" उनमें से किसी के बिना ईवेंट के प्रत्यक्ष धारा में तत्वों बफ़र्स।

जी

Answer 4

कैसे एक सरल विस्तार विधि के बारे में:

public static IObservable<T> StepInterval<T>(this IObservable<T> source, TimeSpan minDelay) 
{ 
    return source.Select(x => 
     Observable.Empty<T>() 
      .Delay(minDelay) 
      .StartWith(x) 
    ).Concat(); 
} 

उपयोग:

var bufferedSource = source.StepInterval(TimeSpan.FromSeconds(1)); 

Answer 5

.Buffer(TimeSpan.FromSeconds(0.2)).Where(i => i.Any()) 
.Subscribe(buffer => 
{ 
    foreach(var item in buffer) Console.WriteLine(item) 
}); 

Answer 6

मैं Observable.Zip उपयोग करने के साथ एक दृष्टिकोण का सुझाव देना चाहते:

// Incoming requests 
var requests = new[] {1, 2, 3, 4, 5}.ToObservable(); 

// defines the frequency of the incoming requests 
// This is the way to emulate flood of incoming requests. 
// Which, by the way, uses the same approach that will be used in the solution 
var requestsTimer = Observable.Interval(TimeSpan.FromSeconds(0.1)); 
var incomingRequests = Observable.Zip(requests, requestsTimer, (number, time) => {return number;}); 
incomingRequests.Subscribe((number) => 
{ 
    Console.WriteLine($"Request received: {number}"); 
}); 

// This the minimum interval at which we want to process the incoming requests 
var processingTimeInterval = Observable.Interval(TimeSpan.FromSeconds(1)); 

// Zipping incoming requests with the interval 
var requestsToProcess = Observable.Zip(incomingRequests, processingTimeInterval, (data, time) => {return data;}); 

requestsToProcess.Subscribe((number) => 
{ 
    Console.WriteLine($"Request processed: {number}"); 
}); 

Answer 7

मैं इसके साथ खेल रहा था और पाया।पिन (जैसा कि पहले उल्लेख किया है) सबसे सरल विधि के लिए:

var stream = "ThisFastObservable".ToObservable(); 
var slowStream = 
    stream.Zip(
     Observable.Interval(TimeSpan.FromSeconds(1)), //Time delay 
     (x, y) => x); // We just care about the original stream value (x), not the interval ticks (y) 

slowStream.TimeInterval().Subscribe(x => Console.WriteLine($"{x.Value} arrived after {x.Interval}")); 

उत्पादन:

T arrived after 00:00:01.0393840 
h arrived after 00:00:00.9787150 
i arrived after 00:00:01.0080400 
s arrived after 00:00:00.9963000 
F arrived after 00:00:01.0002530 
a arrived after 00:00:01.0003770 
s arrived after 00:00:00.9963710 
t arrived after 00:00:01.0026450 
O arrived after 00:00:00.9995360 
b arrived after 00:00:01.0014620 
s arrived after 00:00:00.9993100 
e arrived after 00:00:00.9972710 
r arrived after 00:00:01.0001240 
v arrived after 00:00:01.0016600 
a arrived after 00:00:00.9981140 
b arrived after 00:00:01.0033980 
l arrived after 00:00:00.9992570 
e arrived after 00:00:01.0003520