Added unwrappedStream() method to AsyncObservable that allows you to read a stream of only non-nil values when the type is Optional.

Author: meech-ward

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 All notable changes to AsyncObservable will be documented in this file.
 
+## [0.3.1] - 2025-03-08
+
+### Added
+- Added unwrappedStream() method to AsyncObservable that allows you to read a stream of only non-nil values when the type is Optional.
+
 ## [0.3.0] - 2025-03-08
 
 ### Changed

README.md

@@ -111,6 +111,19 @@ for await value in someProperty.stream {
 }
 ```
 
+## Unwrapped Stream
+
+If you want to read a stream of non-nil values, but your type is an optional, you can use the `unwrappedStream` method.
+
+```swift
+let someProperty = AsyncObservable(Data?)
+let stream = someProperty.unwrappedStream()
+
+for await value in stream {
+  print(value) // only non-nil values
+}
+```
+
 ## Mutate
 
 Sometimes you just want to mutate the original value instead of having to copy and return a new value. This still updates all the observers correctly and is safe.

Sources/AsyncObservable/AsyncObservable.swift

@@ -1,8 +1,8 @@
-import Foundation
 import Dispatch
+import Foundation
 
 #if canImport(Observation)
-import Observation
+  import Observation
 #endif
 
 /// A thread-safe state management system that provides both async stream-based observation
@@ -85,7 +85,7 @@ open class AsyncObservable<T: Sendable>: AsyncObservableReadOnly, @unchecked Sen
 
   /// An async stream of values that can be used with Swift concurrency.
   /// This property provides a convenient way to access the value stream without calling `stream()`.
-  public var stream: StreamOf<T> {
+  open var stream: StreamOf<T> {
     streamOf()
   }
 
@@ -152,7 +152,7 @@ open class AsyncObservable<T: Sendable>: AsyncObservableReadOnly, @unchecked Sen
   }
 
   /// Creates a new state manager with the given initial value.
-  /// This initializer can be called from any context 
+  /// This initializer can be called from any context
   ///
   /// - Parameters:
   ///   - initialValue: The initial value to manage
@@ -162,8 +162,7 @@ open class AsyncObservable<T: Sendable>: AsyncObservableReadOnly, @unchecked Sen
     _ initialValue: T,
     bufferingPolicy: AsyncStream<T>.Continuation.BufferingPolicy = .unbounded,
     serialQueue: DispatchQueue = DispatchQueue(label: "AsyncObservable")
-  )
-  {
+  ) {
     _raw = initialValue
     self.bufferingPolicy = bufferingPolicy
     self.serialQueue = serialQueue
@@ -237,7 +236,7 @@ open class AsyncObservable<T: Sendable>: AsyncObservableReadOnly, @unchecked Sen
   /// The stream immediately yields the current value and then yields all subsequent updates.
   ///
   /// - Returns: A StreamOf<T> instance that can be used with async/await code
-  private func streamOf() -> StreamOf<T> {
+  internal func streamOf() -> StreamOf<T> {
     let id = UUID()
     return StreamOf<T>(
       bufferingPolicy: bufferingPolicy,
@@ -251,4 +250,10 @@ open class AsyncObservable<T: Sendable>: AsyncObservableReadOnly, @unchecked Sen
         continuation.yield(self.raw)
       })
   }
+
+  @available(iOS 17.0, macOS 15.0, tvOS 17.0, watchOS 10.0, *)
+  func unwrappedStream<U>() -> any AsyncSequence<U, Never> where T == U? {
+    stream.compactMap { $0 }
+  }
+
 }

Tests/AsyncObservableTests/StreamTests.swift

@@ -1,5 +1,6 @@
 import Foundation
 import Testing
+
 @testable import AsyncObservable
 
 @Suite("AsyncObservable Stream Tests")
@@ -36,7 +37,7 @@ struct AsyncObservableStreamTests {
 
     #expect(receivedValues == [0, 1, 2])
   }
-  
+
   @Test("Should handle multiple observers")
   @available(iOS 17.0, macOS 14.0, tvOS 17.0, watchOS 10.0, *)
   func testMultipleObservers() async {
@@ -45,7 +46,7 @@ struct AsyncObservableStreamTests {
     // Create two separate streams
     let stream1 = observable.stream
     let stream2 = observable.stream
-    
+
     // Consume initial values first
     var values1: [Int] = []
     var values2: [Int] = []
@@ -56,7 +57,7 @@ struct AsyncObservableStreamTests {
         values1.append(value)
       }
     }
-    
+
     let task2 = Task {
       for await value in stream2 {
         values2.append(value)
@@ -90,7 +91,7 @@ struct AsyncObservableStreamTests {
 
     // Now start observing - we should get the current value (4) and possibly some buffered values
     var receivedValues: [Int] = []
-    
+
     // Create a task with a timeout to avoid hanging
     let task = Task {
       for await value in observable.stream {
@@ -100,7 +101,7 @@ struct AsyncObservableStreamTests {
         }
       }
     }
-    
+
     // Add a timeout to ensure the test doesn't hang if the stream doesn't yield enough values
     try? await Task.sleep(nanoseconds: 100_000_000)
     task.cancel()
@@ -109,7 +110,7 @@ struct AsyncObservableStreamTests {
     #expect(receivedValues.count > 0)
     #expect(receivedValues.contains(4), "Should contain the latest value")
   }
-  
+
   @Test("Should not update observers when notifyObservers is false")
   @available(iOS 17.0, macOS 14.0, tvOS 17.0, watchOS 10.0, *)
   func testSilentUpdate() async {
@@ -140,9 +141,39 @@ struct AsyncObservableStreamTests {
 
     // No value should have been received
     #expect(receivedValue == false)
-    
+
     // But the internal value should be updated
     #expect(observable.raw == 20)
     await #expect(observable.observable == 20)
   }
-} 
+
+  @Test("unsrapped steram")
+  @available(iOS 17.0, macOS 15.0, tvOS 17.0, watchOS 10.0, *)
+  func testUnrappedStream() async {
+    let observable: AsyncObservable<Int?> = AsyncObservable(nil)
+    let stream = observable.unwrappedStream()
+
+    // should only get values that are not nil
+    var sum = 0
+    let task = Task {
+      for await value in stream {
+        sum += value
+        #expect(value != nil)
+      }
+    }
+
+    observable.update(nil)
+    observable.update(1)
+    observable.update(2)
+    observable.update(nil)
+    observable.update(3)
+    observable.update(nil)
+    observable.update(5)
+
+    // Give some time for potential values to arrive
+    try? await Task.sleep(for: .milliseconds(100))
+    task.cancel()
+
+    #expect(sum == 11)
+  }
+}