잠금 장치를 사용하지 않고 Rx 관찰자가 완료 될 때까지 기다리십시오.

Question

여러 관찰자와 관찰 가능한 간격을 실행하는 응용 프로그램이 있습니다. 간격이 0.5 초마다 웹 서버에서 일부 XML 데이터를로드하면 관찰자는 백그라운드 스레드에서 응용 프로그램 특정 처리를 수행합니다. 데이터가 더 이상 필요하지 않으면 구독 및 간격 관찰이 처리되므로 관찰자의 OnNext/OnCompleted/OnError는 더 이상 호출되지 않습니다. 여태까지는 그런대로 잘됐다.

내 문제 : 드물게 내 Observer의 OnNext 메서드를 호출 한 후에도 여전히 실행 중일 수 있습니다! 폐기 후 추가 작업을 진행하기 전에 OnNext가 완료되었는지 확인하고 싶습니다.

내 현재 해결 방법 : 내 관찰자 클래스 (코드 참조)에 보관함 필드를 도입했습니다. 처리를 마친 후 잠금을 획득하고 잠금을 획득 한 후에 만 ​​계속 시도합니다. 이 솔루션이 작동하는 동안 (?), 그것은 어떻게 든 저에게 잘못 느끼는 것입니다.

질문 :이 문제를 해결하기 위해보다 우아하고 "Rx Way"가 있습니까?

using System; 
using System.Collections.Generic; 
using System.Linq; 
using System.Reactive.Concurrency; 
using System.Reactive.Linq; 
using System.Text; 
using System.Threading; 
using System.Threading.Tasks; 

namespace RxExperimental 
{ 
    internal sealed class MyXmlDataFromWeb 
    { 
     public string SomeXmlDataFromWeb { get; set; } 
    } 

    internal sealed class MyObserver : IObserver<MyXmlDataFromWeb> 
    { 
     private readonly object _locker = new object(); 
     private readonly string _observerName; 

     public MyObserver(string observerName) { 
      this._observerName = observerName; 
     } 

     public object Locker { 
      get { return this._locker; } 
     } 

     public void OnCompleted() { 
      lock (this._locker) { 
       Console.WriteLine("{0}: Completed.", this._observerName); 
      } 
     } 

     public void OnError(Exception error) { 
      lock (this._locker) { 
       Console.WriteLine("{0}: An error occured: {1}", this._observerName, error.Message); 
      } 
     } 

     public void OnNext(MyXmlDataFromWeb value) { 
      lock (this._locker) { 
       Console.WriteLine(" {0}: OnNext running on thread {1}... ", this._observerName, Thread.CurrentThread.ManagedThreadId); 
       Console.WriteLine(" {0}: XML received: {1}", this._observerName, value.SomeXmlDataFromWeb); 
       Thread.Sleep(5000); // simulate some long running operation 
       Console.WriteLine(" {0}: OnNext running on thread {1}... Done.", this._observerName, Thread.CurrentThread.ManagedThreadId); 
      } 
     } 
    } 

    internal sealed class Program 
    { 
     private static void Main() { 
      const int interval = 500; 
      // 
      var dataSource = Observable.Interval(TimeSpan.FromMilliseconds(interval), NewThreadScheduler.Default).Select(_ => { 
       var data = new MyXmlDataFromWeb { 
        SomeXmlDataFromWeb = String.Format("<timestamp>{0:yyyy.MM.dd HH:mm:ss:fff}</timestamp>", DateTime.Now) 
       }; 
       return data; 
      }).Publish(); 
      // 
      var observer1 = new MyObserver("Observer 1"); 
      var observer2 = new MyObserver("Observer 2"); 
      // 
      var subscription1 = dataSource.ObserveOn(NewThreadScheduler.Default).Subscribe(observer1); 
      var subscription2 = dataSource.ObserveOn(NewThreadScheduler.Default).Subscribe(observer2); 
      // 
      var connection = dataSource.Connect(); 
      // 
      Console.WriteLine("Press any key to cancel ..."); 
      Console.ReadLine(); 
      // 
      subscription1.Dispose(); 
      subscription2.Dispose(); 
      connection.Dispose(); 
      // 
      lock (observer1.Locker) { 
       Console.WriteLine("Observer 1 completed."); 
      } 
      lock (observer2.Locker) { 
       Console.WriteLine("Observer 2 completed."); 
      } 
      // 
      Console.WriteLine("Can only be executed, after all observers completed."); 
     } 
    } 
} 

Answer 1

네, 여기에는 더 많은 Rx 방식이 있습니다.

첫 번째 관찰은 관찰 가능한 스트림의 구독 취소가 본질적으로 관찰자 내에서 일어나고있는 것과는 독립적이라는 점입니다. 실제로 어떤 피드백도 없습니다. 관측이 끝났을 때 명확하게 알고 있어야한다는 요구 사항이 있으므로 관측 가능한 스트림으로이를 모델링해야합니다. 즉, 스트림에서 수신 거부하는 대신 을 스트림으로 완료해야 OnComplete 이벤트를 관찰 할 수 있습니다. 귀하의 경우에는 TakeUntil을 사용하여 Observable을 구독 취소하는 대신 종료 할 수 있습니다.

두 번째 관찰은 "관찰자"의 작업이 끝나면 주 프로그램이 관찰해야한다는 것입니다. 그러나 "관찰자"를 실제 IObservable으로 만들었으므로이 작업을 수행 할 방법이 없습니다. 이것은 사람들이 처음 Rx를 사용할 때 나타나는 혼란의 일반적인 원인입니다. "관찰자"를 관찰 가능 체인의 다른 링크로 모델링하면 주 프로그램에서 관찰 할 수 있습니다. 특히, "관찰자"는 들어오는 XML 데이터를 "완료된"메시지로 매핑하는 매핑 작업 (부작용이있는 것)에 불과합니다. 당신이 당신의 코드를 리팩토링 경우

그래서, 당신은 ... 당신이 원하는 것을 얻을 수

public class MyObserver 
{ 
    private readonly string _name; 

    public MyObserver(string name) { _name = name; } 

    public IObservable<Unit> Handle(IObservable<MyXmlDataFromWeb source) 
    { 
     return source.Select(value => 
     { 
      Thread.Sleep(5000); // simulate work 
      return Unit.Default; 
     }); 
    } 
} 

// main 
var endSignal = new Subject<Unit>(); 
var dataSource = Observable 
    .Interval(...) 
    .Select(...) 
    .TakeUntil(endSignal) 
    .Publish(); 
var observer1 = new MyObserver("Observer 1"); 
var observer2 = new MyObserver("Observer 2"); 
var results1 = observer1.Handle(dataSource.ObserveOn(...)); 
var results2 = observer2.Handle(dataSource.ObserveOn(...)); 
// since you just want to know when they are all done 
// just merge them. 
// use ToTask() to subscribe them and collect the results 
// as a Task 
var processingDone = results1.Merge(results2).Count().ToTask(); 

dataSource.Connect(); 

Console.WriteLine("Press any key to cancel ..."); 
Console.ReadLine(); 

// end the stream 
endSignal.OnNext(Unit.Default); 

// wait for the processing to complete. 
// use await, or Task.Result 
var numProcessed = await processingDone; 

당신이 긴 당신의 OnNext에서 작업을 차단 수행하는 경우 나, 당신은 수신의 정신에서 작동하지 않는 것을 주장