метод отсутствует в реактивных расширениях

Question

Я пытаюсь реализовать агрегатор событий, используя RX (Reactive extensions), здесь используется код am am, но subject.AddDisposable метод отсутствует. Может кто-нибудь мне помочь? может быть, это старая версия, поэтому в новой версии Rx этот метод был удален? если это так, а что это правильный способ реализовать это?

if (_observablesByTypeKey.ContainsKey(key)) 
      { 
       Tuple<object, object> tuple = _observablesByTypeKey[key]; 
       stream = (IObservable<T>)tuple.Item2; 
      } 
      else 
      { 
       Type specificSubjectType = typeof(Subject<>).MakeGenericType(new[] { typeof(T) }); 
       var subject = (Subject<T>)Activator.CreateInstance(specificSubjectType, new object[] { }); 

       var removeEventStreamFromCache = Disposable.Create(
        () => 
         { 
          lock (_observablesByTypeKeyLock) 
          { 
           _observablesByTypeKey.Remove(key); 
          } 
         } 
        ); 

       stream = subject.AddDisposable(removeEventStreamFromCache).Publish().RefCount(); 

       var tuple = new Tuple<object, object>(subject, stream); 
       _observablesByTypeKey.Add(key, tuple); 

Answer 1

Вы можете заменить

stream = subject.AddDisposable(removeEventStreamFromCache).Publish().RefCount(); 

с

stream = Observable.Create(observer => 
    new CompositeDisposable(
    subject.Subscribe(observer), 
    removeEventStreamFromCache)); 
stream.Publish().RefCount();