These extensions create an IObservable<T>
that creates a long running task that can be cancelled. It runs the Func<T>
after each interval of time and passes the result to OnNext()
until the task is ended by the CancellationToken
. The important thing here is that only one task is created with one loop. I'd prefer to use Observable.Create
reactive extension, but I cannot find the equivalent.
What are the issues with this implementation?
using System;
using System.Collections.Generic;
using System.Reactive.Disposables;
using System.Threading;
using System.Threading.Tasks;
namespace Observables
{
internal class BasicObservable<T> : IObservable<T>
{
readonly List<IObserver<T>> _observers = new List<IObserver<T>>();
public BasicObservable(
Func<T> getData,
TimeSpan? interval = null,
CancellationToken cancellationToken = default
) =>
Task.Run(async () =>
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
await Task.Delay(interval ?? new TimeSpan(0, 0, 1));
_observers.ForEach(o => o.OnNext(getData()));
}
catch (Exception ex)
{
_observers.ForEach(o => o.OnError(ex));
}
}
_observers.ForEach(o => o.OnCompleted());
}, cancellationToken);
public IDisposable Subscribe(IObserver<T> observer)
{
_observers.Add(observer);
return Disposable.Create(observer, (o) => _observers.Remove(o));
}
}
public static class ObservableExtensions
{
public static IObservable<T> CreateObservable<T>(
this Func<T> getData,
CancellationToken cancellationToken = default)
=> new BasicObservable<T>(getData, default, cancellationToken);
public static IObservable<T> CreateObservable<T>(
this Func<T> getData,
TimeSpan? interval = null,
CancellationToken cancellationToken = default)
=> new BasicObservable<T>(getData, interval, cancellationToken);
}
}
Usage
using Microsoft.VisualStudio.TestTools.UnitTesting;
using System;
using System.Diagnostics;
using System.Reactive.Linq;
using System.Threading;
using System.Threading.Tasks;
namespace Observables
{
class Subscriber
{
public string Name;
//Listen for OnNext and write to the debug window when it happens
public Subscriber(IObservable<string> observable, string name)
{
Name = name;
var disposable = observable.Subscribe((s) => Debug.WriteLine($"Name: {Name} Message: {s}"));
}
}
[TestClass]
public class UnitTest1
{
string GetData() => "Hi";
[TestMethod]
public async Task Messaging()
{
var cancellationSource = new CancellationTokenSource();
var cancellationToken = cancellationSource.Token;
Func<string> getData = GetData;
var publisher = getData.CreateObservable(cancellationToken);
new Subscriber(publisher, "One");
new Subscriber(publisher, "Two");
for (var i = 0; true; i++)
{
if (i >= 5)
{
cancellationSource.Cancel();
}
await Task.Delay(1000);
}
}
}
}