Reactive: Awaiting Subject.OnNext

[robertmclaws]

I'm using RX + Azure WebJobs to build a simple queue-based MessageBus. I have dozens of "events" that can be fired based on incoming and outgoing data, and multiple handlers, implementing IObserver, can subscribe to the same event.

It all works great, but I'm having some concurrency issues that are hard to track down. I suspect it's because most of my implementing methods are async, but my RX call is not. When I call subject.OnNext(item), I want a clean way for all of my subscribers to be able to call Async methods all the way up the chain. Ideally, I'd be able to call await subject.OnNextAsync(item) and have everything structured the way I have it now.

Is there a better way to handle this? If not built in, something I can build with an extension method, or something like that?

Thanks!

[LeeCampbell]

I think this is one for StackOverflow.com if you post a reproduce-able sample there, you will no doubt get some quick and helpful responses.

[robertmclaws]

I don't believe StackOverflow is the answer here. What I'm proposing is that there needs to be a new function that can handle the fact that it is possible to call async code within OnNext. The following needs to be added to SubjectBase

public abstract async Task OnNext(T value);

And the following needs to be added to Subject and AsyncSubject:

public override async Task OnNext(T value);

That would allow for OnNext calls to be awaitable.

[shiftkey]

I can't speak for the team, but given how this impacts the IObserver<T> contract - and the potential flow-on effects, I'd like for this to be done as a proposal so we can consider things in a holistic way.

@advancedrei do you have an example of the code you've written to work around this issue?

[RxDave]

Calling asynchronous functions in a sequence is what the "sequential composition" operator of the magical "Continuation Monad" is for; a.k.a, Rx's SelectMany operator. Is it possible that using subjects is simply the wrong way to solve your problem?

c# from e in event from x in CallAsync(e) from y in CallNextAsync(x) from z in CallLastAsync(y) select new { x, y, z }

Note that when building the code currently in GitHub, though not v2.2.5, adding using System.Reactive.Threading.Tasks allows you to add .ToObservable(scheduler) to each *Async method that you're composing if you want to control concurrency; otherwise, the TPL will most likely introduce concurrency between each from clause. I'm not sure whether or not the v2.3 beta includes the ToObservable overload that accepts an IScheduler.

[robertmclaws]

@shiftkey From the interface perspective, you're right - I think maybe a new subject type, like "AwaitableSubject", and a new interface ("lAwaitableObserver" or something like that) may be in order.

@RxDave I think you might be thinking that the purpose of this is to "control" concurrency. My purpose is simply to be able to use Async methods (which are mostly the norm in .NET moving forward) inside RX events, without hacks.

OK, so here's the architecture. I'll try to explain it well and keep it succinct. We have built what we internally call a "Simple Service Bus" (or SBB for short), based on Azure Queue Storage, the WebJobs SDK, and RX. (This was built about a year before the ASP.NET Webhooks SDK and Logic Apps, so I'm well aware of those solutions. I have no interest in migrating for now).

As events are generated by our own apps & APIs, as well as from 3rd party inbound webhooks, we have our SBB publish an event onto the queue. The way it is designed, I can do so without the publisher (and thereby the endpoints) taking a dependency on any 3rd party libraries except Azure Storage.

So then in the Core of our system, we have a Dispatcher assembly that lives inside the WebJob and does the dirty work. MessageDispatcher is IObservable, so it observes each message coming off the Queue from the WebJobs SDK, and manages a set of MessageHandlers, which are IObservers. Those MessageHandlers are instanced-up by our IoC container, and the MessageDispatcher constructor grabs those MessageHandlers and subscribes them by the System.Types they handle.

We have a WebJobs QueueTrigger function that calls Dispatcher.Dispatch(message). That function looks like this:

        public void Dispatch(MessageEnvelope messageEnvelope)
        {
            _subject.OnNext(messageEnvelope);
        }

INSIDE the individual OnNext implementations, however, I have to call code that is often Async. My hacks often have to use Sync workarounds to function, like this actual code, which uses NotificationHubs to send push notifications to my phone:

        internal bool SendChargeFailedNotification(ChargeFailed chargeFailed, TextWriter log, int retryCount = 0)
        {
            try
            {
                var toast = GetNewToast();
                var stripe = chargeFailed.GetCharge();
                toast.Visual.BodyTextLine1.Text = "Charge Failed! (" + (stripe.Amount / 100).ToString("C2") + ")";
                toast.Visual.BodyTextLine2 = new ToastText { Text = stripe.Description };
                ///TOTALLY AWKWARD SYNC HACK BELOW
                var result = HubClient.SendWindowsNativeNotificationAsync(toast.GetContent(), "admin").Result;
                log.WriteLine($"AdminPushNotification Outcome: {result.State}");
                return WasNotificationSuccessful(result);
            }
            catch (Exception ex)
            {
                LogError(log, ex, new Dictionary<string, string>
                {
                    ["StripeEvent"] = chargeFailed.StripeData,
                })
                ///TOTALLY AWKWARD SYNC HACK BELOW (THAT C# 6 CAN NOW HANDLE)
                .Wait();
                throw ex;
            }
        }

I would like to be able to create these internal helper functions as Async methods, so they can be called inside an awaited AwaitableSubject.OnNextAsync(), and have the .NET Framework work its magic. That way, subscribers still fire in order, but the internals can still call awaitable code properly.

I hope that helps. Please let me know if you have any questions.

[RxDave]

The overloads of SelectMany that I'm using in my example accept projections that return Task<T>. Have you seen them?

I'm merely suggesting that piecing together the computation by injecting subjects may not be the best design. If instead you simply stick to the functional world of operator composition, SelectMany overloads that accept Task<T> projections do the job nicely.

[robertmclaws]

@RxDave While I admit that I am a relative n00b in the RX space, it is difficult for me to see how the example you are using is relevant to me, given the explanation I gave of our architecture. By the example given, you are assuming the operations are known at compile time, therefore I can "just code in the calls manually". Which "OnNext" gets called for which Type is dynamic and can change at runtime, based on how the MessageHandlers are registered, and the incoming CLR types they are supporting.

While I can see the possibility of using a LINQ query on the registered IMessageHandlers for a given type, looping through the collection, and calling my own awaitable OnNext... at that point, why would I even need Observables, because the Azure Storage Queue + WebJobs are doing the work of "reacting" for us?

I trust the person that designed the code that we're using, and respectfully ask that you consider that neither one of us are in the best position to discuss which approach is the "best" one. If you had the time, I'd love to show you the complete code, and get your opinion on what might be better. But I assume you have better things to do, and if I _were_ to rewrite it, I see a distinct possibility if just using Azure Logic Apps and removing our dependence on RX altogether.

Regardless of opinions on whether our specific architecture is "best", the design of RX as currently implemented does not allow you to call Async code properly inside a Subscriber's OnNext. As Async is the only possible option for certain things (like ModernHttpClient), the currently-shipping architecture is no longer relevant in the modern development world, and IMHO needs to be updated.

Thanks!

[LeeCampbell]

@advancedrei I can see how you arrive to this conclusion

the currently-shipping architecture is no longer relevant

but how you arrived at that conclusion is IMO incorrect.
So let me see if I can help lead you down a path, that may help you see why this is not what you actually need.

As @RxDave suggested, you should be using SelectMany (or another compositional method) to allow your execution of the Async methods.
i.e. what you are putting in your subscribe method belongs in your query chain instead.

just using Azure Logic Apps and removing our dependence on RX altogether.

This sounds like it probably is a good plan.

Trying to keep on topic, your code you provide as an example (appreciated), shows some other interesting features:

• You throw in your OnNext handler SendChargeFailedNotification
• SendChargeFailedNotification returns a bool

These things are key to telling me that they belong in the query chain

• possibility of failure,
• asynchronous requirement
• and has a return value

Let us consider each of these.

If your OnNext Hander throws, this will bring down your Rx subscription and have unknown consequences in your application. The guidance is that your should not throw in your OnNext handlers. They are just there to be notified that an event happened.

If your OnNext handler is doing so much work that it has a requirement to be asynchronous then you can consider the work it is doing to be an asynchronous sequence with a single event. This then allows it be part of the query chain (composition of asynchronous events). Alternatively the source event can be be passed off to another processor e.g. put in a queue for someone else to process in a blocking, long running or asynchronous manner.

If your OnNext handler needs to return a value, then it is clearly not the end of the chain or the _sink_.
In this case it is a candidate for functional composition.
Specifically the function here translates a ChargeFailed value to a bool.

These three things seem like a compelling reason to look at asynchronous functional composition, with is what Rx excels at.

From the information I have, I would suggest that instead of calling the method provided above in the OnNext Handler, it should be modified to something like the code below and moved into the query.

internal IObservable<bool> SendChargeFailedNotification(ChargeFailed chargeFailed, TextWriter log, int retryCount = 0)
{
    return Observable.Create<bool>(observer =>
    {
        try
        {
            var toast = GetNewToast();
            var stripe = chargeFailed.GetCharge();
            toast.Visual.BodyTextLine1.Text = "Charge Failed! (" + (stripe.Amount / 100).ToString("C2") + ")";
            toast.Visual.BodyTextLine2 = new ToastText { Text = stripe.Description };
        catch (Exception ex)
        {
            return LogError(log, ex, new Dictionary<string, string>
            {
                ["StripeEvent"] = chargeFailed.StripeData,
            }).ToObservable()
            .Subscribe(_=>observer.OnError(ex));

        }

        return HubClient.SendWindowsNativeNotificationAsync(toast.GetContent(), "admin")
            .ToObservable()
            .Catch(ex =>
            {
                return LogError(log, ex, new Dictionary<string, string>
                {
                    ["StripeEvent"] = chargeFailed.StripeData,
                }).ToObservable()
                .IgnoreElements()
                .Select(_=>false)   //Convert what I assume is a Task -> Unit -> Bool
                .Concat(Observable.Throw<bool>(ex))
            })
            .Subscribe(result =>
                {
                    log.WriteLine($"AdminPushNotification Outcome: {result.State}");
                    var value = WasNotificationSuccessful(result);
                    observer.OnNext(value);
                },
                ex => observer.OnError(ex));    
    });
}

It obviously still has plenty of room for improvement.
The logging code, if it really is asynchronous and is pervasive in the code base with Rx, then this is a clear candidate for some helper method.
Logging and Rx has been thrashed out plenty, so I wont go into it here.

There is a retryCount value passed in to the method, that is not used.
Again, Retry is something that Rx can do for you and also something that would then belong in the query chain.

I hope this helps you with your work, and also helps lead you to why people are against the introduction of an async OnNext handler method.

[robertmclaws]

I very much appreciate the super-thoughtful method with which you approached this (no BS or sarcasm here). I will read this over several times tomorrow AM, reflect on it a bit, and get back with you at some point later in the day. Seriously, thanks.

[robertmclaws]

OK, so I've read over this. I really appreciate the time you've taken to explain your viewpoint, and write code that you think will help.

I think the key element you are missing is "what" I'm trying to observe. I'm observing messages coming off a queue. I don't need to observe any of the other stuff. I have no issue if the control flow takes a while, because the event I'm observing _already happened_, and the WebJobs SDK already has a way to handle transient issues.

I throw exceptions inside these individual helper functions because that is how the WebJobs system needs to function in order to drop the message into a "poison" queue to get handled as an error. The application functions very predictably at that point, it shuts down and restarts. I am fine with that, because the whole point of our architecture is to process incoming events from different sources out of band from our web APIs.

To be frank, here's the part I don't understand... if I was using the most common RX use case I've seen: observing click events in an app, and using multiple subscriptions to those events to process various actions, one of them could be an instrumentation subscription that reported back to Application Insights. In that case, I would still have the exact same problem, because AI APIs are asynchronous, and this RX one is not.

And while I really appreciate that you tried to help give me "what I actually need", as you put it, your code is twice as long, convoluted, and unnecessary. I don't see how making my code twice as long and convoluted is any better of a solution than just letting me able to call await AwaitableSubject.OnNext(item), and use the exact same code I'm using today (because that code is clean, it's awesome, and it works well. It's just not what you're used to).

I'm really trying to be constructive here, but I think maybe you guys are so used to using RX a certain way that you aren't seeing the forest for the trees.

Thanks for your time.

[paulpdaniels]

I would have to agree with @shiftkey here. It isn't a simple matter of adding another Subject type, I think you would need to re-implement all of the operators to support this new set of methods. Moreover, I'm not sure I entirely understand what kind of semantics you would expect from it, since it would likely break down at the first sign of a scheduler.

_subject.observeOn(Scheduler.Default).Subscribe(x => Console.WriteLine(x));

//What is being awaited here? The task to be scheduled or the OnNext block to be run?
await _subject.OnNextAsync(event);

I'll throw out another suggestion which is to try moving your logic down stream. Right now it would seem that you have an Observer from which you are trying to handle all the logic within the OnNext callback. I would suggest that instead of simply attaching an Observer during your handler hook up phase, that you inject a pipeline.

As a slightly nonsensical example would be:

class MessageHandlerImpl : MessageHandler<T> {

  public IDisposable OnHookUp(IObservable<T> source) {
    return source.SelectMany(async x => await SendChargeFailedNotificationAsync(x, this.textWriter))
                 .Subscribe(result => HandleResultOfWorkSync(x));
  }

}

// In your MessageDispatcher hookup logic
handlerInstance.OnHookUp(_subject);


//Then this works again 

_subject.OnNext(messageEnvelope);

[robertmclaws]

@paulpdaniels I really appreciate your additions to the discussion. The short answer from the question in your first example is that you're awaiting the code inside the OnNext block. I think that question for the scheduler could be solved with some kind of hinting, and the question for the end-developer could be solved with clear documentation.

I have to admit, initially I didn't like the second example. But the more I think about it, redesigning how the messages are dispatched might help. Would you happen to be available for a few hours to consult on the project?

Thanks!

[bartdesmet]

A parallel universe of Rx with async signatures can definitely be envisioned; in fact, we've built a partial implementation of this for a service here internally, where query operators may have to call out to async functions, hence requiring the observer contract (etc.) to be async as well. The drawback is the big oil spill it creates over the entire operator surface (the same holds for Ix-Async), so more design work is needed. When I find some cycles here, I'll try to distill the essence of our implementation and share it here.

[robertmclaws]

FINALLY!!! I'm not crazy after all! 😀

@bartdesmet Do you envision that as something that could be an add-on package to the non-async version? Or do you see it as something being a part of the core implementation/NuGet package?

[dominikjeske]

Is there any progress in this topic?

[quinmars]

@bigdnf, meanwhile, @bartdesmet has published AsyncRx.NET. To my knowlege there is not any nuget package released for it and any active (at least public) development on it.

[dominikjeske]

It would be nice if someone from MS give some info about roadmap for that feature. I can't imagine Rx without full async/await support.

[robertmclaws]

Honestly, given how difficult it was to convince anyone that this was necessary, I just pulled RX.NET from my codebase and moved on. As cool as it was, it was overcomplicating my use case.

[dominikjeske]

shame, shame, shame ;)

[akarnokd]

Why would you await OnNext? Do you expect some reliable delivery mechanism? Rx is about an unidirectional fire-and-forget by default.

[robertmclaws]

I explained it extensively in this thread and have no interest in rehashing it. In the nearly 3 years since this was last discussed, we've moved even closer to an async-first world. I'm surprised (but somehow not) that the use case for Async even needs to be explained.

[dominikjeske]

For me it is simple - current codebase have async/await in many places and I cannot use them right now. Of course I can use " GetFooAsync(...).GetAwaiter().GetResult();" when I'm using some async code but I would prefer not to do this.

[grahamehorner]

I would be interested in the async OnNext discussed here; and I'm willing to help. as a side and IMHO I think any implementation of an async OnNext should be able to accept a cancellation token so that the observerable many cancel any in progress OnNext calls.

eg. async interfaces

IAsyncSubject<T> : ISubject<T>
IAsyncObserver<T> : IObserver<T> 
IAsyncObservable<T> : IObservable<T>
IAsyncObservableSubscription<T> : IDisposable 

[ericsink]

I find the discussion here interesting. I too have found myself wanting to await OnNext(), but in my case the loop always goes like this:

1. Reach a point where I wish I could await OnNext().
2. Remember that Task and IObservable are two ways of doing (roughly) the same thing (a stream of future items) where Task limits the "stream" to 1 item.
3. Think about how weird it would look for OnNext() have a return type of IObservable, which, because of step 1, should be just as sensible as Task.
4. Pause and wonder if that weirdness might mean that my wanting to await OnNext() is actually something unnatural, arising from my lack of understanding.
5. Set the whole issue aside and find a way to not need to await OnNext().
6. Work for a while until I forget about this stuff.
7. Goto step 1.

[akarnokd]

async interfaces
IAsyncObserver<T> : IObserver<T>

This doesn't work. It conflates two distinct flow behavior into one, forcing the implementation to double on content and requires extra work due cross-concept defensive programming overhead. The independent Async-Ix does work.

limits the "stream" to 1 item

Again, the Async-Ix design is much more suitable for this type of processing.

[ericsink]

"the Async-Ix design is much more suitable for this type of processing"

Agreed. If it's not clear, my silly "loop" was trying to say that while I do understand how the desire for async Task<T> OnNext() arises, it seems to me like the wrong approach, an attempt to make Rx into something it is not.

[robertmclaws]

It's seriously been 4 years since I posted this and people are still arguing about whether or not this is the right way to do things VS just building the effing feature because people need it. Because what really happens in @eriksink's loop is:

1. Think RX might be the right solution to a particular problem.
2. Build something that is reasonably close to functional.
3. Hit a wall because you need to call code you can't control from a dispatcher and don't want to call GetAwaiter().GetResult().
4. Have a BS argument with people that are too smart for their own good on whether or not you're 100% in the lane they envisioned for the software or you should have the freedom to experiment and push the boundaries.
5. Rip out the code because it's not worth your time trying to be cute when a simple For loop will probably suffice anyway.

If your goal is to get people to use your software... you're doing a bang-up job, folks. I didn't realize this was an offshoot of the Linux project. Linus would be proud.

[ericsink]

Wow -- didn't mean to trigger sarcasm and anger. My apologies.

[robertmclaws]

sigh no I'm sorry, didn't mean to be salty. I invested hours of my time explaining the use case and digging through the architecture like 2 years ago, and had the same type of discussion then. Nothing apparently got done (except for someone shipped some code but not a NuGet package, which is essentially the same as shipping nothing) and seemed like the same debate was starting up. The use-cases that were valid for needing it to be threadpool-friendly are still valid today, whether people see them or not. Failure to adapt means that it's just not going to get used... and I doubt I'm the first or last person to have had that experience with this framework.

[ericsink]

No worries here.

I do find the tradeoffs around this API design to be an interesting discussion, but I never like to carelessly or ignorantly stomp through territory which is emotionally charged.

Cheers.

[danielcweber]

@robertmclaws While I greatly appreciate your effort to get your idea over, I have now started too to resent your tone. Especially, you got the following wrong:

If your goal is to get people to use your software... you're doing a bang-up job, folks.

Speaking just for myself: No, it is not the objective of my employer (which is not Microsoft) to allow me to contribute to open source to get people to use "our software" (the "our" being a weak concept in OSS anyway IMO) or so that @robertmclaws will eventually have what he so direly needs. I don't remember your paychecks coming in.

@bartdesmet already did some work on async variants of the interfaces and implementations. Check the license agreement of this repo and fork it, make it happen.

[grahamehorner]

async interfaces
IAsyncObserver<T> : IObserver<T>

This doesn't work. It conflates two distinct flow behavior into one, forcing the implementation to double on content and requires extra work due cross-concept defensive programming overhead. The independent Async-Ix does work.

limits the "stream" to 1 item

Again, the Async-Ix design is much more suitable for this type of processing.

@ericsink not sure I’ve expressed my thoughts on this currently given the above response.
my thoughts are simply the any developer should be able to use the IObserver or IAsyncObserver to provide a sync or async task implementation the base class of AsyncObsever can simply provide a Function which is wrapped into a task and returns a T when completed.

[theodorzoulias]

@robertmclaws would you consider as a solution an overload of the Subscribe method that accepts asynchronous onNext actions?

public static IDisposable Subscribe<T>(this IObservable<T> source,
    Func<T, Task> onNext, Action<Exception> onError, Action onCompleted)
{
    return source.Select(async x =>
    {
        await onNext(x);
        return Unit.Default;
    })
    .Merge()
    .Subscribe(_ => { }, onError, onCompleted);
}

This implementation notifies all subscribers concurrently, and each subscriber is also notified concurrently about new values. So if for example the source sequence emits 10 values in quick succession, and there are 10 observers with long-running asynchronous onNext subscriptions, it is possible that all 100 tasks will be up and running concurrently. Is this behavior desirable?