Using RX for exception retrying

Question

I have an interesting pattern I'm trying out, using RX extensions for exception retrying. I have the following code.

Public Async Function InitializeCommunications() As Task
    Await Observable.Create(Of Unit)(Async Function(observer)
                                         MessageBox.Show("Please press reset")
                                         Await Me.Cockpit.LoadDriverToPLC()
                                         MessageBox.Show("Please press start")
                                         observer.OnNext(Unit.Default)
                                     End Function).Retry()
End Function

The code works almost perfectly. If there is an exception thrown in LoadDriverToPLC then the sequence is resubscribed to and the body rerun and the user is asked to press reset again.

There is only one problem with this code. The first time through the MessageBox calls are correctly modal. This means they ride above the main window and you can't hide them by accidently clicking on the main window. However if an exception is thrown and the body is retried then the MessageBox calls are no longer modal.

I'm sure this is to do with the RX schedular. What is the trick to make sure the body always runs in the same context?

I intend to wrap this pattern later on when I can get it reliable into something like

Public Async Function InitializeCommunications() As Task
    Await RetryOn(of Exception)(Async Sub()
                                         MessageBox.Show("Please press reset")
                                         Await Me.Cockpit.LoadDriverToPLC()
                                         MessageBox.Show("Please press start")
                                End Sub)
End Function

Answer 1

The solution seems to be to capture the synchronization context and force the observer to be notified on the correct context so it then retries using the same context

Public Async Function InitializeCommunications(msgFeedback As Func(Of String, Task)) As Task
    Dim context = SynchronizationContext.Current
    Await Observable.Create(Of Unit)(
             Async Function(observer)
                 MessageBox.Show("Please press reset")
                 Await Me.Cockpit.LoadMessageLoopToPLC(111)
                 MessageBox.Show("Please press start")
                 observer.OnNext(Unit.Default)
             End Function).ObserveOn(context).Retry()
End Function

Another, maybe better, way to do this is to create some extensions methods on Object like so

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace FunctionalExtensions
{
    public static class ObjectMixins
    {
        public static T Retry<Ex, T>(this Object This, int count, Func<T> action)
        where  Ex : Exception
        {
            while (true)
            {
                try
                {
                    return action();
                }
                catch(Ex)
                {
                    if (count==0)
                    {
                        throw;
                    }
                    count--;
                }
            }
        }

        public static async Task<T> Retry<Ex, T>(this Object This, int count, Func<Task<T>> action)
        where  Ex : Exception
        {
            while (true)
            {
                try
                {
                    return await action();
                }
                catch(Ex)
                {
                    if (count==0)
                    {
                        throw;
                    }
                    count--;
                }
            }
        }

        public static void Retry<Ex>(this Object This, int count, Action action)
        where  Ex : Exception
        {
            This.Retry<Ex, bool>(count, () => { action(); return true; });
        }

        public static async Task Retry<Ex>(this Object This, int count, Func<Task> action)
        where  Ex : Exception
        {
            await This.Retry<Ex, bool>(count, async () => { await action(); return true; });
        }

        public static void Retry<Ex>(this Object This, Action action)
        where  Ex : Exception
        {
            This.Retry<Ex, bool>(() => { action(); return true; });
        }

        public static T Retry<Ex, T>(this Object This, Func<T> action)
        where Ex : Exception
        {
            while (true)
            {
                try
                {
                    return action();
                }
                catch(Ex)
                {
                }
            }
        }

    }
}

which supports synchronous and asynchrous operation. You can use it like below to retry 10 times before throwing the exception

Public Async Function InitializeCommunications(msgFeedback As Func(Of String, Task)) As Task
    Await Retry(Of Exception)(10, 
        Async Function()
          MessageBox.Show("Please press reset on the NUM control unit then press ok here.")
          Await Me.Cockpit.LoadMessageLoopToPLC(111)
          MessageBox.Show("Please press start on control unit.")
       End Function)
End Function