Merge hot source1 with a cold source2

user_161739 0
Viewed 109

source1 emits A,B,C,D etc and never completes

source2 emit 1,2 and completes

I want to merge to A1, B2, C1, D2 etc

update

my initial attemp was to Zip and Repeat as suggested from Theodor however this creates a lock cause source2 generation is expensive.

Last comment from Enigmativity addresses that problem

source1.Zip(source2.ToEnumerable().ToArray().Repeat())
rx.net system.reactive c#
2 Answers

Since you want to repeat source2 indefinitely and you say that it is cold (in the sense that it produces the same set of values each time, and generally in the same sort of cadence) and it is expensive, we want to turn the IObservable<T> into a T[] to ensure it's computed once and only once.

var array = source2.ToEnumerable().ToArray();
var output = source1.Zip(array.Repeat(), (x, y) => (x, y));
user_2597690

Assuming that the desirable marble diagram is like this:

Source1: +--------A-------B-------C--------D-------|
Source2: +----1--------------2--------|
Merged:  +--------A1---------B2-------C1---D2------|

Here is a ZipWithRepeated operator having this behavior:

static IObservable<(TFirst First, TSecond Second)> ZipWithRepeated<TFirst, TSecond>(
    this IObservable<TFirst> first, IObservable<TSecond> second)
{
    return second.Replay(replayed => first.ToAsyncEnumerable()
        .Zip(replayed.ToAsyncEnumerable().Repeat())
        .ToObservable());
}

Usage example:

var merged = source1.ZipWithRepeated(source2);

This solution requires a dependency to the System.Linq.Async and System.Interactive.Async packages, because both sequences are converted to IAsyncEnumerable<T>s before the zipping.

Alternative: Instead of relying on the Rx Replay operator for the buffering of the source2 sequence, a more efficient solution would be to do the buffering after the conversion from observable to async-enumerable. AFAICS there is no built-in support for replaying/memoizing IAsyncEnumerable<T>s in the official Rx/Ix libraries, but creating a custom Repeat operator with embedded buffering is not very difficult. Below is an alternative implementation of the ZipWithRepeated operator, which is based on this idea:

static IObservable<(TFirst First, TSecond Second)> ZipWithRepeated<TFirst, TSecond>(
    this IObservable<TFirst> first, IObservable<TSecond> second)
{
    return first.ToAsyncEnumerable()
        .Zip(second.ToAsyncEnumerable().RepeatBuffered())
        .ToObservable();
}

private async static IAsyncEnumerable<TSource> RepeatBuffered<TSource>(
    this IAsyncEnumerable<TSource> source,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    var buffer = new List<TSource>();
    await foreach (var item in source
        .WithCancellation(cancellationToken).ConfigureAwait(false))
    {
        buffer.Add(item); yield return item;
    }
    while (true) foreach (var item in buffer) yield return item;
}

This implementation does not depend on the System.Interactive.Async package, but only on the System.Linq.Async package.

user_111785490
Related