How to use Interval as a timer to check a webAPI service and return an observable sequence of new values

Question

So, in my mind this is pretty simple. I want to use Interval as a timer to check a WebAPI service every second and then use Distinct to filter it such that I only get notifications when a new value is added.

(ed. I suppose it's good form to highlight early on that I'm using Reactive Extensions for .Net and that's where these terms come from, Interval(...) and Distinct()).

When it comes to implementation I have an async function that returns a Task<PostalCodeModel[]> where the PostalCodeModel is pretty simple dummy model:

public class PostalCodeModel
{
    public int PostalCode { get; set; }
    public string City { get; set; }

    public PostalCodeModel(int code, string name)
    {
        this.PostalCode = code;
        this.City = name;
    }

    public override string ToString()
    {
        return string.Format("{0} = {1}", PostalCode, City);
    }
}

In my test app I have so far been able to figure out how to create the Interval timer:

var interval = Observable.Interval(TimeSpan.FromSeconds(1));

and using the ToObservable() on the Async method:

///Bad!
private static IObservable<PostalCodeModel[]> CheckPostalCodes()
{
    using (ApiClient client = new ApiClient("http://localhost:55457"))
    {
        return client.GetPostalCodesAsync().ToObservable();
    }
}

//Good, thanks Enigmativity!
private static IObservable<PostalCodeModel[]> CheckPostalCodes()
{
    return Observable.Using(
        () => new ApiClient("http://localhost:55457"),
        client => client.GetPostalCodesAsync().ToObservable())
        .Take(1);
}

However, I'm having trouble figuring out how to chain these together and where to apply the Distinct method, my attempts so far have been unsucessful:

// No workie :(
var sup = interval.Subscribe(
            CheckPostalCodes,
            () => Console.WriteLine("Completed"));

Ideally I would do something like:

// No workie :(
Observable.Interval(TimeSpan.FromSeconds(1)).CheckPostalCodes().Distinct()

or similar, but I can't figure out how to do that. Or even if that's a good idea. Eventually the output of this needs to dictate changes in the GUI. And I'm not even sure if using Distinct on an Array of objects works the way I expect it to (only let me know if there is a new value in the array).

As I noted in the comment to Enigmativitys answer below, after correcting the CheckPostalCodes() like he suggested, the subscription was pretty simple, however I did need to add a keySelector to the Distinct call, for it to be able to correctly determine if the returned array is the same. I think simply using array.Count() is sufficient for our purposes, i.e.:

        var query = Observable
                        .Interval(TimeSpan.FromSeconds(1))
                        .SelectMany(_ => CheckPostalCodes())
                        .Distinct(array => array.Count());

score 1 · Accepted Answer · answered May 11 '17 at 12:47

1

You've written the CheckPostalCodes method incorrectly. The using statement has well and truly finished before any observable will have subscribed. You need to use the Observable.Using method instead.

Also, if you are only expecting one value back from your API call don't forget to add a .Take(1) to make sure the observable stops after one return value.

So CheckPostalCodes should look like this:

private static IObservable<PostalCodeModel[]> CheckPostalCodes()
{
    return
        Observable
            .Using(
                () => new ApiClient("http://localhost:55457"),
                client => client.GetPostalCodesAsync().ToObservable())
            .Take(1);
    }
}

Now you can easily put it all together:

var query =
    Observable
        .Interval(TimeSpan.FromSeconds(1.0))
        .SelectMany(n => CheckPostalCodes())
        .Distinct();

Let me know if this works for you.

answered May 11 '17 at 12:47

Enigmativity

113,464
11
89
172

This works to create the sequence, but it is as I feared, Distinct doesn't "deep compare" the array and I get a repeating sequence of the same array value, i.e subscribing with: ```query.Subscribe(array => Console.WriteLine($"{DateTime.Now} {string.Join(", ", (object[]) array)}"));``` I get repeats of something like: ```11.5.2017 13:15:36 110 = Name1, 210 = Name2 11.5.2017 13:15:37 110 = Name1, 210 = Name2``` – gakera May 11 '17 at 13:18
I wonder if I need to do a comparer, like in this answer: http://stackoverflow.com/questions/11916038/rx-operator-to-distinct-sequences - or does one exist somewhere for arrays? – gakera May 11 '17 at 13:22
Actually, I suppose this will be sufficient, since we control on the other end what / how we add to the array: ```var query = Observable .Interval(TimeSpan.FromSeconds(1)) .SelectMany(_ => CheckPostalCodes()) .Distinct(array => array.Count());``` – gakera May 11 '17 at 13:26
Can you explain why the Take(1) is needed, would there be a memory leak if it was not included? – gakera May 11 '17 at 13:53
Also, what would be the best / correct way to handle errors in the Async call, i.e. if the service times-out or returns anything other than OK (it's a HTTP service after all)? – gakera May 11 '17 at 14:05
@gakera - The line "if you are only expecting one value back from your API call don't forget to add a `.Take(1)` to make sure the observable stops after one return value" explains the `.Take(1))`. – Enigmativity May 12 '17 at 00:00
@gakera - You really need to look for existing answers with regard to the handling of errors. If you can't find an existing question/answer then ask a new question. – Enigmativity May 12 '17 at 00:01

How to use Interval as a timer to check a webAPI service and return an observable sequence of new values

1 Answers1