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

Hmm... My tests pass with this as well:

public static IObservable<TResult> WithLatestFrom<TResult, TFirst, TSecond>(
    this IObservable<TFirst> first,
    IObservable<TSecond> second,
    Func<TFirst, TSecond, TResult> resultSelector)
{
    return Observable.Create<TResult>(observer =>
    {
        var published = first.Publish();

        var connection = published.Connect();

        var subscription = second
            .Select(a => published
                .Select(b => resultSelector(b, a)))
            .Switch()
            .Subscribe(observer);

        return new CompositeDisposable(connection, subscription);
    });
}

What's the difference between the two? New subscribers won't get the last published value on the first stream?

Actually, you are right. My impl will also raise when first yeilds, which is like combine latest.

@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());
}