Update

Sources/Verge/Store/Accumulator.swift

@@ -1,56 +1,58 @@
 
-public protocol Sink {
+public protocol Sink<Source> {
   associatedtype Source
   func receive(source: Source)
 }
 
 public struct AccumulationBuilder<Source>: ~Copyable {
 
-  public func ifChanged<U: Equatable>(_ selector: @escaping (Source) -> U) -> IfChangedSink<U> {
+  public func ifChanged<U: Equatable>(_ selector: @escaping (Source) -> U) -> SinkIfChanged<Source, U> {
     .init(selector: selector)
   }
 
-  public final class IfChangedSink<Target: Equatable>: Sink {
+}
 
-    private let selector: (Source) -> Target
+public final class SinkIfChanged<Source, Target: Equatable>: Sink {
 
-    private var latestValue: Target?
-    private var handler: ((consuming Target) -> Void)?
+  private let selector: (Source) -> Target
 
-    public init(selector: @escaping (Source) -> Target) {
-      self.selector = selector
-    }
+  private var latestValue: Target?
+  private var handler: ((consuming Target) -> Void)?
 
-    public func `do`(_ perform: @escaping (consuming Target) -> Void) -> Self {
-      self.handler = perform
-      return self
-    }
+  public init(selector: @escaping (Source) -> Target) {
+    self.selector = selector
+  }
 
-    public func receive(source: Source) {
+  public func `do`(_ perform: @escaping (consuming Target) -> Void) -> Self {
+    self.handler = perform
+    return self
+  }
 
-      let selected = selector(source)
+  public func receive(source: Source) {
 
-      guard latestValue != selected else {
-        return
-      }
+    let selected = selector(source)
 
-      latestValue = selected
+    guard latestValue != selected else {
+      return
+    }
 
-      handler?(selected)
+    latestValue = selected
 
-    }
-  }
+    handler?(selected)
 
+  }
 }
 
 extension DispatcherType {
 
-  public func accumulate(@SinkComponentBuilder<Scope> _ buildSubscription: (consuming AccumulationBuilder<Scope>) -> SinkGroup<Scope>) -> Cancellable {
+  public func accumulate(
+    queue: MainActorTargetQueue = .mainIsolated(),
+    @SinkComponentBuilder<Scope> _ buildSubscription: (consuming AccumulationBuilder<Scope>) -> SinkGroup<Scope>) -> Cancellable {
 
     let builder = AccumulationBuilder<Scope>()
     let group = buildSubscription(consume builder)
 
-    return sinkState { state in
+    return sinkState(dropsFirst: false, queue: queue) { state in
 
       group.receive(source: state.primitive)
 
@@ -60,6 +62,19 @@ extension DispatcherType {
 
 }
 
+public struct SinkBox<Source>: Sink {
+
+  private let base: any Sink<Source>
+
+  public init(base: some Sink<Source>) {
+    self.base = base
+  }
+
+  public func receive(source: Source) {
+    base.receive(source: source)
+  }
+}
+
 public struct SinkGroup<Source>: Sink {
 
   private let _receive: (Source) -> Void
@@ -79,16 +94,30 @@ public struct SinkComponentBuilder<Source> {
   public static func buildBlock() -> SinkGroup<Source> {
     return .init(receive: { _ in })
   }
+//
+//  public static func buildBlock<each Target>(_ components: repeat AccumulationBuilder<Source>.IfChangedSink<each Target>) -> SinkGroup<Source> {
+//    .init { source in
+//
+//      func run<T>(_ component: AccumulationBuilder<Source>.IfChangedSink<T>) {
+//        component.receive(source: source)
+//      }
+//
+//      repeat run(each components)
+//
+//    }
+//  }
+
+  public static func buildExpression(_ expression: any Sink<Source>) -> SinkBox<Source> {
+    .init(base: expression)
+  }
 
-  public static func buildBlock<each Target>(_ components: repeat AccumulationBuilder<Source>.IfChangedSink<each Target>) -> SinkGroup<Source> {
+  public static func buildBlock(_ sinks: (SinkBox<Source>)...) -> SinkGroup<Source> {
     .init { source in
 
-      func run<T>(_ component: AccumulationBuilder<Source>.IfChangedSink<T>) {
-        component.receive(source: source)
+      for sink in sinks {
+        sink.receive(source: source)
       }
 
-      repeat run(each components)
-
     }
   }
 

Tests/VergeTests/AccumulationTests.swift

@@ -22,6 +22,10 @@ final class AccumulationTests: XCTestCase {
 
       }
 
+      SinkIfChanged(selector: \.count).do { value in
+        print(value)
+      }
+
     }
 
     store.commit {