Update

Author: muukii

Sources/Verge/Library/EventEmitter.swift

@@ -21,9 +21,9 @@
 
 import Atomics
 import Combine
+import DequeModule
 import Foundation
 import os
-import DequeModule
 
 public final class EventEmitterCancellable: Hashable, Cancellable, @unchecked Sendable {
 
@@ -57,11 +57,12 @@ public protocol EventEmitterEventType {
 /// Instead of Combine
 open class EventEmitter<Event: EventEmitterEventType>: EventEmitterType, @unchecked Sendable {
 
-  public var publisher: some Publisher<Event, Never> {
-    self
+  public var publisher: Publisher {
+    return .init(eventEmitter: self)
   }
 
-  private var subscribers: VergeConcurrency.UnfairLockAtomic<[EventEmitterCancellable : (Event) -> Void]> = .init([:])
+  private var subscribers:
+    VergeConcurrency.UnfairLockAtomic<[EventEmitterCancellable: (Event) -> Void]> = .init([:])
 
   private let queue: VergeConcurrency.UnfairLockAtomic<Deque<Event>> = .init(.init())
 
@@ -70,7 +71,6 @@ open class EventEmitter<Event: EventEmitterEventType>: EventEmitterType, @unchec
   private var deinitHandlers: VergeConcurrency.UnfairLockAtomic<[() -> Void]> = .init([])
 
   public init() {
-
   }
 
   deinit {
@@ -106,10 +106,10 @@ open class EventEmitter<Event: EventEmitterEventType>: EventEmitterType, @unchec
           return nil
         }
       }) {
-        
+
         // Emits
         receiveEvent(event)
-        
+
         for subscriber in capturedSubscribers {
           vergeSignpostEvent("EventEmitter.emitForSubscriber")
           subscriber.1(event)
@@ -128,14 +128,15 @@ open class EventEmitter<Event: EventEmitterEventType>: EventEmitterType, @unchec
     }
 
   }
-  
+
   open func receiveEvent(_ event: consuming Event) {
 
   }
 
   @_spi(EventEmitter)
   @discardableResult
-  public func addEventHandler(_ eventReceiver: @escaping (Event) -> Void) -> EventEmitterCancellable {
+  public func addEventHandler(_ eventReceiver: @escaping (Event) -> Void) -> EventEmitterCancellable
+  {
     let token = EventEmitterCancellable(owner: self)
     subscribers.modify {
       $0[token] = eventReceiver
@@ -157,7 +158,7 @@ open class EventEmitter<Event: EventEmitterEventType>: EventEmitterType, @unchec
     // then unfair-lock raises runtime error.
     withExtendedLifetime(itemToRemove, {})
   }
-  
+
   public func onDeinit(_ onDeinit: @escaping () -> Void) {
     deinitHandlers.modify {
       $0.append(onDeinit)
@@ -166,28 +167,29 @@ open class EventEmitter<Event: EventEmitterEventType>: EventEmitterType, @unchec
 
 }
 
-extension EventEmitter: Publisher {
-  
-  public typealias Output = Event
+extension EventEmitter {
 
-  public typealias Failure = Never
+  @available(iOS 13, macOS 10.15, tvOS 13, watchOS 6, *)
+  public struct Publisher: Combine.Publisher {
 
-  public func receive<S>(
-    subscriber: S
-  )
-  where S: Subscriber, Failure == S.Failure, Output == S.Input {
+    public typealias Output = Event
 
-    let subscription = Subscription<S>(
-      subscriber: subscriber,
-      eventEmitter: self
-    )
-    
-    subscriber.receive(subscription: subscription)
-  }
-  
-}
+    public typealias Failure = Never
 
-extension EventEmitter {
+    private weak var eventEmitter: EventEmitter<Event>?
+
+    public init(eventEmitter: EventEmitter<Event>) {
+      self.eventEmitter = eventEmitter
+    }
+
+    public func receive<S>(subscriber: S)
+    where S: Subscriber, Failure == S.Failure, Output == S.Input {
+
+      let subscription = Subscription<S>(subscriber: subscriber, eventEmitter: eventEmitter)
+      subscriber.receive(subscription: subscription)
+    }
+
+  }
 
   @available(iOS 13, macOS 10.15, tvOS 13, watchOS 6, *)
   public struct Subscription<S: Subscriber>: Combine.Subscription where S.Input == Event {
@@ -198,22 +200,15 @@ extension EventEmitter {
     private let eventEmitterSubscription: EventEmitterCancellable?
     private weak var eventEmitter: EventEmitter<Event>?
 
-    init(
-      subscriber: S,
-      eventEmitter: EventEmitter<Event>?
-    ) {
+    init(subscriber: S, eventEmitter: EventEmitter<Event>?) {
 
       self.subscriber = subscriber
       self.eventEmitter = eventEmitter
-
-      eventEmitter?.onDeinit {        
-        subscriber.receive(completion: .finished)        
-      }
-          
+      
       self.eventEmitterSubscription = eventEmitter?
         .addEventHandler { (event) in
           _ = subscriber.receive(event)
-      }
+        }
     }
 
     public func request(_ demand: Subscribers.Demand) {

Sources/Verge/Store/Store+Combine.swift

@@ -19,8 +19,6 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 // THE SOFTWARE.
 
-#if canImport(Combine)
-
 import Combine
 
 @available(iOS 13, macOS 10.15, tvOS 13, watchOS 6, *)
@@ -35,16 +33,25 @@ extension Store {
   @_spi(Package)
   public func _statePublisher() -> some Combine.Publisher<Changes<Value>, Never> {
 
-    return valuePublisher
-      .dropFirst()
+    return
+      publisher
       .associate(resource: self, retains: keepsAliveForSubscribers)
+      .flatMap { event in
+        guard case .state(.didUpdate(let state)) = event else {
+          return Empty<Changes<Value>, Never>().eraseToAnyPublisher()
+        }
+        return Just<Changes<Value>>(state)
+          .eraseToAnyPublisher()
+      }
       .merge(with: Just(state.droppedPrevious()))
+
   }
 
-//  @_spi(Package)
+  //  @_spi(Package)
   public func _activityPublisher() -> some Combine.Publisher<Activity, Never> {
 
-    return publisher
+    return
+      publisher
       .associate(resource: self, retains: keepsAliveForSubscribers)
       .flatMap { event in
         guard case .activity(let a) = event else {
@@ -70,7 +77,7 @@ extension Publisher {
 
 }
 
-fileprivate final class ResourceBox {
+private final class ResourceBox {
 
   private let object: AnyObject?
 
@@ -82,5 +89,3 @@ fileprivate final class ResourceBox {
     }
   }
 }
-
-#endif

Sources/Verge/Store/Store.swift

@@ -125,10 +125,6 @@ open class Store<State, Activity: Sendable>: EventEmitter<_StoreEvent<State, Act
 
   public let sanitizer: RuntimeSanitizer
 
-  public var valuePublisher: some Combine.Publisher<Changes<State>, Never> {
-    return _valueSubject
-  }
-
   private var middlewares: [AnyStoreMiddleware<State>] = []
 
   private let externalOperation: @Sendable (inout InoutRef<State>, Changes<State>, inout Transaction) -> Void
@@ -137,8 +133,6 @@ open class Store<State, Activity: Sendable>: EventEmitter<_StoreEvent<State, Act
 
   private let _lock: StoreOperation
 
-  private let _valueSubject: CurrentValueSubject<Changes<State>, Never>
-
   /**
    Holds subscriptions for sink State and Activity to finish them with its store life-cycle.
    */
@@ -180,9 +174,6 @@ open class Store<State, Activity: Sendable>: EventEmitter<_StoreEvent<State, Act
     self.nonatomicValue = .init(old: nil, new: initialState)
     self._lock = storeOperation
 
-    // TODO: copying value
-    self._valueSubject = .init(nonatomicValue)
-
     self.logger = logger
     self.sanitizer = sanitizer ?? RuntimeSanitizer.global
     self.name = name ?? "\(file):\(line)"
@@ -215,8 +206,6 @@ open class Store<State, Activity: Sendable>: EventEmitter<_StoreEvent<State, Act
 
     self.nonatomicValue = .init(old: nil, new: reduced)
     self._lock = storeOperation
-    // TODO: copying value
-    self._valueSubject = .init(nonatomicValue)
 
     self.logger = logger
     self.sanitizer = sanitizer ?? RuntimeSanitizer.global
@@ -249,7 +238,6 @@ open class Store<State, Activity: Sendable>: EventEmitter<_StoreEvent<State, Act
       case .willUpdate:
         break
       case .didUpdate(let state):
-        _valueSubject.send(state)
         stateDidUpdate(newState: state)
       }
     case .activity:
@@ -277,11 +265,7 @@ open class Store<State, Activity: Sendable>: EventEmitter<_StoreEvent<State, Act
 
     storeLifeCycleCancellable.cancel()
 
-    Task { [taskManager, _valueSubject] in
-      // send completion in hop as Combine is using unfair lock (non-recursive). Avoid crash.
-      // It happens if the stream ratains this store, canceled that stream triggers this deinit operation.
-      // that deinit operation will be inside of locking session.
-      _valueSubject.send(completion: .finished)
+    Task { [taskManager] in
       await taskManager.cancelAll()
     }
   }

Tests/VergeTests/PropertyWrapperTests.swift

@@ -13,12 +13,10 @@ final class SubjectCompletionTests: XCTestCase {
     var store: DemoStore? = DemoStore()
     weak var weakStore: DemoStore? = store
 
-    let exp = expectation(description: "completion")
-
     store?.statePublisher()
       .sink(
         receiveCompletion: { _ in
-          exp.fulfill()
+          XCTFail()
         },
         receiveValue: { _ in }
       )
@@ -30,7 +28,6 @@ final class SubjectCompletionTests: XCTestCase {
 
     XCTAssertNil(weakStore)
 
-    wait(for: [exp], timeout: 10)
   }
 
   func testActivityPublisherCompletion1() {
@@ -40,13 +37,11 @@ final class SubjectCompletionTests: XCTestCase {
     var store: DemoStore? = DemoStore()
     weak var weakStore: DemoStore? = store
 
-    let exp = expectation(description: "completion")
-
     store?
       .activityPublisher()
       .sink(
         receiveCompletion: { _ in
-          exp.fulfill()
+          XCTFail()
         },
         receiveValue: { _ in
 
@@ -62,7 +57,6 @@ final class SubjectCompletionTests: XCTestCase {
     bag.forEach { $0.cancel() }
     XCTAssertNil(weakStore)
 
-    wait(for: [exp], timeout: 10)
   }
 
   func testActivityPublisherCompletion2() {
@@ -103,13 +97,10 @@ final class SubjectCompletionTests: XCTestCase {
     var strongRef: Ref? = Ref()
     weak var weakRef = strongRef
 
-    let e = expectation(description: "completion")
-
     store!.activityPublisher()
       .sink(
         receiveCompletion: { _ in
-          print("")
-          e.fulfill()
+          XCTFail()
         },
         receiveValue: { _ in
           print("")
@@ -125,7 +116,6 @@ final class SubjectCompletionTests: XCTestCase {
 
     XCTAssertNil(weakRef)
 
-    wait(for: [e], timeout: 10)
   }
 
   func testDerived_publisher_retains_derived() {
@@ -172,16 +162,13 @@ final class SubjectCompletionTests: XCTestCase {
 
     XCTAssertNil(storeRef.value)
 
-    let onComplete = expectation(description: "onComplete")
-
     c = derivedRef.value!
       .statePublisher()
       .sink(
         receiveCompletion: { _ in
-          onComplete.fulfill()
+          XCTFail()
         },
         receiveValue: { _ in
-
         }
       )
 
@@ -193,7 +180,6 @@ final class SubjectCompletionTests: XCTestCase {
 
     XCTAssertNil(storeRef.value)
 
-    await fulfillment(of: [onComplete], timeout: 10)
     withExtendedLifetime(c, {})
 //    c?.cancel()