WithLatestFrom

WithLatestFrom.cs

void Main()
{
	var stream1 = Observable.Interval(TimeSpan.FromSeconds(5)).Select(_ => "a" + _);
	var stream2 = Observable.Interval(TimeSpan.FromSeconds(1)).Select(_ => "b" + _);
	
	stream1.WithLatestFrom(stream2, (a, b) => new {a, b}).DumpLatest(true);
}

public static class ObservableExtensions
{
    public static IObservable<TResult> WithLatestFrom<TResult, TFirst, TSecond>(
        this IObservable<TFirst> first,
        IObservable<TSecond> second,
        Func<TFirst, TSecond, TResult> resultSelector)
    {
        return Observable.Create<TResult>(obs => {
			// Need this so when we switch we instantly replay the last value of first
			var replay = first.Replay(1);
			// Connect to make replay (or first) hot
			var replaySubs = replay.Connect();
			// Now we subscribe to second, not sure which way around you want this api to work.
			var subs = second
				.Select(a => replay.Select(b => new { a, b }))
				.Switch()
				.Select(val => resultSelector(val.b, val.a))
				.Subscribe(obs);
			return new CompositeDisposable(replaySubs, subs);
		});
    }
}

@james-world came up with an even shorter approach 😁

public static IObservable<TResult> WithLatestFrom<TResult, TFirst, TSecond>(
    this IObservable<TFirst> first,
    IObservable<TSecond> second,
    Func<TFirst, TSecond, TResult> resultSelector)
{
    return first.Publish(os => second.Select(a => os.Select(b => resultSelector(b, a))).Switch());
}