Rxswift: Merge on a collection of observables

[joeljeske]

From the merge docs here, it would make sense that I could merge a collection of similarly typed observables like the following

let taps: Observable<Void> = [btn1.rx_tap, btn2.rx_tap].merge()

Instead, it appears that it only accepts an observable stream of similarly typed observables.

let taps: Observable<Void> = [btn1.rx_tap, btn2.rx_tap].toObservable().merge()

The toObservable() call seems like unnecessary overhead as the set of observables won't change and is completely known upfront.

Am I missing something or is this how merge() is designed and used throughout all of Rx?

[thanegill]

How's this?

Observable<Void>.of(btn1.rx_tap, btn2.rx_tap).merge()

[joeljeske]

Its not as much the syntax but the reason why an observable of observables must be created before they can be merged. Why couldn't it be merged exactly at the time of the call instead of as they are emitted?

[kzaher]

Hi @joeljeske ,

Observable<Void>.of(btn1.rx_tap, btn2.rx_tap).merge()

... this does give a little more verbosity, but also clarity to the code.

[ob1, ob2].merge()

... is little less clear that it's related with Rx, but not too bad. I'm thinking of maybe creating additional convenience for this since we do have the same for concat.

[LukaJCB]

Is there any reason why we can't have btn1.rx_tap.merge(btn2.rx_tap) ?

[Dwar3xwar]

@LukaJCB I think withLatestFrom is a better method to use when combining only two observables so having the merge syntax be similar to that will become unclear. For merge, the design is to combine an much larger size of observables into a single stream.

[LukaJCB]

@Dwar3xwar correct me if I'm wrong, but doesn't withLatestFrom have completely different semantics? Merge emits if any of the given Observables emit, while withLatestFrom only emits when the first one does.

[kzaher]

We'll change this in RxSwift 3.0.

Observable.merge([ob1, ob2])
Observable.concat([ob1, ob2])

and another overloads

ob1.merge(ob2)
ob1.concat(ob2)

This should please everyone I believe.

[guitcastro]

@kzaher Those overload were added ?

[carlosypunto]

@kzaher I think actually is Observable.from([ob1, ob2]).merge()

[acchou]

It doesn't look like ob1.merge(ob2) got implemented?

[DivineDominion]

Neither does Observable.of([ob1, ob2]) work, for that matter

[pschneider]

This combined with the mergeDelayError Operator would be awesome.
http://reactivex.io/documentation/operators/merge.html

Had the first use case where this would've been quite neat today :)

[kzaher]

Hi guys,

I've just pushed additional merge operator overloads to develop branch.

The only operator that isn't implemented now is ob1.merge(ob2).

For merge you can now do:

Observable.merge(btn1.rx_tap, btn2.rx_tap)
Observable.merge([btn1.rx_tap, btn2.rx_tap])

The reason why we currently don't plan to support ob1.merge(ob2) is because merge is a symmetrical operation. ob1.merge(ob2) would imply ob1 is somehow different than obs2.

After some discussions with other core contributors, we've decided Observable.merge(btn1.rx_tap, btn2.rx_tap) is enough for the time being.

ob1.concat(ob2) makes sense IMHO because ob1 really comes before ob2 so they don't have a symmetrical relationship.

I'm going to close this issue for now, but feel free to reopen it if there is something wrong.

[arctouch-marcelogobetti]

Hi @kzaher , I'm running a playground on RxSwift 3.3.1 and maybe I didn't fully understand this new merge() operator. Please look at these examples, I expected them to behave equally:

let disposeBag = DisposeBag()
let resources = ["a", "b", "c", "d"]
let observables = resources.map { Observable.just($0) }

example("Observable.from([...]).merge()") {
  let observablesFrom = Observable.from(observables)
  let merge = observablesFrom.merge()
  merge.subscribe(onNext: { print($0) }).disposed(by: disposeBag)
  merge.toArray().subscribe(onNext: { print($0) }).disposed(by: disposeBag)
}

example("Observable.merge([...])") {
  let merge = Observable.merge(observables)
  merge.subscribe(onNext: { print($0) }).disposed(by: disposeBag)
  merge.toArray().subscribe(onNext: { print($0) }).disposed(by: disposeBag)
}

example("Observable.from([...]).concat()") {
  let observablesFrom = Observable.from(observables)
  let concat = observablesFrom.concat()
  concat.subscribe(onNext: { print($0) }).disposed(by: disposeBag)
  concat.toArray().subscribe(onNext: { print($0) }).disposed(by: disposeBag)
}

example("Observable.concat([...])") {
  let concat = Observable.concat(observables)
  concat.subscribe(onNext: { print($0) }).disposed(by: disposeBag)
  concat.toArray().subscribe(onNext: { print($0) }).disposed(by: disposeBag)
}

The first merge() example prints the 4 strings: each one on its own onNext, and all of them in a single array when using toArray() - which is what I want. The second merge() example does only print "a" and ["a"]. The concat() examples both work like the first merge() example.
Please ignore the fact that I'm creating several variables, it was just to have a better visualization of the return type on each step.

Thank you in advance and sorry if I'm doing any silliness here!