带序列超时提示的Observale

Posted 2023-02-21 henreash

Observable.Timeout扩展提供了序列超时的实现，但不够灵活，超时后就异常导致订阅失效。自定义的Observable更加灵活。 

IDisposable customerTimeoutSubscriber;

        private void button10_Click(object sender, EventArgs e)
       
            var cnt = 0;
            var observable = Observable.Create<Tuple<int, bool, string>>(ob =>
           
                var timeoutEvent = new ManualResetEvent(false);
                Task.Factory.StartNew(() =>
               
                    while (true)
                   
                        var isOk = timeoutEvent.WaitOne(2500);
                        if (cancelSource.Token.IsCancellationRequested)
                            break;
                        if (!isOk)
                       
                            ob.OnNext(new Tuple<int, bool, string>(cnt, false, "序列超时"));
                            //Interlocked.Increment(ref cnt);
                       
                        timeoutEvent.Reset();
                   
                );
                while (true)
               
                    if (cancelSource.Token.IsCancellationRequested)
                        break;
                    ob.OnNext(new Tuple<int, bool, string>(cnt, false, DateTime.Now.ToString()));
                    Interlocked.Increment(ref cnt);
                    timeoutEvent.Set();
                    //超时策略
                    var r = new Random().Next(1, 4);
                    Debug.WriteLine($"cnt sleep rs");
                    Thread.Sleep(r * 1000);
               
                ob.OnCompleted();
                return Disposable.Empty;
            );
            //observable = Observable.Timeout(observable, new TimeSpan(0, 0, 2));
            customerTimeoutSubscriber = observable.ObserveOn(new DispatcherScheduler(Dispatcher.CurrentDispatcher)).SubscribeOn(NewThreadScheduler.Default).Subscribe((v) =>
           
                Debug.WriteLine($"v.Item1 v.Item3");
            ,
            ex =>
           
                Debug.WriteLine(ex.Message);
                //cancelSource.Cancel();
            ,
            () => Debug.WriteLine("complate"); );