Zip operator doesn't work properly for more than 2 streams

[jachnik]

Bug for versions Versions 4.1.0 and 4.1.1

The code posted below causes the following exception:

Sequence contains no elements

var stream = Observable.Return(1);

var stream2 = stream.Zip(stream, stream, stream, (a, b, c, d) =>
{
       return a + b + c + d;
});
var i = stream2.ToTask().GetAwaiter().GetResult();

Zipping of 2 streams works fine.

As Shlomo on StackOverflow pointed out, if you re-define stream as such:

var stream = Observable.Return(1).Delay(TimeSpan.FromMilliseconds(15));

then the operator works as intended.

[quinmars]

For reference, the mentioned SO question can be found here:

https://stackoverflow.com/questions/52654306/problem-with-zipping-more-than-2-observable-sequences

[quinmars]

The problem is that the queue of the ZipObserver gets cleared, when the sequence completes, although the queue is still needed:

reactive/Rx.NET/Source/src/System.Reactive/Linq/Observable/Zip.cs

Line 527 in 418a037

Dispose();

Commenting out the _values.Clear() line in Dispose() would make the tests of #827 pass, although it's good idea to clear them when all sequences are done. I think the easiest way would be to partially revert #635 and reinsert the nth disposable of the disposables array in the Zip.Run method :

 disposables[3] = Disposable.Create(() =>
                {
                    _observer1.Values.Clear();
                    _observer2.Values.Clear();
                    _observer3.Values.Clear();
                });

see:

https://github.com/dotnet/reactive/pull/635/files#diff-9e17b95d856736d153ecd78c6876ec83L71

[quinmars]

Hm, #828 looks even more simpler. 😊

[danielcweber]

On an abstract level, we might run into this type of problems anywhere, where an Observer inherits from SafeObserver and overrides Dispose. For immediate sources, when we did everything with SingleAssignmentDisposables, the order of execution was obviously different then. I'll give it a review. Thx @akarnokd for investigating this.

Update: Zip is indeed the only operator that uses a SafeObserver inherited class which overrides Dispose. When #828 gets merged, there are none at all and we might consider removing the virtual modifier on Dispose so we don't run into that issue in the future. @akarnokd Any thoughts ?