如何在服务器端取消订阅观察者答案

【问题标题】：How to unsubscribe observer on server side如何在服务器端取消订阅观察者
【发布时间】：2013-01-09 08:36:10
【问题描述】：

我有一个应用程序，它通过与一台服务器（可观察的）和许多客户端（观察者）进行远程处理来使用 RX。我的问题是，当客户端（观察者）在不执行 ubsubscribe(dispose) 的情况下无法正确断开连接时，服务器上的 OnNext() 函数开始引发远程处理异常。

是否有任何机制可以在服务器端取消订阅有问题的观察者？

客户端代码部分：

internal void SetRemoting(bool refreshInstance)
{
    string channelName = "RemotingClientUI";
    IDictionary dict = new Hashtable();
    dict["port"] = 9988;
    dict["name"] = channelName;

    var bcp = new BinaryClientFormatterSinkProvider();
    var channel = new TcpClientChannel(dict, bcp);
    ChannelServices.RegisterChannel(channel, false);

    _remoteServer = (IRemoteServerService) Activator.
        GetObject(typeof (IRemoteServerService),
            tcp://...");
}

private void SubscribeToRemoteEvents(bool unSubscrubeFirst)
{            
    _jobRowUpdate = _remoteServer.JobRowUpdate.Subscribe(UpdateJobQueueRow);
    _packageRowUpdate = _remoteServer.PackageRowUpdate.   
        Subscribe(UpdatePackageQueueRow);
    _miscUpdate = _remoteServer.MiscAction.Subscribe(MiscRemoteActions);
}

服务器代码部分：

public class RemoteServiceService
{
    public RemoteServiceService()
    {
        JobRowUpdate = LoggerFactory.GetLogger(
            LoggerType.RemoteService, this).JobRowUpdate.Remotable();
        PackageRowUpdate = LoggerFactory.GetLogger(
            LoggerType.RemoteService, this).PackageRowUpdate.Remotable();
        MiscAction = LoggerFactory.GetLogger(
            LoggerType.RemoteService, this).MiscActions.Remotable();
    }
}

public class RemoteLoggerForService
{
    private RemoteLoggerForService(IService service)
    {
        _jobRowUpdate = new Subject<IJobQueueRow>();
        _packageRowUpdate = new Subject<IPackageQueueRow>();
        _miscActions = new Subject<MiscRemoteObjects>();
        _service = service;
    }

    #region Overrides of LoggerBase

    public override void WriteToLog<T>(T stringFormatOrObject, 
        params object[] args)
    {
        lock (this)
        try
        {
            lock (LockLogger)
            {
                if (stringFormatOrObject is IJobQueueRow && 
                    _jobRowUpdate != null)
                {
                    _jobRowUpdate.OnNext(
                        stringFormatOrObject as IJobQueueRow);
                }

                if (stringFormatOrObject is IPackageQueueRow && 
                    _packageRowUpdate != null)
                {
                    _packageRowUpdate.OnNext(
                        stringFormatOrObject as IPackageQueueRow);
                }

                if (stringFormatOrObject is MiscRemoteObjects && 
                    _miscActions != null)
                {
                    _miscActions.OnNext(
                        stringFormatOrObject as MiscRemoteObjects);
                }
            }
        }
        catch(Exception ex)
        {
            LoggerFactory.GetLogger(LoggerType.File, null).
                WriteToLog(
                    Utils.GetFullException("RemoteLoggerForService", ex));
        }
    }

    #endregion
}

【问题讨论】：

考虑重新措辞。我不明白你的问题。当问题出在其中之一时，您似乎将问题与 Rx 和远程处理混为一谈。
observable 和observers 在不同的计算机上。我正在使用 ISubject.Remotable() 扩展。一切正常，但客户端断开连接会导致服务器崩溃并出现“远程异常”。问题是如何使 Observable 集合忽略或删除已断开连接的客户端的订阅
你订阅的怎么样了？您是否添加了错误/异常处理程序？还是渔获？您需要提供更多信息。
我已经更新了问题
您可能需要考虑另一种方法......例如使用 Rx + SignalR 和 OData 和 IQbservables 进行远程订阅的 PushQa。 petegoo.github.com/Pushqa

标签： system.reactive .net-remoting

【解决方案1】：

我认为您正在尝试做的是吞下异常并让您的服务器继续运行。您可以通过捕获异常（就像您在此处所做的那样）来做到这一点，然后您可以将其传递给相关 Observable 的 OnError 方法，以便订阅者可以选择如何做出反应。

【讨论】：