Observable that combines behavior of groupby with combinelatest and exhibits "outer join" behavior

Let's say your foo$ stream emits values of type Foo, and bar$ stream emits values of Bar. Here is how you can combine them:

combineLatest([
  foo$,
  bar$
    // use startWith(null) to ensure combineLatest will emit as soon as foo$ emits, not waiting for bar$ to emit its first value
    .pipe(startWith(null))
]).pipe(
  map(([foo, bar]) => ({
    // always keep all properties from foo
    ...foo,
    // only add properties from bar if it has the matching Key
    ...(bar && bar.Key === foo.Key ? bar : null)
  }))
)

This may not be "cosher" by some standards but works for what I need it to do. Posting for anyone looking for same functionality (.NET Version of RX)

public static class Extensions
{
    public static IObservable<TResult> CombineLatestGrouped<TFirst,TSecond,TKey, TResult>(
        this IObservable<TFirst> first, 
        IObservable<TSecond> second, 
        Func<TFirst, TKey> firstKeySelector, 
        Func<TSecond, TKey> secondKeySelector, 
        Func<TKey,TFirst,TSecond,TResult> resultSelector)

    {
        var dic = new ConcurrentDictionary<TKey,Tuple<TFirst,TSecond>>();

        return Observable.Create<TResult>(obs =>
        {
            var d1 = first
            .Select(x =>
            {
                var key = firstKeySelector(x);
                var tuple = dic.AddOrUpdate(
                    key,
                    addValueFactory: key => Tuple.Create(x, default(TSecond)),
                    updateValueFactory: (key, existing) => Tuple.Create(x, existing.Item2));
                return resultSelector(key, tuple.Item1, tuple.Item2);
            })
            .Subscribe(obs);

            var d2 = second
            .Select(x =>
            {
                var key = secondKeySelector(x);
                var tuple = dic.AddOrUpdate(
                    key,
                    addValueFactory: key => Tuple.Create(default(TFirst), x),
                    updateValueFactory: (key, existing) => Tuple.Create(existing.Item1, x));
                return resultSelector(key, tuple.Item1, tuple.Item2);
            })
            .Subscribe(obs);

            return new CompositeDisposable(d1, d2);
        });
    }
}

As I see it, you would like following:

a new "key" will yield a Target class with the other side being null

When left or right side emmits a NEW key (prev: null or different)

, and then once the other side also produces the same joining key,

precondition: a stream emitted a value -- other stream now emits a value and the key for left and right eq

the latest from both sides is yielded whenever there's a new value in either sequence for the given key.

emit full target (composed of left, right) on each left,right emit, when a value of left,right changes distinctly

RxJava2 solution for my assumption:

  @Test
  void test2() {
    PublishSubject<Foo> foo$ = PublishSubject.create();
    PublishSubject<Bar> bar$ = PublishSubject.create();

    Observable<Target> target$ = Observable.merge(Arrays.asList(foo$, bar$))
        // filter invalid values
        .filter(hasId -> hasId.key() != null)
        .scan(Target.NULL, (prev, change) -> {
          // when prev. target and current value#key are eq -> emit composed value
          if (change.key().equals(prev.key)) {
            return composedTarget(prev, change);
          } else if (change instanceof Foo) {
            return Target.fromFoo((Foo) change);
          } else if (change instanceof Bar) {
            return Target.fromBar((Bar) change);
          }
          return prev;
        }).filter(target -> target != Target.NULL)
        .distinctUntilChanged();

    TestObserver<Target> test = target$.test();

    // emit
    foo$.onNext(new Foo("123", "f1"));
    // emit
    bar$.onNext(new Bar("123", "f2"));
    // emit
    bar$.onNext(new Bar("123", "f3"));
    // skipped
    foo$.onNext(new Foo("123", "f1"));
    // emit
    foo$.onNext(new Foo("123", "f5"));
    // emit
    foo$.onNext(new Foo("key", "value"));
    // emit
    foo$.onNext(new Foo("key2", "value2"));
    // emit
    bar$.onNext(new Bar("bar2", "Berlin"));
    // emit
    foo$.onNext(new Foo("foo2", "Funkeey"));

    test.assertValues(
        new Target("123", "f1", null),
        new Target("123", "f1", "f2"),
        new Target("123", "f1", "f3"),
        new Target("123", "f5", "f3"),
        new Target("key", "value", null),
        new Target("key2", "value2", null),
        new Target("bar2", null, "Berlin"),
        new Target("foo2", "Funkeey", null)
    );
  }

  private Target composedTarget(Target prev, HasId change) {
    if (change instanceof Foo) {
      Foo foo = (Foo) change;
      return new Target(prev.key, foo.funky, prev.town);
    }
    if (change instanceof Bar) {
      Bar bar = (Bar) change;
      return new Target(prev.key, prev.funky, bar.town);
    }
    return prev;
  }

Domain-Classes

  interface HasId {
    String key();
  }

  static final class Foo implements HasId {
    final String key;
    final String funky;

    Foo(String key, String funky) {
      this.key = key;
      this.funky = funky;
    }

    @Override
    public String key() {
      return key;
    }
  }

  static final class Bar implements HasId {
    String key;
    String town;

    Bar(String key, String town) {
      this.key = key;
      this.town = town;
    }

    @Override
    public String key() {
      return key;
    }
  }

  static final class Target {
    private static final Target NULL = new Target(null, null, null);
    final String key;
    final String funky;
    final String town;

    Target(String key, String funky, String town) {
      this.key = key;
      this.funky = funky;
      this.town = town;
    }

    static Target fromFoo(Foo foo) {
      return new Target(foo.key, foo.funky, null);
    }

    static Target fromBar(Bar bar) {
      return new Target(bar.key, null, bar.town);
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;
      Target target = (Target) o;
      return key.equals(target.key) &&
          Objects.equals(funky, target.funky) &&
          Objects.equals(town, target.town);
    }

    @Override
    public int hashCode() {
      return Objects.hash(key, funky, town);
    }

    @Override
    public String toString() {
      return "Target{" +
          "key='" + key + '\'' +
          ", funky='" + funky + '\'' +
          ", town='" + town + '\'' +
          '}';
    }
  }

Please correct my assumptions, if I am mistaken. The solution could be implemented way better in C# with pattern-matching. Actually if C# has union types like F#, that would be best.