如何查看我的反向扩展查询正在做什么？ [英] How can I see what my reactive extensions query is doing?

问题描述

我正在编写一个复杂的Reactive Extensions查询，其中包含许多操作符。我如何看到发生了什么？

我正在询问和回答这个问题，因为它出现了很多，可能是很好的一般使用。

解决方案

您可以将这个函数自由地附加到您的Rx运算符，同时开发它们以查看发生的情况：

  public static IObservable< T> Spy< T>（该IObservable< T>源，字符串opName = null）
 {
 opName = opName？ IObservable; 
 Console.WriteLine（{0}：在线程：{1}获取的Observable，
 opName，
 Thread.CurrentThread.ManagedThreadId）; 
 
 return Observable.Create< T>（obs => 
 {
 Console.WriteLine（{0}：订阅到线程：{1}，
 opName，
 Thread.CurrentThread.ManagedThreadId）; 
 
 try 
 {
 var subscription = source 
 .Do（x => Console。 WriteLine（{0}：OnNext（{1}）on Thread：{2}，
 opName，
x，
 Thread.CurrentThread.ManagedThreadId），
 ex => ; Console.WriteLine（{0}：OnError（{1}）on Thread：{2}，
 opName，
 ex，
 Thread.CurrentThread.ManagedThreadId），
（）=> ; Console.WriteLine（{0}：OnCompleted（）on Thread：{1}，
 opName，
 Thread.CurrentThread.ManagedThreadId）
）
 .Subscribe（obs） ; 
 return new CompositeDisposable（
 subscription，
 Disposable.Create（（）=> Console.WriteLine（
{0}：Threaded on Thread：{1}， 
 opName，
 Thread.CurrentThread.ManagedThreadId）））; 
} 
 finally 
 {
 Console.WriteLine（{0}：Subscription completed。，opName）; 
} 
}）; 
} 
  

以下是一个示例用法，显示了范围：

  Observable.Range（0，1）.Spy（Range）。订阅（）; 
  

提供输出：

 范围：可观察获得线程：7 
范围：订阅线程：7 
范围：订阅完成。 
范围：OnNext（0）on线程：7 
范围：OnCompleted（）on线程：7 
范围：清理线程：7 
  

但是这个：

  Observable.Range ，1，Scheduler.Immediate）.Spy（Range）。Subscribe（）; 
  

提供输出：

 范围：可观察获得线程：7 
范围：订阅线程：7 
范围：OnNext（0）on线程：7 
范围：OnCompleted（）在线程：7 
范围：订阅完成。 
范围：在线程清理7 
  

找到差异？

显然，您可以将其改为写入日志或调试，或使用预处理指令在发布版本等上执行精简直通订阅。

您可以在整个运营商链中申请 Spy 。例如：

  Observable.Range（0,3）.Spy（Range）
 .Scan（（acc ，i）=> acc + i）.Spy（Scan）。Subscribe（）; 
  

提供输出：

 范围：可观察获得线程：7 
扫描：可观察获得线程：7 
扫描：订阅线程：7 
范围：订阅线程： 7 
范围：订阅完成。 
扫描：订阅完成。 
范围：OnNext（1）on线程：7 
扫描：OnNext（1）on线程：7 
范围：OnNext（2）on线程：7 
扫描：OnNext 3）on线程：7 
范围：OnCompleted（）线程：7 
扫描：OnCompleted（）线程：7 
范围：清理线程：7 
扫描：清理线程：7 
  

我相信你可以找到丰富这个方式来适应你的目的。

I'm writing a complex Reactive Extensions query with lots of operators. How can I see what's going on?

I'm asking and answering this as it comes up a fair bit and is probably of good general use.

解决方案

You can append this function liberally to your Rx operators while you are developing them to see what's happening:

    public static IObservable<T> Spy<T>(this IObservable<T> source, string opName = null)
    {
        opName = opName ?? "IObservable";
        Console.WriteLine("{0}: Observable obtained on Thread: {1}",
                          opName,
                          Thread.CurrentThread.ManagedThreadId);

        return Observable.Create<T>(obs =>
        {
            Console.WriteLine("{0}: Subscribed to on Thread: {1}",
                              opName,
                              Thread.CurrentThread.ManagedThreadId);

            try
            {
                var subscription = source
                    .Do(x => Console.WriteLine("{0}: OnNext({1}) on Thread: {2}",
                                                opName,
                                                x,
                                                Thread.CurrentThread.ManagedThreadId),
                        ex => Console.WriteLine("{0}: OnError({1}) on Thread: {2}",
                                                 opName,
                                                 ex,
                                                 Thread.CurrentThread.ManagedThreadId),
                        () => Console.WriteLine("{0}: OnCompleted() on Thread: {1}",
                                                 opName,
                                                 Thread.CurrentThread.ManagedThreadId)
                    )
                    .Subscribe(obs);
                return new CompositeDisposable(
                    subscription,
                    Disposable.Create(() => Console.WriteLine(
                          "{0}: Cleaned up on Thread: {1}",
                          opName,
                          Thread.CurrentThread.ManagedThreadId)));
            }
            finally
            {
                Console.WriteLine("{0}: Subscription completed.", opName);
            }
        });
    }

Here's an example usage, shows a subtle behaviour difference of Range:

Observable.Range(0, 1).Spy("Range").Subscribe();

Gives the output:

Range: Observable obtained on Thread: 7
Range: Subscribed to on Thread: 7
Range: Subscription completed.
Range: OnNext(0) on Thread: 7
Range: OnCompleted() on Thread: 7
Range: Cleaned up on Thread: 7

But this:

Observable.Range(0, 1, Scheduler.Immediate).Spy("Range").Subscribe();

Gives the output:

Range: Observable obtained on Thread: 7
Range: Subscribed to on Thread: 7
Range: OnNext(0) on Thread: 7
Range: OnCompleted() on Thread: 7
Range: Subscription completed.
Range: Cleaned up on Thread: 7

Spot the difference?

Obviously you can alter this to write to logs or to Debug, or use preprocessor directives to do a lean pass-through subscription on a Release build etc...

You can apply Spy throughout a chain of operators. e.g.:

Observable.Range(0,3).Spy("Range")
          .Scan((acc, i) => acc + i).Spy("Scan").Subscribe();

Gives the output:

Range: Observable obtained on Thread: 7
Scan: Observable obtained on Thread: 7
Scan: Subscribed to on Thread: 7
Range: Subscribed to on Thread: 7
Range: Subscription completed.
Scan: Subscription completed.
Range: OnNext(1) on Thread: 7
Scan: OnNext(1) on Thread: 7
Range: OnNext(2) on Thread: 7
Scan: OnNext(3) on Thread: 7
Range: OnCompleted() on Thread: 7
Scan: OnCompleted() on Thread: 7
Range: Cleaned up on Thread: 7
Scan: Cleaned up on Thread: 7

I'm sure you can find ways of enriching this to suit your purposes.

这篇关于如何查看我的反向扩展查询正在做什么？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！