CODEWITHSUNDEEP
c#.netsystem.reactivec#-5.0
Ivan Milutinović
Ivan Milutinović

Reputation: 319

How to await an object inside IObservable with a specific property value?

To clarify, i have a method:

public static IObservable<Node> GetNodes()
    {            
        var computers = GetComputersInLan();
        return computers.Select(computerAddress => GetNode(computerAddress));
    }

GetComputersInLan method returns IObservable of IPAddress

private static IObservable<IPAddress> GetComputersInLan()
    {
        var tasks = new List<Task<PingReply>>();
        for (int i = 1; i < 255; i++)
        {
            Ping p = new Ping();
            ipBytes[3] = (byte)(++ipBytes[3]);
            IPAddress address = new IPAddress(ipBytes);
            tasks.Add(p.SendPingAsync(address, 2000));
        }
        return tasks.ToObservable().Where(x => x.Result.Status == IPStatus.Success).Select(y => y.Result.Address);
    }

GetNode method constructs a Node. 

private static Node GetNode(IPAddress ipAddress)
    {
       return new Node(ipAddress, (IHandler)Activator.GetObject(typeof(Handler), "tcp://" + ipAddress + ":1337/handler"));
    }

public class Node
{
    private IHandler Handler { get; set; }
    public IPAddress Address { get; set; }
    public int AvailableCores { get; set; }

    public async Task<TResult> Invoke<TResult>(Func<TResult> method)
    {
        AvailableCores--;
        var result = await Task.Run<TResult>(() => Handler.Invoke(method));
        AvailableCores++;
        return result;
    }
}

Handler is a remote computer, and AvailableCores represents its cpu cores.

What I want is to await method GetNodes to return the first Node that has more than 0 AvailableCores.

await GetNodes().FirstAsync(node => node.AvailableCore > 0)

But what happens, is that after enough calls to method Invoke, instead of waiting for cores to become available, it fires an exception "sequence contains no elements".

Upvotes: 4

Views: 1084

Answers (1)

Gideon Engelberth
Gideon Engelberth

Reputation: 6155

That is expected behavior for this method. FirstAsync will only check the current state of the items you pass to it, either returning the first match or throwing the exception you are encountering if there is no match.

You will have to manage the case of waiting until a core becomes available yourself. You could try FirstOrDefaultAsync to return null instead of throwing an exception when all cores are busy. From there, you will need some scheme to detect when a core becomes available for the next unit of work, be that an event or polling.

Upvotes: 1

Related Questions