zip is not Rx compatible

[inamiy]

    func test() async throws {
        let now = ContinuousClock.now

        let countUpStream = AsyncTimerSequence(interval: .seconds(0.1), clock: .continuous)
            .map { now.duration(to: $0).components.seconds }
        let countUpStream2 = AsyncTimerSequence(interval: .seconds(0.3), clock: .continuous)
            .map { now.duration(to: $0).components.seconds }

        let zipped = zip(countUpStream, countUpStream2)

        for try await value in zipped {
            print("===> zip", value)
        }
    }

Result:

===> zip (0, 0)
===> zip (0, 0)
===> zip (0, 0)
===> zip (0, 1)
===> zip (1, 1)
===> zip (1, 1)
===> zip (1, 2)
===> zip (2, 2)
===> zip (2, 2)
===> zip (2, 3)
===> zip (3, 3)
===> zip (3, 3)
===> zip (3, 3)
===> zip (3, 4)
===> zip (4, 4)
===> zip (4, 4)
===> zip (4, 5)
===> zip (5, 5)
===> zip (5, 5)
===> zip (5, 6)
===> zip (6, 6)
===> zip (6, 6)
===> zip (6, 6)
===> zip (6, 7)
===> zip (7, 7)
===> zip (7, 7)
===> zip (7, 8)

To Be:

===> zip (0, 0)
===> zip (1, 1)
===> zip (2, 2)
...

[DevAndArtist]

I don't think the example is correct. zip operation has nothing to do with the actual equality of the emitted value. If the LHS sequence emits 0 multiple times, it's expected that the zip operation generate the result containing an equal (or less) amount of 0es.

func test_zip() async {
  let a = [1, 1, 2, 2]
  let b = ["a", "b", "c", "d"]
  let expected = Array(zip(a, b))
  let actual = await Array(zip(a.async, b.async))
  XCTAssertEqual(expected, actual)
}

// expected result: [(1, "a"), (1, "b"), (2, "c"), (2, "d")]

The marble diagram from ReactiveX demonstrates this behavior pretty well: https://reactivex.io/documentation/operators/zip.html

[inamiy]

@DevAndArtist
zip will need to take time-interval into consideration.
First example shows 0.1 sec and 0.3 sec difference, which are needed to take into account.

[DevAndArtist]

I don't understand. AsyncTimerSequences are only the producer sequences and have nothing to do with how zip should behave. The first sequence emits multiple 0ers, so does the second sequence, then zip is just zipping those 0ers together into a bunch of tuples.

--0--0--0---------------------------->
---------0---------0---------0------->

zip
--------(0,0)-----(0,0)------(0,0)--->

zip is only awaiting until each sequence emits at least one new value to perform zipping, nothing more nothing less.

[DevAndArtist]

If you need a result that looks more like this:

===> zip (0, 0)
===> zip (1, 1)
===> zip (2, 2)
...

then you'd need to extend both async sequences with something like removeDuplicates().

let zipped = zip(countUpStream.removeDuplicates(), countUpStream2.removeDuplicates())

[inamiy]

Ah, my bad. The example was a too complicated (using less than 1 sec interval that causes duplicated same seconds).
After fixing, I noticed other issue so I will close here and start over again.

[inamiy]

Added simpler example issue in:

• zip is not Rx compatible (take 2) #117