Raise an event with the most recent data after some time

Question

In my WPF application the data that UI displays will be updated too frequently. I figured out that it will be great to leave the logic intact and solve this issue with an extra class that stores the most recent data and raises the update event after some delay.

So the goal is to update UI, lets say every 50 ms, and display the most recent data. But if there is no new data to show, then the UI shan't be updated.

Here is an implementation I have created so far. Is there a way to do it without locking? Is my implementation correct?

class Publisher<T>
{
    private readonly TimeSpan delay;
    private readonly CancellationToken cancellationToken;
    private readonly Task cancellationTask;

    private T data;

    private bool published = true;
    private readonly object publishLock = new object();

    private async void PublishMethod()
    {
        await Task.WhenAny(Task.Delay(this.delay), this.cancellationTask);
        this.cancellationToken.ThrowIfCancellationRequested();

        T dataToPublish;
        lock (this.publishLock)
        {
            this.published = true;
            dataToPublish = this.data;
        }
        this.NewDataAvailable(dataToPublish);
    }

    internal Publisher(TimeSpan delay, CancellationToken cancellationToken)
    {
        this.delay = delay;
        this.cancellationToken = cancellationToken;
        var tcs = new TaskCompletionSource<bool>();
        cancellationToken.Register(() => tcs.TrySetCanceled(), useSynchronizationContext: false);
        this.cancellationTask = tcs.Task;
    }

    internal void Publish(T data)
    {
        var runNewTask = false;

        lock (this.publishLock)
        {
            this.data = data;
            if (this.published)
            {
                this.published = false;
                runNewTask = true;
            }
        }

        if (runNewTask)
            Task.Run((Action)this.PublishMethod);
    }

    internal event Action<T> NewDataAvailable = delegate { };
}

Answer 1

I'd suggest that you don't reinvent the wheel. The Microsoft Reactive Framework handles this situation super easily. The reactive framework allows you to turn events into linq queries.

I'm assuming that you're trying to call DownloadStringAsync and therefore need to handle the DownloadStringCompleted event.

So first you have to turn the event into an IObservable<>. That's easy:

var source = Observable
    .FromEventPattern<
        DownloadStringCompletedEventHandler,
        DownloadStringCompletedEventArgs>(
        h => wc.DownloadStringCompleted += h,
        h => wc.DownloadStringCompleted -= h);

This returns an object of type IObservable<EventPattern<DownloadStringCompletedEventArgs>>. It might be better to turn this into an IObservable<string>. That's easy too.

var sources2 =
    from ep in sources
    select ep.EventArgs.Result;

Now to actually get the values out, but limit them to every 50ms is also easy.

sources2
    .Sample(TimeSpan.FromMilliseconds(50))
    .Subscribe(t =>
    {
        // Do something with the text returned.
    });

That's it. Super easy.

Answer 2

I would do it the other way around, i.e., run the UI update task on the UI thread, and request the data from there. In a nutshell:

async Task UpdateUIAsync(CancellationToken token)
{
    while (true)
    {
        token.ThrowIfCancellationRequested();

        await Dispatcher.Yield(DispatcherPriority.Background);

        var data = await GetDataAsync(token);

        // do the UI update (or ViewModel update)
        this.TextBlock.Text = "data " + data;
    }
}

async Task<int> GetDataAsync(CancellationToken token)
{
    // simulate async data arrival
    await Task.Delay(10, token).ConfigureAwait(false);
    return new Random(Environment.TickCount).Next(1, 100);
}

This updates the status as fast as data arrives, but note await Dispatcher.Yield(DispatcherPriority.Background). It's there to keeps the UI responsive if data is arriving too fast, by giving the status update iterations a lower priority than user input events.

[UPDATE] I decided to take this a bit further and show how to handle the case when there's a background operation constantly producing the data. We might use Progress<T> pattern to post updates to the UI thread (as shown here). The problem with this would be that Progress<T> uses SynchronizationContext.Post which queues callbacks asynchronously. Thus, the currently shown data item might not have been the most recent one already when it got displayed.

To avoid that, I created Buffer<T> class, which is essentially a producer/consumer for a single data item. It exposes async Task<T> GetData() on the consumer side. I couldn't find anything similar in System.Collections.Concurrent, although it may already exist somewhere (I'd be interested if someone points that out). Below is a complete WPF app:

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Threading;

namespace Wpf_21626242
{
    public partial class MainWindow : Window
    {
        public MainWindow()
        {
            InitializeComponent();

            this.Content = new TextBox();

            this.Loaded += MainWindow_Loaded;
        }

        async void MainWindow_Loaded(object sender, RoutedEventArgs e)
        {
            try
            {
                // cancel in 10s
                var cts = new CancellationTokenSource(10000);
                var token = cts.Token;
                var buffer = new Buffer<int>();

                // background worker task
                var workerTask = Task.Run(() =>
                {
                    var start = Environment.TickCount;
                    while (true)
                    {
                        token.ThrowIfCancellationRequested();
                        Thread.Sleep(50);
                        buffer.PutData(Environment.TickCount - start);
                    }
                });

                // the UI thread task
                while (true)
                {
                    // yield to keep the UI responsive
                    await Dispatcher.Yield(DispatcherPriority.Background);

                    // get the current data item
                    var result = await buffer.GetData(token);

                    // update the UI (or ViewModel)
                    ((TextBox)this.Content).Text = result.ToString();
                }
            }
            catch (Exception ex)
            {
                MessageBox.Show(ex.Message);
            }
        }

        /// <summary>Consumer/producer async buffer for single data item</summary>
        public class Buffer<T>
        {
            volatile TaskCompletionSource<T> _tcs = new TaskCompletionSource<T>();
            object _lock = new Object();  // protect _tcs

            // consumer
            public async Task<T> GetData(CancellationToken token)
            {
                Task<T> task = null;

                lock (_lock)
                    task = _tcs.Task;

                try
                {
                    // observe cancellation
                    var cancellationTcs = new TaskCompletionSource<bool>();
                    using (token.Register(() => cancellationTcs.SetCanceled(),
                        useSynchronizationContext: false))
                    {
                        await Task.WhenAny(task, cancellationTcs.Task).ConfigureAwait(false);
                    }

                    token.ThrowIfCancellationRequested();

                    // return the data item
                    return await task.ConfigureAwait(false);
                }
                finally
                {
                    // get ready for the next data item
                    lock (_lock)
                        if (_tcs.Task == task && task.IsCompleted)
                            _tcs = new TaskCompletionSource<T>();
                }
            }

            // producer
            public void PutData(T data)
            {
                TaskCompletionSource<T> tcs;
                lock (_lock)
                {
                    if (_tcs.Task.IsCompleted)
                        _tcs = new TaskCompletionSource<T>();
                    tcs = _tcs;
                }
                tcs.SetResult(data);
            }
        }

    }
}

Answer 3

Assuming that you are updating your UI through data binding (as you should in WPF), and that you are on .NET 4.5, you can simply use the delay property on your binding expression instead of all this infrastructure.

Read a nice, comprehensive article here.

---EDIT--- Our fake model class:

public class Model
{
    public async Task<int> GetDataAsync()
    {
        // Simulate work done on the web service
        await Task.Delay(1000);
        return new Random(Environment.TickCount).Next(1, 100);
    }
}

Our view model, that gets updated as many times as needed (always on the UI thread):

public class ViewModel : INotifyPropertyChanged
{
    public event PropertyChangedEventHandler PropertyChanged = delegate { };

    private readonly Model _model = new Model();
    private int _data;

    public int Data
    {
        get { return _data; }
        set
        {
            // NotifyPropertyChanged boilerplate
            if (_data != value)
            {
                _data = value;
                PropertyChanged(this, new PropertyChangedEventArgs("Data"));
            }
        }
    }

    /// <summary>
    /// Some sort of trigger that starts querying the model; for simplicity, we assume this to come from the UI thread.
    /// If that's not the case, save the UI scheduler in the constructor, or pass it in through the constructor.
    /// </summary>
    internal void UpdateData()
    {
        _model.GetDataAsync().ContinueWith(t => Data = t.Result, TaskScheduler.FromCurrentSynchronizationContext());
    }
}

And finally our UI, that only gets updated after 50 ms, with no regards to how many times the view model property changed in the meantime:

    <TextBlock Text="{Binding Data, Delay=50}" />