The power of Reactive Programming and CancellationTokens

One of the most common scenarios that a mobile developer could encounter is to have an async method that depends on a platform capability like Internet connectivity, Bluetooth, Camera,…

While checking for the capability availability sound straight forward and a method like this one can do the job

public async Task<string> GetDataAsync()

{

if (Connectivity.NetworkAccess != NetworkAccess.Internet

||Connectivity.ConnectionProfiles.All(x=>x!=ConnectionProfile.WiFi))

return null;

HttpResponseMessage response;

using (var httpClient = new HttpClient())

{

response = await httpClient.GetAsync("https://medium.com/feed/@Medium");

}

response.EnsureSuccessStatusCode();

return await response.Content.ReadAsStringAsync();

}

What if after the checking pass or while downloading the internet connection gets interrupted, In this case, you could get an HttpException may be in this case it can handle the exception but what the connection type changed from wifi to cellular for example, now you will download the data from the user cellular data.

The good news is that HttpClient.GetAsync has an overload that takes a CancellationToken as a parameter, so now all that we need is to use this overlead and pass to it a CancellationToken that gets canceled when the internet connection changed.

but here one challenged the xamarin essential expose a ConnectivityChanged event and implementing a CancellationTokenSource based on events is a bit tricky and can introduce more bugs to your code.

In this article, I will provide you with a reusable method that you can use to convert any .net event to a CancellationTokenSource.

We first will convert our Event to an Observableusing C# reactive extension

Observable

.FromEventPattern<Xamarin.Essentials.ConnectivityChangedEventArgs>(

handler => Xamarin.Essentials.Connectivity.ConnectivityChanged += handler, handler => Xamarin.Essentials.Connectivity.ConnectivityChanged -= handler)

.FirstAsync(x=>x.EventArgs.NetworkAccess!=NetworkAccess.Internet)

and then will convert the observableto a custom implementation of CancellationTokenSource

	public class ObservableCancellationTokenSource<T> : CancellationTokenSource,IDisposable
	{
	private readonly IDisposable _subscription;
	public ObservableCancellationTokenSource(IObservable<T> observable)
	{
	_subscription = observable.Subscribe(token=>this.Cancel());
	}
	public new void Dispose()
	{
	_subscription.Dispose();
	base.Dispose();
	}
	}
	public static class Extension
	{
	public static ObservableCancellationTokenSource<T> ToCancellationTokenSource<T>(this IObservable<T> source)
	{
	return new ObservableCancellationTokenSource<T>(source);
	}
	}

view raw ObservableCancellationTokenSource.cs hosted with ❤ by GitHub

our ObservableCancellationTokenSource is just an adapter that takes IObservavle As a parameter subscribe on it and just call the cancellation method when the observable publish a new value.

The dispose method takes care of disposing of the CanellationTokenSourceand the subscription so no memory leaks will get introduced.

now the modified version of our GetDataAsync method should look like that

public static async Task<string> GetDataAsync(CancellationToken cancellationToken)

{

if (cancellationToken.IsCancellationRequested)

return null;

HttpResponseMessage response;

using (var httpClient = new HttpClient())

{

response = await httpClient.GetAsync("https://medium.com/feed/@Medium", cancellationToken);

}

response.EnsureSuccessStatusCode();

return await response.Content.ReadAsStringAsync();

}

and the caller will look like

var cancellationTokenSource = Observable

.FromEventPattern<ConnectivityChangedEventArgs>(

handler => Connectivity.ConnectivityChanged += handler,

handler => Connectivity.ConnectivityChanged -= handler)

.FirstAsync(x => x.EventArgs.NetworkAccess != NetworkAccess.Internet

||Connectivity.ConnectionProfiles.All(connectionProfile => connectionProfile != ConnectionProfile.WiFi))

.ToCancellationTokenSource();

using (cancellationTokenSource)

{

var result = await FullState.GetDataAsync(cancellationTokenSource.Token);

}

Rx Bonus (combine events)

as we are using RX observable, we have the opportunity of combining multiple events to a single CancellationToken.

Func<EventPattern<ConnectivityChangedEventArgs>, bool> connectivityPredicate = x => x.EventArgs.NetworkAccess != NetworkAccess.Internet

||Connectivity.ConnectionProfiles.All(connectionProfile => connectionProfile != ConnectionProfile.WiFi);

Func<EventPattern<EnergySaverStatusChangedEventArgs>, bool> batteryPredicate = x => x.EventArgs.EnergySaverStatus == EnergySaverStatus.On;

var connectivityObservable = Observable

.FromEventPattern<ConnectivityChangedEventArgs>(

handler => Connectivity.ConnectivityChanged += handler,

handler => Connectivity.ConnectivityChanged -= handler);

var batteryObservable = Observable

.FromEventPattern<EnergySaverStatusChangedEventArgs>(

handler => Battery.EnergySaverStatusChanged += handler,

handler => Battery.EnergySaverStatusChanged -= handler);

var cancellationTokenSource = connectivityObservable

.CombineLatest(batteryObservable,(connectivityArgs, batteryArgs) => (connectivityArgs, batteryArgs))

.FirstAsync(x=>connectivityPredicate(x.connectivityArgs)|| batteryPredicate(x.batteryArgs))

.ToCancellationTokenSource();

similar behavior can be achieved using CancellationTokenSource.CreateLinkedTokenSource but I prefer the observable approach.