# Merging union Observables

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?

• Is IObservable your own interface because the original one does not support IEnumerable and thus does not have the SelectMany extension. If so, could you include its definition too? – t3chb0t Apr 8 '17 at 13:32
• Oh, I now notice the system.reactive tag. I guess it belongs to this package. – t3chb0t Apr 8 '17 at 13:34
• @t3chb0t Yup, that's right :p – Bruno Costa Apr 8 '17 at 13:34
• I wanted to try your code out so I installed the System.Reactive package from NuGet and added the System.Reactive namespace in LINQPad but it still does not recognize this interface as IEnumerable. Could you tell me what else I need to do? – t3chb0t Apr 8 '17 at 13:38
• @t3chb0t Did you include all dependecies, such as system.reactive.core, system.reactive.infrastructure, etc...? – Bruno Costa Apr 8 '17 at 13:39

# 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);
}
}

• I had a look at zip as well but for some reason I than discarded it. Good to see it in action again, I barely use it. Looks good except for that Enumerable repeat, I prefer to create an array instance :p – Bruno Costa Apr 8 '17 at 14:46