debugging - manager - ¿Cómo puedo ver lo que está haciendo mi consulta de extensiones reactivas?

google tag manager

Question

Estoy escribiendo una consulta compleja de Reactive Extensions con muchos operadores. ¿Cómo puedo ver lo que está pasando?

Estoy preguntando y respondiendo esto, ya que aparece un poco y es probablemente de un buen uso general.

Answer 1

Puede agregar esta función libremente a sus operadores Rx mientras los desarrolla para ver lo que está sucediendo:

 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);
}
});
}

Aquí hay un ejemplo de uso, muestra una sutil diferencia de comportamiento de Range :

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

Da la salida:

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

Pero esto:

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

Da la salida:

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

¿Encuentra la diferencia?

Obviamente, puede modificar esto para escribir en los registros o para depurar, o usar las directivas de preprocesador para hacer una suscripción mínima de paso en una compilación de lanzamiento, etc.

Puede aplicar Spy través de una cadena de operadores. p.ej:

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

Da la salida:

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

Estoy seguro de que puedes encontrar formas de enriquecer esto para tus propósitos.