捕获异常(乒乓反应样例)

我从响应式框架开始。基于示例:

http://rxwiki.wikidot.com/101samples(乒乓球)

我在我的应用程序中做了类似的实现。我不能也不知道如何将一个动作传递给OnError(在类之外访问，issubject实现)

我该怎么办?

public class PFocusPing : ISubject<PFocusPong, PFocusPing>
{
    private readonly AdvancedConnection device;
    public PFocusPing(AdvancedConnection con)
    {
        this.device = con;
    }
    #region Implementation of IObserver<PFocusPong>
    public void OnNext(PFocusPong value)
    {
        this.device.SendData(clsPowerFocus.DataKeepAlive);
        Console.WriteLine("PFocusPing received PFocusPong.");
    }
    public void OnError(Exception exception)
    {
        Console.WriteLine("PFocusPing experienced an exception and had to quit playing.");
    }
    public void OnCompleted()
    {
        Console.WriteLine("PFocusPing finished.");
    }
    #endregion
    #region Implementation of IObservable<PFocusPing>
    public IDisposable Subscribe(IObserver<PFocusPing> observer)
    {
        return Observable.FromEventPattern<Args<DataReceive>>(add => device.DataReceived += add,
                                                     rem => device.DataReceived -= rem)
                                                    .Where(obj => obj.EventArgs.Value.Result == DataReceive.CONNECTION_RESULT.SUCESS)
                                                    .Where(obj => obj.EventArgs != null &&
                                                                  obj.EventArgs.Value != null &&
                                                                  obj.EventArgs.Value.Data != null &&
                                                                  obj.EventArgs.Value.Data.Length > 0)
                                                    .Select(obj => this)
                                                    .Subscribe(observer);
    }
    #endregion
    #region Implementation of IDisposable
    public void Dispose()
    {
        OnCompleted();
    }
    #endregion
 }
public class PFocusPong : ISubject<PFocusPing, PFocusPong>
{
    private readonly int timeout;
    private readonly int reply;
    private DateTime lastAnswer;
    public PFocusPong(int msPing, int msTimeout)
    {
        this.reply = Math.Max(1000, msPing);
        this.timeout = Math.Max(2000, msTimeout);
    }
    #region Implementation of IObserver<Ping>
    public void OnNext(PFocusPing value)
    {
        this.lastAnswer = DateTime.Now;
        Console.WriteLine("PFocusPong received Ping.");
    }
    public void OnError(Exception exception)
    {
        Console.WriteLine("PFocusPong experienced an exception and had to quit playing.");
    }
    public void OnCompleted()
    {
        Console.WriteLine("PFocusPong finished.");
    }
    #endregion
    #region Implementation of IObservable<PFocusPong>
    public IDisposable Subscribe(IObserver<PFocusPong> observer)
    {
        var keepAlive =
            Observable.Timer(TimeSpan.FromMilliseconds(0), TimeSpan.FromMilliseconds(reply))
            .Select(obj =>
                {
                    if ((DateTime.Now - this.lastAnswer).TotalMilliseconds < reply)
                    {
                        Console.WriteLine("True");
                        return true;
                    }
                    Console.WriteLine("False");
                    return false;
                });
        var keepDead = keepAlive.Where(obj => obj).Timeout(TimeSpan.FromMilliseconds(timeout)).Subscribe(r =>
            {
                if (!r)
                    throw new TimeoutException();
            });

        var disposable = keepAlive.Finally(keepDead.Dispose).Select(n => this).Subscribe(observer);
        return disposable;
    }
    #endregion
    #region Implementation of IDisposable
    public void Dispose()
    {
        OnCompleted();
    }
    #endregion
}