Merge branch 'feature/streamed'

Author: Thibault Wittemberg

CHANGELOG.md

@@ -1,3 +1,8 @@
+**v0.4.0 - Bore:**
+
+- AsyncStreams: new @Streamed property wrapper
+- AsyncSequences: finish Timer when canceled
+
 **v0.3.0 - Beryllium:**
 
 - Operators: new Share operator

README.md

@@ -35,6 +35,7 @@ AsyncSequences
 * [Passthrough](#Passthrough)
 * [CurrentValue](#CurrentValue)
 * [Replay](#Replay)
+* [Streamed](#Streamed)
 
 ### Operators
 * [Collect](#Collect)
@@ -297,6 +298,34 @@ for try await element in replay {
 }
 ```
 
+### Streamed
+
+`Streamed` is a property wrapper that streams a property as an AsyncSequence. It is a structured concurrency equivalent to **Combine @Published**.
+
+```swift
+class Weather {
+    @Streamed var temperature: Double
+    init(temperature: Double) {
+        self.temperature = temperature
+    }
+}
+
+let weather = Weather(temperature: 20)
+Task {
+    for try await element in weather.$temperature {
+        print ("Temperature now: \(element)")
+    }
+}
+
+// ... later in the application flow
+
+weather.temperature = 25
+
+// will print:
+// Temperature now: 20.0
+// Temperature now: 25.0
+```
+
 ## Operators
 
 ### Collect

Sources/AsyncStreams/Streamed.swift

@@ -0,0 +1,60 @@
+//
+//  Streamed.swift
+//  
+//
+//  Created by Thibault Wittemberg on 20/03/2022.
+//
+
+/// A type that streams a property marked with an attribute as an AsyncSequence.
+///
+/// Streaming a property with the `@Streamed` attribute creates an AsyncSequence of this type. You access the AsyncSequence with the `$` operator, as shown here:
+///
+///     class Weather {
+///         @Streamed var temperature: Double
+///         init(temperature: Double) {
+///             self.temperature = temperature
+///         }
+///     }
+///
+///     let weather = Weather(temperature: 20)
+///     Task {
+///         for try await element in weather.$temperature {
+///             print ("Temperature now: \(element)")
+///         }
+///     }
+///
+///     // ... later in the application flow
+///
+///     weather.temperature = 25
+///
+///     // Prints:
+///     // Temperature now: 20.0
+///     // Temperature now: 25.0
+@propertyWrapper public struct Streamed<Element> {
+    let currentValue: AsyncStreams.CurrentValue<Element>
+
+    /// Creates the streamed instance with an initial wrapped value.
+    ///
+    /// Don't use this initializer directly. Instead, create a property with the `@Streamed` attribute, as shown here:
+    ///
+    ///     @Streamed var lastUpdated: Date = Date()
+    ///
+    /// - Parameter wrappedValue: The stream's initial value.
+    public init(wrappedValue: Element) {
+        self.currentValue = AsyncStreams.CurrentValue<Element>(wrappedValue)
+        self.wrappedValue = wrappedValue
+    }
+
+    public var wrappedValue: Element {
+        willSet {
+            self.currentValue.element = newValue
+        }
+    }
+
+    /// The property for which this instance exposes an AsyncSequence.
+    ///
+    /// The ``Streamed/projectedValue`` is the property accessed with the `$` operator.
+    public var projectedValue: AnyAsyncSequence<Element> {
+        self.currentValue.eraseToAnyAsyncSequence()
+    }
+}

Tests/AsyncStreams/StreamedTests.swift

@@ -0,0 +1,53 @@
+//
+//  StreamedTests.swift
+//  
+//
+//  Created by Thibault Wittemberg on 20/03/2022.
+//
+
+import AsyncExtensions
+import Combine
+import XCTest
+
+final class StreamedTests: XCTestCase {
+    @Streamed
+    var sut: Int = 0
+
+    func test_streamed_gets_and_sets_element() {
+        XCTAssertEqual(sut, 0)
+        let newValue = Int.random(in: 0...100)
+        self.sut = newValue
+        XCTAssertEqual(sut, newValue)
+    }
+
+    func test_streamed_projects_in_asyncSequence() {
+        let firstElementIsReceivedExpectation = expectation(description: "The first element has been received")
+        let fifthElementIsReceivedExpectation = expectation(description: "The fifth element has been received")
+
+        let expectedElements = [0, 1, 2, 3, 4]
+        let task = Task(priority: .high) {
+            var receivedElements = [Int]()
+            for try await element in self.$sut {
+                receivedElements.append(element)
+
+                if element == 0 {
+                    firstElementIsReceivedExpectation.fulfill()
+                }
+                if element == 4 {
+                    fifthElementIsReceivedExpectation.fulfill()
+                    XCTAssertEqual(receivedElements, expectedElements)
+                }
+            }
+        }
+
+        wait(for: [firstElementIsReceivedExpectation], timeout: 1)
+
+        sut = 1
+        sut = 2
+        sut = 3
+        sut = 4
+
+        wait(for: [fifthElementIsReceivedExpectation], timeout: 1)
+        task.cancel()
+    }
+}

Tests/Operators/AsyncSequence+MulticastTests.swift

@@ -73,7 +73,7 @@ final class AsyncSequence_MulticastTests: XCTestCase {
         let stream = AsyncStreams.Passthrough<Int>()
         let sut = spyUpstreamSequence.multicast(stream)
 
-        Task {
+        Task(priority: .high) {
             var receivedElement = [Int]()
             for try await element in sut {
                 receivedElement.append(element)
@@ -82,7 +82,7 @@ final class AsyncSequence_MulticastTests: XCTestCase {
             tasksHaveFinishedExpectation.fulfill()
         }
 
-        Task {
+        Task(priority: .high) {
             var receivedElement = [Int]()
             for try await element in sut {
                 receivedElement.append(element)