Bdtjtk

I am using Observable / reactive extensions to debounce certain events, like button clicks or entering text into a textbox. However, in the event of a shutdown or close, I need to await any pending events so that save operations can complete, etc.

The following code will deadlock.

Button b1 = new Button();



var scheduler = new EventLoopScheduler(ts => new Thread(ts)

{

    IsBackground = false

});



var awaiter = Observable.FromEventPattern(h => b1.Click += h, h => b1.Click -= h, scheduler)                

     .Throttle(TimeSpan.FromMilliseconds(5000), scheduler)

     .FirstOrDefaultAsync();



someTaskList.add(awaiter.ToTask());



awaiter.Subscribe

(

    x =>

    {

        //do some work in response to click event

    }

);



//program continues...

Then, elsewhere in the application

private async Task CloseApplicationSafely()

{

    await AwaitPendingEvents();

}



private async Task AwaitPendingEvents()

{

    if(someTaskList.Count > 0)

    {

        await Task.WhenAll(someTaskList);

    }

}

The program will then deadlock, awaiting forever if a button click has never occurred. Here is another example, but with a textbox.

var completedTask = Observable.FromEventPattern(h => t1.TextChanged += h, h => t1.TextChanged -= h, scheduler)

    .Select(x => ((TextBox)x.Sender).Text)

    .DistinctUntilChanged()

    .Throttle(TimeSpan.FromMilliseconds(5000), scheduler)

    .ForEachAsync(txt =>

    {

        //do some work, save the text

    });



someTaskList.Add(completedTask);

In this case it doesn't matter if text was ever changed or not. The variable completedTask will deadlock forever if you await it. ForEachAsync() returns a task, which seems to never be activated.

What am I doing wrong? Hopefully my intended function is clear. I am debouncing events. But I need to await any pending events that are in the process of being debounced to ensure they complete. And if there are no pending events, continue without waiting. Thanks.

Comments from both @Servy and @Enigmativity helped me pin it down. For those interested, here is the solution I came up with. Any suggestions on my approach let me know.

I created a static helper class named WaitableEventHelper, which includes the following function.

public static Task WaitableDebouncer(

    this Control c, 

    Action<EventHandler> addHandler, 

    Action<EventHandler> removeHandler, 

    IScheduler scheduler,

    CancellationToken cancelToken,

    TimeSpan limit,

    Func<Task> func)

{

    var mycts = new CancellationTokenSource();



    bool activated = false;

    bool active = false;



    Func<Task> pending = null;



    var awaiter = Observable.FromEventPattern(addHandler, removeHandler, scheduler)

        .TakeUntil(x => { return cancelToken.IsCancellationRequested; })

        .Do((x) => { activated = true; })

        .Do((x) =>

        {

            //sets pending task to last in sequence

            pending = func;

        })

        .Throttle(limit, scheduler)

        .Do((x) => { active = true; })    //done with throttle

        .ForEachAsync(async (x) =>

        {

            //get func

            var f = pending;



            //remove from list

            pending = null;



            //execute it

            await f();



            //have we been cancelled?

            if (cancelToken.IsCancellationRequested)

            {

                mycts.Cancel();

            }



            //not active

            active = false;



        }, mycts.Token);



    //if cancelled 

    cancelToken.Register(() => 

    {

        //never activated, force cancel

        if (!activated)

        {

            mycts.Cancel();

        }



        //activated in the past but not currently active

        if (activated && !active)

        {

            mycts.Cancel();

        }

    });



    //return new awaiter based on conditions

    return Task.Run(async () =>

    {

        try

        {

            //until awaiter finishes or is cancelled, this will block

            await awaiter;

        }

        catch (Exception)

        {

            //cancelled, don't care

        }



        //if pending isn't null, that means we terminated before ForEachAsync reached it

        //execute it

        if (pending != null)

        {

            await pending();

        }

    });

}

I then use it like this. Here is an example with a button click, b1 is a System.Windows.Forms.Button object. This could be anything. For my test app I was changing the colors in some panels on the main form. Per previous code in OP, tasks is just a List of type Task.

var awaiter1 = b1.WaitableDebouncer(h => b1.Click += h, h => b1.Click -= h, 

    scheduler, 

    canceller.Token, 

    TimeSpan.FromMilliseconds(5000), 

    async () =>

    {

        Invoke(new Action(() =>

        {

            if (p1.BackColor == Color.Red)

            {

                p1.BackColor = Color.Orange;

            }

            else if (p1.BackColor == Color.Orange)

            {

                p1.BackColor = Color.Yellow;

            }

            else if (p1.BackColor == Color.Yellow)

            {

                p1.BackColor = Color.HotPink;

            }

            else

            {

                p1.BackColor = Color.Red;

            }

        }));

    });



tasks.Add(awaiter1);

Another for a TextChanged on a text box. t1 is System.Windows.Forms.TextBox. Again this could be anything, I'm just setting a static someValue string variable and updating a label on UI.

var awaiter2 = t1.WaitableDebouncer(h => t1.TextChanged += h, h => t1.TextChanged -= h, 

    scheduler, 

    canceller.Token, 

    TimeSpan.FromMilliseconds(5000), 

    async () =>

    {

        savedValue = t1.Text;



        Invoke(new Action(() => l1.Text = savedValue));

    });



tasks.Add(awaiter2);  

Then this is what termination or shutdown looks like. This could be an application closure, or a file closure. Just some event where we need to unbind these events but save any pending work that has been initiated by the user before doing so. Imagine user typing into the textbox then quickly hitting X to close the app. The 5 seconds has not run out yet.

private async Task AwaitPendingEvents()

{

    if (tasks.Count > 0)

    {

        await Task.WhenAll(tasks);

    }            

}

We have an app-wide waiting routine. On close we do.

//main cancel signal

canceller.Cancel();



await AwaitPendingEvents();

So far with my tests it appears to work. If no event has ever been generated, it will cancel. If an event has been generated we then look to see if there is any pending work that hasn't made it through the throttle yet. If so, we cancel the observable and execute that pending work ourselves so we don't have to wait on the timer. If there is pending work and we've already made it through the throttle, then we just wait and let the observable subscription finish executing it. The subscription then cancels itself after execution if a cancellation has been requested.