How to zip/merge by key many sources using .net reactive?

Question

I have many sources (number is known only in run-time) in a list. All sources emit the same type of elements (Data). How to group them by a key (currentDate) which is one of its properties? Then I need to convert them into one different element (FullData) only if all sources emit valid Data element. So FullData is emited only if every source emit valid Data for a particular DateTime.

class Program
{
    static void Main(string[] args)
    {
        var rand = new Random();
        List<IObservable<Data>> sources = new List<IObservable<Data>>();

        //let's assume that value comes from a user
        var sourcesCounter = 4;
        for (int i = 0; i < sourcesCounter; i++)
        {
            sources.Add(
                Observable.Interval(TimeSpan.FromSeconds(1))
                    .Select(e => new Data
                    {
                        currentDate = DateTime.Now, //let's assume it is round to seconds
                        Samples = new List<double>(1000),
                        IsValid = rand.Next(5) < 4 //Generate true/false randomly
                    })

            );
        }

        var merged = sources.Merge();
        merged.Subscribe(
            e =>
            {
                Console.WriteLine($"received: {e.currentDate.Second} {e.IsValid}");
            },
            ex => Console.WriteLine(ex),
            () => Console.WriteLine("Completed - merged")
        );

        Console.ReadKey();
    }
}

public class Data
{
    public DateTime currentDate { get; set; }
    public bool IsValid { get; set; }
    public List<double> Samples { get; set; }
}

public class FullData
{
    public DateTime currentDate { get; set; }
    public List<List<double>> Samples { get; set; }
}

Answer 1

The code below groups the data by currentDate until it gets an invalid data for a particular key.

var keys = new ConcurrentDictionary<DateTime, DateTime>();
var dataDictionary = new ConcurrentDictionary<DateTime, FullData>();
sources
    .Merge()
    .GroupByUntil(data => data.currentDate, s => s.Any(data => !data.IsValid)) // group by currentDate until an invalid data appears in the group (the last invalid data can be in this group)
    .Where(g => keys.TryAdd(g.Key, g.Key)) // skip the reborned groups for the same key (they are created because of durationSelector, which controls the lifetime of a group)
    .Merge() // switch to the previous flattened structure
    .Where(data => data.IsValid) // remove the last invalid item emitted by GroupByUntil
    .Subscribe(x =>
    {
        var fullData = dataDictionary.GetOrAdd(x.currentDate, f => new FullData { currentDate = x.currentDate, Samples = new List<List<double>>() });
        fullData.Samples.Add(x.Samples);

        Console.WriteLine($"received: {x.currentDate.ToLocalTime()} {x.IsValid} {string.Join(", ", x.Samples)}");
    }, () => Console.WriteLine("Completed"));

Console.ReadKey();

foreach (var item in dataDictionary)
{
    Console.WriteLine($"{item.Key.ToLocalTime()}, {string.Join(",", item.Value.Samples.SelectMany(t => t))}");
}

---

If you know that all observable sequences are finite, and you need to create FullData only if every source emit only valid data, you can use different approach:

sources.Merge().ToList().Subscribe(list =>
{
    var fullDataList = list
        .GroupBy(data => data.currentDate)
        .Where(g => g.All(data => data.IsValid))
        .Select(g => new FullData { currentDate = g.Key, Samples = g.Select(data => data.Samples).ToList() });
    foreach (var fullDataItem in fullDataList)
    {
        Console.WriteLine($"{fullDataItem.currentDate.ToLocalTime()}, {string.Join(",", fullDataItem.Samples.SelectMany(t => t))}");
    }
});

The above code waits for the completion of all observables, creates a list of all received items, and finally generates FullData using a simple LINQ query.

Answer 2

There's an overload of Zip that takes a IEnumerable<IObservable<Data>>.

Try this:

var rand = new Random();

var sourcesCounter = 4;

IEnumerable<IObservable<Data>> sources =
    Enumerable
        .Range(0, sourcesCounter)
        .Select(x =>
            Observable
                .Interval(TimeSpan.FromSeconds(1))
                .Select(e => new Data()
                {
                    currentDate = DateTime.Now, //let's assume it is round to seconds
                    Samples = new List<double>(1000),
                    IsValid = rand.Next(5) < 4 //Generate true/false randomly
                }));

    IObservable<IList<Data>> zipped = sources.Zip(values => values);