Merging union Observables

Question

I am having a scenario where I need to execute observables that depends on the result of the first one. However I need to keep the result of the first observable. I couldn't find any extension that would help me do this. For instance SelectMany does a projection of the first observable discarding the source results. Therefore I made my own extension:

public static IObservable<T> MergeWithResultPropagation<T>(this IObservable<T> src, Func<T, IEnumerable<IObservable<T>>> elems)
{
    return src.SelectMany(result => elems(result).Union(new[] { Observable.Return(result) }))
        .SelectMany(r => r);
}

Here's a dummy example:

Observable.Range(1, 3)
.MergeWithResultPropagation(item => new[]{
    Observable.Return(item * 2), 
    Observable.Return(item * 3)
}).Dump();

Is this an adequate way to solve the problem? Did I miss an extension method that does this?

Answer 1

Union vs Concat

As you addmited, there is no need to use the Union extension because no two items will ever have the same value. Concat would be more appropriate because it'll better show what is going on.

Difference between the two is that

[ 1, 2 ].Union([2, 3]) = [1, 2, 3]

whereas the same with Concat would be

[ 1, 2 ].Concat([2, 3]) = [1, 2, 2, 3] 

because of the not clear intention I had a hard time understanding this short code. The descriptions says something else then the implementation.

---

Zip

I don't know any extension that could do the same job but I think this one could be expressed cleaner by first producing the results and then zipping each result with the corresponding item that lead to this result using the Zip extension. I also think that it's nicer to use Enumerable.Repeat rather then new []{}

return 
    src
        .Select(x => elems(x))
        .Zip(src, (results, x) => results.Concat(Enumerable.Repeat(Observable.Return(x), 1)))
        .SelectMany(z => z)
        .SelectMany(x => x);

Functional

To make it even cleaner I suggest encapsulating the Concat in its own method so the final extension could be:

public static IObservable<T> MergeWithResultPropagation3<T>(this IObservable<T> values, Func<T, IEnumerable<IObservable<T>>> factory)
{
    return
        values
            .Select(x => factory(x))
            .Zip(values, AppendValue())
            .SelectMany(z => z)
            .SelectMany(x => x);

    Func<IEnumerable<IObservable<T>>, T, IEnumerable<IObservable<T>>> AppendValue()
    {
        return (results, value) => results.Concat(Enumerable.Repeat(Observable.Return(value), 1));
    }
}

---

yield return

Alternatively to new []{} and Concat you could make the helper work with yield return

public static IObservable<T> MergeWithResultPropagation3<T>(this IObservable<T> values, Func<T, IEnumerable<IObservable<T>>> factory)
{
    return
        values
            .Select(x => factory(x))
            .Zip(values, (results, value) => AppendValue(results, value))
            .SelectMany(z => z)
            .SelectMany(x => x);

    IEnumerable<IObservable<T>> AppendValue(IEnumerable<IObservable<T>> results, T value)
    {
        foreach (var result in results) yield return result;
        yield return Observable.Return(value);
    }
}