遞歸/扇出在無擴展

Question

我試圖拼湊出的Rx流水線的工作方式如下：

1. 我寫了一個函數，它在提供的IObservable我包含有關信息檔案公司
2. 我查詢各種數據源以查找潛在的相關公司配置文件，並行查詢。我將它合併成一個公司簡介的單個IObservable。
3. 當我拿回這些潛在相關的配置文件，我把它們比作是我已經觀察到型材和，如果他們有針對性> 80％，是不一樣的，因爲我已經觀察到的任何個人資料，我認爲他們比賽。
4. 我想匹配的公司反饋到第1步，這樣我可以搜索相關的數據，這些新的配套文件。

我引導了一些已知的良好配置文件的過程。

最後，還有那些尚未看到任何更多的匹配型材等過程結束。

我無法編程此。如果我使用Subject來允許管道尾端將其配置文件發送到工作流程的開始，那麼沒有人會調用OnCompleted，並且我從未發現該流程已經結束。如果我用遞歸來開發這個，我似乎總是以堆棧溢出結束，因爲我試圖用自己的返回值來調用一個函數。

誰能幫我我怎麼能在某種程度上，我可以確定的過程已經結束完成這項任務？

Answer 1

這聽起來像你想這樣的數據流：

seed profiles --> source --> get related --> output 
        ^     | 
        |     v 
        -<--- transform <----- 

這看起來是解決地方的一般問題是不是特定的一個容易或更容易的情況下，所以我會提出一個通用的「反饋「應該給你的積木功能，你需要：

編輯：固定函數來完成

IObservable<TResult> Feedback<T, TResult>(this IObservable<T> seed, 
              Func<T, IObservable<TResult>> produce, 
              Func<TResult, IObservable<T>> feed) 
    { 
     return Observable.Create<TResult>(
       obs => 
       { 
        var ret = new CompositeDisposable(); 
        Action<IDisposable> partComplete = 
         d => 
         { 
          ret.Remove(d); 
          if (ret.Count == 0) obs.OnCompleted(); 
         }; 
        Action<IObservable<T>, Action<T>> ssub = 
         (o, n) => 
         { 
          var disp = new SingleAssignmentDisposable(); 
          ret.Add(disp); 
          disp.Disposable = o.Subscribe(n, obs.OnError,() => partComplete(disp)); 
         }; 
        Action<IObservable<TResult>, Action<TResult>> rsub = 
         (o, n) => 
         { 
          var disp = new SingleAssignmentDisposable(); 
          ret.Add(disp); 
          disp.Disposable = o.Subscribe(n, obs.OnError,() => partComplete(disp)); 
         }; 

        Action<T> recurse = null; 
        recurse = s => 
           { 
            rsub(produce(s), 
             r => 
             { 
              obs.OnNext(r); 
              ssub(feed(r), recurse); 
             }); 
           }; 

        ssub(seed, recurse); 
        return ret; 
       }); 
    } 

在你的情況下，T和TResult看起來是一樣的，所以feed將是身份函數。 produce將用於實施步驟2和3

我測試了這個功能的一些示例代碼的功能：

void Main() 
{ 
    var seed = new int[] { 1, 2, 3, 4, 5, 6 }; 
    var found = new HashSet<int>(); 
    var mults = seed.ToObservable() 
        .Feedback(i => 
           { 
            return Observable.Range(0, 5) 
             .Select(r => r * i) 
             .TakeWhile(v => v < 100) 
             .Where(v => found.Add(v)); 
           }, 
           i => Observable.Return(i)); 

    using (var disp = mults.Dump()) 
    { 
     Console.WriteLine("Press any key to stop"); 
     Console.ReadKey(); 
    } 
    Console.WriteLine("Press any key to exit"); 
    Console.ReadKey(); 
} 

static IDisposable Dump<T>(this IObservable<T> source) 
{ 
    return source.Subscribe(item => Console.WriteLine(item), 
          ex => Console.WriteLine("Error occurred in dump observable: " + ex.ToString()), 
          () => Console.WriteLine("Dump completed")); 
}