轉化的一系列事件到值

Question

總之更細化的順序，我試圖用反應式庫實現一個簡單的尾巴工具來主動監控新的生產線，因爲它們可以附加到文件。以下是我走到這一步：

static void Main(string[] args) 
    { 
     var filePath = @"C:\Users\wbrian\Documents\"; 
     var fileName = "TestFile.txt"; 
     var fullFilePath = filePath + fileName; 
     var fs = new FileStream(fullFilePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite); 
     var sr = new StreamReader(fs, true); 
     sr.ReadToEnd(); 
     var lastPos = fs.Position; 

     var watcher = new FileSystemWatcher(filePath, fileName); 
     watcher.NotifyFilter = NotifyFilters.Size; 
     watcher.EnableRaisingEvents = true; 

     Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
      action => watcher.Changed += action, 
      action => watcher.Changed -= action) 
      .Throttle(TimeSpan.FromSeconds(1)) 
      .Select(e => 
       { 
        var curSize = new FileInfo(fullFilePath).Length; 
        if (curSize < lastPos) 
        { 
         //we assume the file has been cleared, 
         //reset the position of the stream to the beginning. 
         fs.Seek(0, SeekOrigin.Begin); 
        } 
        var lines = new List<string>(); 
        string line; 
        while((line = sr.ReadLine()) != null) 
        { 
         if(!string.IsNullOrWhiteSpace(line)) 
         { 
          lines.Add(line); 
         } 
        } 
        lastPos = fs.Position; 
        return lines; 
       }).Subscribe(Observer.Create<List<string>>(lines => 
       { 
        foreach (var line in lines) 
        { 
         Console.WriteLine("new line = {0}", line); 
        } 
       })); 

     Console.ReadLine(); 
     sr.Close(); 
     fs.Close(); 
    } 

正如你所看到的，我創建了一個可觀察從FileWatcher事件，觸發該文件的大小改變時的事件。從那裏，我確定哪些行是新的，並且observable返回新行的列表。理想情況下，可觀察序列只是一個代表每個新行的字符串。觀察結果返回列表的唯一原因是因爲我根本不知道如何按摩它以做到這一點。任何幫助將不勝感激。

Answer 1

您可以使用SelectMany：

SelectMany(lines => lines) 
.Subscribe(Observer.Create<string>(line => { Console.WriteLine("new line = {0}", line); });