觸發第一個事件幾秒鐘後的RX緩衝區事件

Question

我有一個filewatcher，從中觀察創建和更改的事件。 我希望當第一個事件被觸發（創建或改變）時，它需要開始緩衝10秒，並在那10秒後我想處理緩衝的事件。

我得已經是這樣的：

Observable.FromEventPattern<FileSystemEventArgs>(FileSystemWatcher, "Created") 
       .Merge(Observable.FromEventPattern<FileSystemEventArgs>(FileSystemWatcher, "Changed")) 
       .Buffer(TimeSpan.FromSeconds(10)) 
       .Subscribe(list => 
       { 
        Debug.WriteLine("Do something"); 
       }); 

此代碼 '的Debug.WriteLine（ 「做什麼」）;'每10秒鐘一次。

編輯： 好吧，讓我試着用時間線來解釋它。

1. 文件觀察器空閒，沒有事件觸發。
2. 一段未知的時間後的文件被放置在目錄被觸發
3. 創建的事件
4. 可觀察名單開始，持續10秒
5. 緩衝（所有事件）的10秒訂閱操作後被執行，它會處理所有的事件一次

希望這會使事情清楚

Answer 1

我假設你想要的以下行爲：

1. 初始事件發生後，緩衝所有事件在接下來的10秒。
2. 一旦這個10秒的窗口關閉，下一個應該在10秒後爲所有事件觸發一個新的10秒緩衝區。

所以我們假設我們有5個事件在5秒內均勻分佈，13秒間隔，然後在5秒內均勻分佈另外5個事件。大理石圖是這樣的：

timeline: 0--1--2--3--4--5--6--7--8--9-10-11-12-13-14-15-16-17-18-19-20-21-22-23-24-25-26-27 
events : x--x--x--x--x-------------------------------------x--x--x--x--x------------------ 
stdbuff : |----------------------------|-----------------------------|--------------------- 
desired : BeginCapture-----------------Return---------------BeginCapture------------------Return 

使用直線前進Buffer的是，它看起來像上面譜寫的stdbuff，而分手事件的第二組分爲兩組，造成兩人列出了問題第二組活動：一個有三個活動，一個有兩個活動。您需要一個列表（針對第二組），使用類似desired流的邏輯。開始在0捕獲，返回列表在10.開始在17捕獲，在27返回列表。

如果我誤解了你（再次），請張貼大理石圖表，類似於上面，代表你想要如何事情要工作。

---

假設我理解你正確，下面的代碼將工作...

//var initialSource = Observable.FromEventPattern<FileSystemEventArgs>(fileWatcher, nameof(FileSystemWatcher.Created)) 
// .Merge(Observable.FromEventPattern<FileSystemEventArgs>(fileWatcher, nameof(FileSystemWatcher.Changed))); 

    //Comment this out, and use the above lines for your code. This just makes testing the Rx components much easier. 
var initialSource = Observable.Interval(TimeSpan.FromSeconds(1)).Take(5) 
    .Concat(Observable.Empty<long>().Delay(TimeSpan.FromSeconds(13))) 
    .Concat(Observable.Interval(TimeSpan.FromSeconds(1)).Take(5)); 

initialSource 
    .Publish(_source => _source 
     .Buffer(_source 
      .Scan(DateTimeOffset.MinValue, (lastPrimary, _) => DateTimeOffset.Now - lastPrimary > TimeSpan.FromSeconds(10) ? DateTimeOffset.Now : lastPrimary) 
      .DistinctUntilChanged() 
      .Delay(TimeSpan.FromSeconds(10)) 
     ) 
    ) 
    .Subscribe(list => 
    { 
     Debug.WriteLine($"Time-stamp: {DateTime.Now.ToLongTimeString()}"); 
     Debug.WriteLine($"List Count: {list.Count}"); 
    }); 

說明：

首先，我們需要確定 '主要事件'，那些代表BeginCapture上面的desired流描述中的註釋。可以發現這樣的：

var primaryEvents = initialSource 
     .Scan(DateTimeOffset.MinValue, (lastPrimary, _) => DateTimeOffset.Now - lastPrimary > TimeSpan.FromSeconds(10) ? DateTimeOffset.Now : lastPrimary) 
     .DistinctUntilChanged(); 

一旦我們有了BeginCapture事件，它可以表示一個窗口開放，這是很容易找到Return事件，或閉窗：

var closeEvents = primaryEvents.Delay(TimeSpan.FromSeconds(10)); 

在實踐中，因爲沒有密切和開放的，我們關心之間發生的，我們只需要擔心關閉事件，所以我們可以把它收縮到這一點：

var closeEvents = initialSource 
     .Scan(DateTimeOffset.MinValue, (lastPrimary, _) => DateTimeOffset.Now - lastPrimary > TimeSpan.FromSeconds(10) ? DateTimeOffset.Now : lastPrimary) 
     .DistinctUntilChanged() 
     .Delay(TimeSpan.FromSeconds(10)); 

把它插入到Buffer與closeEvents作爲bufferBoundaries：

var bufferredLists = initialSource 
    .Buffer(initialsource 
     .Scan(DateTimeOffset.MinValue, (lastPrimary, _) => DateTimeOffset.Now - lastPrimary > TimeSpan.FromSeconds(10) ? DateTimeOffset.Now : lastPrimary) 
     .DistinctUntilChanged() 
     .Delay(TimeSpan.FromSeconds(10)) 
    ); 

最後，因爲我們有多個訂閱initialSource，我們需要使用Publish確保併發正常工作，導致最終的答案。