C# 的另一个“如何在不使用 Subject<t> 的情况下做到这一点”问题

Question

7 年前 StackOverflow 上有很多关于使用 ReactiveX 的问题，以及使用 Subjects 的应该或不。这些问题中的大多数最终都被人们在哲学基础上来回争论，最终在实际示例中相当轻松，并谈论而不是“正确”方法与“最佳”方法。

我编写了一个简单的示例类，它尝试将 SignalR 的 IHubProxy.On 桥接到 IObservable。

这是我的课，请教我如何重写这个课，这样就不需要主题了。

我只是没有办法这样做。

using Microsoft.AspNet.SignalR;
using Microsoft.AspNet.SignalR.Client;
using Microsoft.AspNet.SignalR.Hubs;
using System.Reactive.Linq;
using Owin;
using System;
using System.Reactive.Subjects;

namespace Acme.Core.SignalR
{
    public class SignalRPipe :IObservable<PipeCommand>, IDisposable
    {
        readonly IHubProxy _MyHub;
        readonly IDisposable _HubOnListener;

        // this next line is BAD and I want to get rid of it, but I can't figure out how.
        private readonly Subject<PipeCommand> subject = new Subject<PipeCommand>();

        private SignalRPipe() : base()
        {
        }

        private SignalRPipe( IHubProxy hub ) : this()
        {
            _MyHub = hub;
            _HubOnListener = _MyHub.On<PipeCommand>( "OnPipeCommand" , OnPipeCommand );
        }

        private void OnPipeCommand( PipeCommand obj )
        {
            subject.OnNext( obj );
        }

        public IDisposable Subscribe( IObserver<PipeCommand> observer )
        {
            return subject.Subscribe( observer );
            //var rtrn = Observable.Create<PipeCommand>( ob => {observer.Subscribe(t=> {  }, ,  } );
            //return rtrn;
        }


        #region IDisposable Support
        private bool disposedValue = false; // To detect redundant calls

        protected virtual void Dispose( bool disposing )
        {
            if ( !disposedValue )
            {
                disposedValue = true;
                if ( disposing )
                {
                    _HubOnListener.Dispose();
                    subject.Dispose();
                }

            }
        }

        public void Dispose()
        {
            // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
            Dispose( true );
        }

        #endregion

    }


}

Answer 1

事实证明，这比我最初预期的要容易得多。不断挖掘它并想出了这个答案......

这行得通吗（不用担心“是否正确”）？

using Microsoft.AspNet.SignalR.Client;
using System;

namespace Acme.Core.SignalR
{
    public class SignalRPipe : IObservable<PipeCommand>
    {
        readonly IHubProxy _MyHub;

        private SignalRPipe() : base()
        {
        }

        private SignalRPipe( IHubProxy hub ) : this()
        {
            _MyHub = hub;
        }

        public IDisposable Subscribe( IObserver<PipeCommand> observer )
        {
            var rtrn = _MyHub.On<PipeCommand>( "OnPipeCommand" , pc => observer.OnNext( pc ) );
            return rtrn;
        }

    }

}