Merge pull request #11 from AsyncCommunity/fix/ConcurrentAccessRegulator

project: fix onNext scheduling in ConcurrentAccessRegulator

Author: twittemb

CHANGELOG.md

@@ -1,7 +1,12 @@
-**v0.2.0 **
+**v0.2.1 - Lithium:**
+
+- Enforce the call of onNext in a determinitisc way
+
+**v0.2.0 - Helium:**
 
 - AsyncStreams.CurrentValue `element` made public and available with get/set
 - new Multicast operator
+- new Assign operator
 
 **v0.1.0 - Hydrogen:**
 

Sources/Internal/ConcurrentAccessRegulator.swift

@@ -58,13 +58,12 @@ final class ConcurrentAccessRegulator<UpstreamAsyncSequence: AsyncSequence>: @un
 
         do {
             let next = try await self.upstreamAsyncIterator?.next()
-
+            await self.onNext(next)
+            
             await self.gate.unlock()
-
+            
             // yield allows to promote other tasks to resume, giving a chance to request a next element
             await Task.yield()
-
-            await self.onNext(next)
         } catch is CancellationError {
             await self.onCancel()
         } catch {

Tests/AsyncSequences/AsyncSequences+MergeTests.swift

@@ -48,19 +48,17 @@ private struct TimedAsyncSequence<Element>: AsyncSequence, AsyncIteratorProtocol
 
 final class AsyncSequences_MergeTests: XCTestCase {
     func testMerge_merges_sequences_according_to_the_timeline_using_asyncSequences() async throws {
-        // -- 0 ------------------------------- 1200 ---------------------------
-        // ------- 300 ------------- 900 ------------------------------ 1800 ---
-        // --------------- 600 --------------------------- 1500 ----------------
-        // -- a --- c ----- f ------- d --------- b -------- g ---------- e ----
+        // -- 0 ------------------------------- 1000 ----------------------------- 2000 -
+        // --------------- 500 --------------------------------- 1500 -------------------
+        // -- a ----------- d ------------------ b --------------- e --------------- c --
         //
-        // output should be: a c f d b g e
-        let expectedElements = ["a", "c", "f", "d", "b", "g", "e"]
+        // output should be: a, d, b, e, c
+        let expectedElements = ["a", "d", "b", "e", "c"]
 
-        let asyncSequence1 = TimedAsyncSequence(intervalInMills: [0, 1200], sequence: ["a", "b"])
-        let asyncSequence2 = TimedAsyncSequence(intervalInMills: [300, 600, 900], sequence: ["c", "d", "e"])
-        let asyncSequence3 = TimedAsyncSequence(intervalInMills: [600, 1100], sequence: ["f", "g"])
+        let asyncSequence1 = TimedAsyncSequence(intervalInMills: [0, 1000, 1000], sequence: ["a", "b", "c"])
+        let asyncSequence2 = TimedAsyncSequence(intervalInMills: [500, 1000], sequence: ["d", "e"])
 
-        let sut = AsyncSequences.Merge(asyncSequence1, asyncSequence2, asyncSequence3)
+        let sut = AsyncSequences.Merge(asyncSequence1, asyncSequence2)
 
         var receivedElements = [String]()
         for try await element in sut {
@@ -187,7 +185,6 @@ final class AsyncSequences_MergeTests: XCTestCase {
             var receivedElements = [Int]()
             do {
                 for try await element in sut {
-                    print("Received element \(element)")
                     receivedElements.append(element)
                     if element == 1 {
                         canSend2Expectation.fulfill()