CODEWITHSUNDEEP
c#.netsystem.reactiveasync-awaitc#-5.0
AK_
AK_

Reputation: 8099

Reactive Extensions Observable from a collection of web queries

relevant: async function over a list

I Have a function that looks like this:

public async Task<decimal> GoToWeb(string Sym){}

I want to call it periodically over a list of string once a minute. How can I convert this into a Reactive Extensions Observable?

Upvotes: 0

Views: 1110

Answers (2)

Enigmativity
Enigmativity

Reputation: 117027

Since the reactive framework can handle the asynchrony for you you could try this:

var syms = new List<string>() { "ANZ", "BHP", };

var query =
    from i in Observable.Interval(TimeSpan.FromSeconds(1.0))
    from sym in syms.ToObservable()
    from d in GoToWeb(sym).ToObservable()
    select new
    {
        Symbol = sym,
        Value = d,
    };

You'll need to add a reference to the namespace System.Reactive.Threading.Tasks to get the ToObservable() extension for tasks.

Does this meet your needs?

Upvotes: 2

James Manning
James Manning

Reputation: 13579

I'm an Rx noob, so there's likely a better option, but without knowing what you want to happen for each completed batch of web calls, I avoided using the linq syntax (even though I prefer using it when possible).

I did this in LINQPad using the NuGet package for Rx-Main(Prerelease)

enter image description here

void Main()
{
    var eachMinuteSequence = Observable.Timer(TimeSpan.Zero, TimeSpan.FromMinutes(1));

    var symbols = new[] { "GOOG", "MSFT", "AAPL" };

    Action<long> eachMinuteAction = async _ =>
    {
        var tasks = StartWebCalls(symbols);
        var results = await Task.WhenAll(tasks);
        // do something with the results
    };
    eachMinuteSequence.Subscribe(eachMinuteAction);
}

// Define other methods and classes here
public Task<decimal>[] StartWebCalls(IEnumerable<string> stockSymbols)
{
    return (
        from symbol in stockSymbols
        select GoToWeb(symbol)
    ).ToArray();
}

public async Task<decimal> GoToWeb(string Sym){throw new NotImplementedException();}

Upvotes: 0

Related Questions