Rx를 사용하여 비동기 호출에서 콜백을 스로틀

Question

내 질문은 this 질문과 비슷하지만 항상 약간 다릅니다.

현재 간단한 패턴을 사용하여 기본 데이터 변경 사항에 대한 알림을 비동기 적으로 수신하는 프로젝트에서 작업하고 있습니다.

클래스 푸는 이러한 데이터 변경에 가입하는 책임과 클래스가 지정된 인터페이스를 구현하는 클래스의 인스턴스 등록하여 이러한 변화에 대한 관심을 등록하는 방법을 제공한다 :

public Guid SubscribeToChanges(ISomeCallbackInterface callback) 

클래스 바 구현을 이 콜백 레지스터 자체 :

이상적

public class Bar : ISomeCallbackInterface 
{ 
    ... 
    //initialise instance of Foo class and subscribe to updates 
    _foo.SubscribeToChanges(this); 
    ... 

    public void CallbackMethod(string id, IEnumerable<Tuple<string, int, object>> data) 
    { 
     ... 
    } 
} 

우리는 예를 들면 X 위스콘신 뒤쪽 x에서 y까지의 특정 데이터의 값을 변경하고 콜백을받을 수 있으므로,이 콜백 스로틀하고자 얇은 1 초 공간. Rx 문서를 보면 DistinctUntilChanged 연산자를 사용하면 간단합니다.

그러나 질문은 위에서 주어진 콜백 구현에 연산자를 적용 할 수있는 IObservable 컬렉션을 만드는 방법입니다. 문서는 표준 .Net 이벤트 또는 Begin .../End ... 메소드 쌍으로부터 IObservable을 작성하는 방법에 대해 매우 명확합니다.

업데이트 : Richard Szalay가 자신의 의견에서 지적했듯이 DistinctUntilChanged를 Throttle과 함께 사용해야 내가 필요한 것을 얻을 수 있습니다.

리차드 덕분에 콜백 수신 거부 기능이 필요하다는 것도 언급해야합니다. 현재 모델에서는 Foo 인스턴스에서 Unscubscribe (Guid subscriptionToken)를 호출하기 만합니다.

Answer 1

내가 할 수있는 유일한 방법은 ISomeCallbackInterface과 Observable.Create의 맞춤 구현입니다. 세 대

public static IObservable<Tuple<string,IEnumerable<Tuple<string, int, object>>> 
    FromCustomCallbackPattern(ISomeCallbackPublisher publisher) 
{ 
    return Observable.CreateWithDisposable<T>(observer => 
    { 
     var subject = new Subject< 
      Tuple<string,IEnumerable<Tuple<string, int, object>>>(); 

     var callback = new ObservableSomeCallback(subject); 

     Guid subscription = publisher.SubscribeToChanges(callback); 

     return new CompositeDisposable(
      subject.Subscribe(observer), 
      Disposable.Create(() => publisher.UnsubscribeFromChanges(subscription)) 
     ); 
    }); 
} 

private class ObservableSomeCallback : ISomeCallbackInterface 
{ 
    private IObserver<Tuple<string,IEnumerable<Tuple<string, int, object>>> observer; 

    public ObservableSomeCallback(
     IObserver<Tuple<string,IEnumerable<Tuple<string, int, object>>> observer) 
    { 
     this.observer = observer; 
    } 

    public void CallbackMethod(string id, IEnumerable<Tuple<string, int, object>> data) 
    { 
     this.observer.OnNext(new Tuple< 
      string,IEnumerable<Tuple<string, int, object>>(id, data)); 
    } 
}