Improving Debuggability of ReactiveX

[dwcullop]

One thing I hear from a lot of people who are new to Rx is they find it very hard to debug, and so I'd like to gauge the interest in the development of a new Spy operator that will improve people's ability to debug and/or log their Rx applications.

First of all, I want to stress this is not my idea. All of this code has been adapted from this StackOverflow thread.

In my Q&D test projects, I use something very similar (with added events). It isn't optimal and won't work for everyone, but it is SO handy to be able to see the subscriptions and values and which thread they come on when trying to figure out why something Rx-related isn't working.

The idea is a new operator that will receive a callback interface to log each Rx event as it happens (including subscription and disposal). If there's interest and a suitable API can be made, then I'm happy to submit a PR for this.

The SO post does everything with strings, but that won't work for everyone. So, instead of just handling strings, I'm picturing something that uses an interface or some callback to handle the different events that can occur so that consumers can customize the logging or text that is associated with those events. At that point, if we want an overload that converts the events into a string and emit them to something like Console.WriteLine, then that would be easy to layer on top.

A first draft of pulling out all of the events to be logged and putting them into an interface looks something like this:

public interface IObservableSpy<T>
{
    void OnCreate();
    void OnSubscribeStart(int subscriptionId);
    void OnSubscribeCompleted(int subscriptionId, int activeSubs);
    void OnNext(int subscriptionId, T value);
    void OnError(int subscriptionId, Exception e);
    void OnCompleted(int subscriptionId);
    void OnDownstreamError(int subscriptionId, Exception e);
    void OnDisposeStart(int subscriptionId);
    void OnDisposeCompleted(int subscriptionId, int activeSubs);
}

public static class SpyEx
{
    public static IObservable<T> Spy<T>(this IObservable<T> source, IObservableSpy<T> spy)
    {
        var activeSubscriptionCounter = 0;
        var subscriptionCounter = 0;

        spy.OnCreate();
        return Observable.Create<T>(obs =>
        {
            var subId = Interlocked.Increment(ref subscriptionCounter);

            spy.OnSubscribeStart(subId);
            try
            {
                var subscription = source
                    .Do(x => spy.OnNext(subId, x),
                        ex => spy.OnError(subId, ex),
                        () => spy.OnCompleted(subId))
                    .Subscribe(x =>
                    {
                        try
                        {
                            obs.OnNext(x);
                        }
                        catch (Exception ex)
                        {
                            spy.OnDownstreamError(subId, ex);
                            throw;
                        }
                    }, obs.OnError, obs.OnCompleted);

                return Disposable.Create(() =>
                {
                    spy.OnDisposeStart(subId);
                    subscription?.Dispose();
                    spy.OnDisposeCompleted(subId, Interlocked.Decrement(ref activeSubscriptionCounter));
                });
            }
            finally
            {
                spy.OnSubscribeCompleted(subId, Interlocked.Increment(ref activeSubscriptionCounter));
            }
        });
    }
}

And then a very basic implementation of IObservableSpy could be something like:

public static class LoggingSpyEx
{
    private enum DisposeReason
    {
        Unsubscribe,
        Error,
        Completed
    }

    private class LoggingSpy<T> : IObservableSpy<T>
    {
        private DisposeReason _disposeReason;
        private int _valueCount;

        public LoggingSpy(string? info, Func<T, string>? formatter = null, Action<String>? logger = null)
        {
            Formatter = formatter ?? (t => t?.ToString() ?? "{null}");
            Logger = CreateLogger(logger ?? Console.WriteLine, info ?? $"IObservable<{typeof(T).Name}>");
        }

        private Func<T, string> Formatter { get; }
        private Action<string> Logger { get; }

        private static Action<string> CreateLogger(Action<string> baseLogger, string opName) =>
            msg => baseLogger($"{DateTimeOffset.UtcNow.ToString("HH:mm:ss.fffffff")} [{Environment.CurrentManagedThreadId:X2}] |{opName}| {msg}");

        public void OnCompleted(in SpyInfo info) => Logger($"OnCompleted() [SubId:{info.SubscriptionId}] ({info.ValueCount} Values)");
        public void OnCreate() => Logger("Creating Observable");
        public void OnSubscribeStart(int subscriptionId) => Logger($"Creating Subscription #{subscriptionId}");
        public void OnSubscribeCompleted(int subscriptionId, int activeSubs) => Logger($"Subscription #{subscriptionId} Created ({activeSubs} Active Subscriptions)");
        public void OnNext(int subscriptionId, T value) => Logger($"OnNext() [SubId:{subscriptionId}] (#{Interlocked.Increment(ref _valueCount)}): {Formatter(value)}");
        public void OnError(int subscriptionId, Exception e)
        {
            _disposeReason = DisposeReason.Error;
            Logger($"OnError() [SubId:{subscriptionId}] ({_valueCount} Values) [Exception: {e}]");
        }

        public void OnCompleted(int subscriptionId)
        {
            _disposeReason = DisposeReason.Completed;
            Logger($"OnCompleted() [SubId:{subscriptionId}] ({_valueCount} Values)");
        }

        public void OnDownstreamError(int subscriptionId, Exception e) => Logger($"Downstream Exception [SubId:{subscriptionId}] ({e})");
        public void OnDisposeStart(int subscriptionId) => Logger($"Disposing SubId #{subscriptionId} [{_disposeReason}]");
        public void OnDisposeCompleted(int subscriptionId, int activeSubs) => Logger($"Dispose of SubId #{subscriptionId} Completed ({activeSubs} Active Subs)");
    }

    public static IObservable<T> Spy<T>(this IObservable<T> source, string? infoText = null, Action<string>? logger = null,
                                                                    Func<T, string>? formatter = null) =>
        source.Spy(new LoggingSpy<T>(infoText, formatter, logger));
}

Example Usage:

    IObservable<long> CreateMainObservable() => Observable.Interval(TimeSpan.FromMilliseconds(100)).Take(5);
    IObservable<string> CreateOtherObservable(long n) => Observable.Interval(TimeSpan.FromMicroseconds(50)).Select(x => $"{n}-{x}").Take(5);

    IDisposable cleanup = CreateMainObservable().Spy("Main")
                              .SelectMany(x => CreateOtherObservable(x).Spy($"Child {x}"))
                              .Spy("Combined")
                              .Subscribe();

The only thing that I'm 100% sure about is that this approach has been very useful to me and I feel like others would benefit from it as well. Everything else is up in the air and I'd love to hear feedback and suggestions for improvement.

[JakenVeina]

I'm not convinced that OnCreate() and the Start/Completed variations on OnDisposed and OnSubscribed are particularly valuable, how useful have those been for you? On the flip side, the exposure of subscriptionId across all the notification handlers seems like it has a lot of value. Do you find yourself using that value frequently in practice? If not, then I have to question what value is added with this operator, over just a .Do().

Is there some way we could take the subscriptionId concept and make it global? Or at least global to a particular stream? I.E. have .Spy() calls at multiple points within a stream, where they all publish the same subscriptionId values, when the same downstream subscription is responsible. Maybe with something like .SpyRoot() as the most-upstream call, or .SpyLeaf() as the most-downstream calls, with some shared state behind-the-scenes? Or maybe .BeginSpy() and .EndSpy() pairs? Maybe that could be made to work so long as there's no .Schedule() calls in-between? Having this kind of correlated data would be hugely-beneficial, and could even be used to feed into a UI tool or IDE plugin, to visualize data flows.

[dwcullop]

You're not wrong. In a lot of cases... Do is enough. However, sometimes, especially when trying to make your own operators, you really need to be able to see what is going on at every step. The beauty of the Spy operator is how easy it is to add or remove from a given Rx operation (sometimes in between every operator) to quickly gain the insights you need.

I also intend for this to be a basis for implementing other scenarios. For example, in production, I'll use an overload that takes a Microsoft.Logging.Extension.ILogger instance to emit the values/errors that come through.

All of the elements in there have been useful at some point. I've experimented with other features/events in the past, but this proposal is distilled down to the things that have actually been useful. This is also the superset of events that you might want. An implementation of IObservableSpy is always free to no-op any of them if the information isn't useful for that given situation (the ILogger overload I mentioned, for example, has a flag that controls whether or not to emit the subscription events which is usually false).

Separating OnCreate from OnSubscribe is useful for distinguishing between an IObservable being created and actually being subscribed to. If the events are not flowing, it makes it easy to spot the problem the Subscribe step isn't happening for whatever reason.

Having Start and Completed as separate events is also very useful. In a large chain of operators, there is a lot that can happen between the time you start the subscription and the subscription being completed because of how the subscription call has to work up the chain of operators. If there is a problem with creating a subscription, you can see how far up the chain it got in order to find the source of the problem.
The exact same thing applies to Dispose.

Being able to see subscriptions and disposals has also helped me on numerous occasions when working with sub-observables, like with Switch.

It's also extremely illustrative for those trying to learn Rx see how Observable events flow in different directions for different stages.

For example, consider this toy example:

    using var cleanup = Observable.Interval(TimeSpan.Zero)
                            .Spy("Timer")
                            .Select(x => x)
                            .Spy("Select 1")
                            .Select(x => x)
                            .Spy("Select 2")
                            .Take(1)
                            .Spy("Take 1")
                            .Select(x => x)
                            .Spy("Select 3")
                            .Select(x => x)
                            .Spy("Select 4")
                            .Subscribe();

Which produces output like:

17:41:50.2033919 [01] |Timer| Creating Observable
17:41:50.2036884 [01] |Select 1| Creating Observable
17:41:50.2037211 [01] |Select 2| Creating Observable
17:41:50.2037436 [01] |Take 1| Creating Observable
17:41:50.2037599 [01] |Select 3| Creating Observable
17:41:50.2037753 [01] |Select 4| Creating Observable
17:41:50.2052791 [01] |Select 4| Creating Subscription #1
17:41:50.2054394 [01] |Select 3| Creating Subscription #1
17:41:50.2054656 [01] |Take 1| Creating Subscription #1
17:41:50.2054865 [01] |Select 2| Creating Subscription #1
17:41:50.2055116 [01] |Select 1| Creating Subscription #1
17:41:50.2055300 [01] |Timer| Creating Subscription #1
17:41:50.2061543 [01] |Timer| Subscription #1 Created (1 Active Subscriptions)
17:41:50.2062176 [01] |Select 1| Subscription #1 Created (1 Active Subscriptions)
17:41:50.2062415 [01] |Select 2| Subscription #1 Created (1 Active Subscriptions)
17:41:50.2063234 [01] |Take 1| Subscription #1 Created (1 Active Subscriptions)
17:41:50.2063613 [01] |Select 3| Subscription #1 Created (1 Active Subscriptions)
17:41:50.2063757 [01] |Select 4| Subscription #1 Created (1 Active Subscriptions)
17:41:50.2070772 [68] |Timer| OnNext() [SubId:1] (#1): 0
17:41:50.2078248 [68] |Select 1| OnNext() [SubId:1] (#1): 0
17:41:50.2079012 [68] |Select 2| OnNext() [SubId:1] (#1): 0
17:41:50.2079358 [68] |Take 1| OnNext() [SubId:1] (#1): 0
17:41:50.2079789 [68] |Select 3| OnNext() [SubId:1] (#1): 0
17:41:50.2080190 [68] |Select 4| OnNext() [SubId:1] (#1): 0
17:41:50.2082572 [68] |Take 1| OnCompleted() [SubId:1] (1 Values)
17:41:50.2083181 [68] |Select 3| OnCompleted() [SubId:1] (1 Values)
17:41:50.2083402 [68] |Select 4| OnCompleted() [SubId:1] (1 Values)
17:41:50.2099776 [68] |Select 4| Disposing SubId #1 [Completed]
17:41:50.2106196 [68] |Select 3| Disposing SubId #1 [Completed]
17:41:50.2106593 [68] |Take 1| Disposing SubId #1 [Completed]
17:41:50.2107118 [68] |Select 2| Disposing SubId #1 [Unsubscribe]
17:41:50.2107373 [68] |Select 1| Disposing SubId #1 [Unsubscribe]
17:41:50.2107602 [68] |Timer| Disposing SubId #1 [Unsubscribe]
17:41:50.2111042 [68] |Timer| Dispose of SubId #1 Completed (0 Active Subscriptions)
17:41:50.2112682 [68] |Select 1| Dispose of SubId #1 Completed (0 Active Subscriptions)
17:41:50.2113026 [68] |Select 2| Dispose of SubId #1 Completed (0 Active Subscriptions)
17:41:50.2113199 [68] |Take 1| Dispose of SubId #1 Completed (0 Active Subscriptions)
17:41:50.2113381 [68] |Select 3| Dispose of SubId #1 Completed (0 Active Subscriptions)
17:41:50.2113785 [68] |Select 4| Dispose of SubId #1 Completed (0 Active Subscriptions)

It is easy to see that:

• Create goes from top to bottom
• Subscribe Start goes from bottom to top
• Subscribe Complete goes from top to bottom
• OnNext goes from top to bottom
• OnCompleted starts with "Take 1" and then goes down (but not up)
• Dispose Start goes from bottom to top
• Dispose Complete goes from top to bottom

Such things are obvious to those who have worked with Rx for a while, but for new people, it can be, as I said, quite illuminating.

[dwcullop]

Is there some way we could take the subscriptionId concept and make it global? Or at least global to a particular stream? I.E. have .Spy() calls at multiple points within a stream, where they all publish the same subscriptionId values, when the same downstream subscription is responsible. Maybe with something like .SpyRoot() as the most-upstream call, or .SpyLeaf() as the most-downstream calls, with some shared state behind-the-scenes? Or maybe .BeginSpy() and .EndSpy() pairs? Maybe that could be made to work so long as there's no .Schedule() calls in-between? Having this kind of correlated data would be hugely-beneficial, and could even be used to feed into a UI tool or IDE plugin, to visualize data flows.

That's a very interesting idea. I do see how that would be useful, but it could be tricky because operators are usually independent with no shared state between them. Perhaps you could do something like:

// Holds shared state needed to coordinate tracing the entire stream sequence
class StreamSpy(string Name)
{
    // Returns an implementation that is specific to one operator in the stream
    public IObservableSpy<T> OpSpy<T>(string name) { /* ... */ }
}

IObservable<long> GetObservable()
{
    var streamSpy = new StreamSpy("Main Stream");
    
    return Observable.Interval(TimeSpan.Zero)
            .Spy(streamSpy.OpSpy<long>("Interval"))
            .Select(x => x)
            .Spy(streamSpy.OpSpy<long>("Select"));
}

The interface is kind of rough, but you get the general idea: You could put the global counter or whatever else you wanted in the StreamSpy class and use it to implement whatever stream-level tracing you wanted! That could be very useful for sure!

However, I think that is just a layer that would be built on top of what I'm proposing. Is there anything missing in the base interface that you would need to build what you're thinking?