Zipping Rx IObservable with infinite number set
- by Toni Kielo
I have a IObservable [named rows in the sample below] from Reactive extensions framework and I want to add index numbers to each object it observes.
I've tried to implement this using Zip function:
rows.Zip(Enumerable.Range(1, int.MaxValue), (row, index) =>
new { Row = row, Index = index })
.Subscribe(a => ProcessRow(a.Row, a.Index), () => Completed());
.. but unfortunately this throws
ArgumentOutOfRangeException:
Specified argument was out of the range of valid values.
Parameter name: disposables
Am I understanding the Zip function wrong or is there a problem with my code?
The Range part of the code doesn't seem to be the problem and the IObservable isn't yet receiving any events.
