Add CombineLatest, remove unnecessary side effects

Playgrounds/02 - Continuation Passing Style.playground/contents.xcplayground

@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<playground version='5.0' target-platform='ios' buildActiveScheme='true' executeOnSourceChanges='false' importAppTypes='true'>
+    <timeline fileName='timeline.xctimeline'/>
+</playground>

Playgrounds/FreeCombine.playground/Pages/00a-Introduction (Examples).xcplaygroundpage/Contents.swift → Playgrounds/03 - FreeCombine.playground/Pages/00a-Introduction (Examples).xcplaygroundpage/Contents.swift

@@ -55,8 +55,8 @@ import FreeCombine
 import _Concurrency
 
 Task {
-    let fsubject1 = await FreeCombine.PassthroughSubject(Int.self)
-    let fsubject2 = await FreeCombine.PassthroughSubject(String.self)
+    let fsubject1 = await PassthroughSubject(Int.self)
+    let fsubject2 = await PassthroughSubject(String.self)
 
     let fseq1 = "abcdefghijklmnopqrstuvwxyz".asyncPublisher
     let fseq2 = (1 ... 100).asyncPublisher

Playgrounds/04 - Publishers.playground/Contents.swift

@@ -0,0 +1,3 @@
+import UIKit
+
+var greeting = "Hello, playground"

Playgrounds/04 - Publishers.playground/timeline.xctimeline

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Timeline
+   version = "3.0">
+   <TimelineItems>
+   </TimelineItems>
+</Timeline>

Playgrounds/05 - Operators.playground/Contents.swift

@@ -0,0 +1,3 @@
+import UIKit
+
+var greeting = "Hello, playground"

Playgrounds/05 - Operators.playground/Pages/01-Map.xcplaygroundpage/Contents.swift

@@ -0,0 +1,7 @@
+//: [Previous](@previous)
+
+import Foundation
+
+var greeting = "Hello, playground"
+
+//: [Next](@next)

Playgrounds/05 - Operators.playground/Pages/02-FlatMap.xcplaygroundpage/Contents.swift

@@ -0,0 +1,7 @@
+//: [Previous](@previous)
+
+import Foundation
+
+var greeting = "Hello, playground"
+
+//: [Next](@next)

Playgrounds/05 - Operators.playground/Pages/03-Zip.xcplaygroundpage/Contents.swift

@@ -0,0 +1,7 @@
+//: [Previous](@previous)
+
+import Foundation
+
+var greeting = "Hello, playground"
+
+//: [Next](@next)

Playgrounds/05 - Operators.playground/Pages/04-Reduce.xcplaygroundpage/Contents.swift

@@ -0,0 +1,7 @@
+//: [Previous](@previous)
+
+import Foundation
+
+var greeting = "Hello, playground"
+
+//: [Next](@next)

Playgrounds/05 - Operators.playground/Pages/05-Filter.xcplaygroundpage/Contents.swift

@@ -0,0 +1,7 @@
+//: [Previous](@previous)
+
+import Foundation
+
+var greeting = "Hello, playground"
+
+//: [Next](@next)

Playgrounds/05 - Operators.playground/contents.xcplayground

@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<playground version='5.0' target-platform='ios' buildActiveScheme='true' executeOnSourceChanges='false' importAppTypes='true'>
+    <timeline fileName='timeline.xctimeline'/>
+</playground>

Playgrounds/05 - Operators.playground/timeline.xctimeline

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Timeline
+   version = "3.0">
+   <TimelineItems>
+   </TimelineItems>
+</Timeline>

Sources/FreeCombine/Channel/Channel.swift

@@ -19,13 +19,11 @@ public struct Channel<Element>: AsyncSequence {
 
     public init(
         _: Element.Type = Element.self,
-        buffering: AsyncStream<Element>.Continuation.BufferingPolicy = .bufferingOldest(1),
-        onTermination: (@Sendable (AsyncStream<Element>.Continuation.Termination) -> Void)? = .none
+        buffering: AsyncStream<Element>.Continuation.BufferingPolicy = .bufferingOldest(1)
     ) {
         var localContinuation: AsyncStream<Element>.Continuation! = .none
         stream = .init(bufferingPolicy: buffering) { continuation in
             localContinuation = continuation
-            localContinuation.onTermination = onTermination
         }
         continuation = localContinuation
     }
@@ -79,7 +77,6 @@ public extension Channel {
 
     func stateTask<State>(
         initialState: @escaping (Self) async -> State,
-        onCancel: @Sendable @escaping () -> Void = { },
         reducer: Reducer<State, Self.Element>
     ) async -> StateTask<State, Self.Element> {
         var stateTask: StateTask<State, Self.Element>!
@@ -88,7 +85,6 @@ public extension Channel {
                 channel: self,
                 initialState: initialState,
                 onStartup: stateTaskContinuation,
-                onCancel: onCancel,
                 reducer: reducer
             )
         }
@@ -98,14 +94,12 @@ public extension Channel {
     func stateTask<State>(
         initialState: @escaping (Self) async -> State,
         onStartup: UnsafeContinuation<Void, Never>? = .none,
-        onCancel: @Sendable @escaping () -> Void = { },
         reducer: Reducer<State, Self.Element>
     ) -> StateTask<State, Self.Element> {
         .init(
             channel: self,
             initialState: initialState,
             onStartup: onStartup,
-            onCancel: onCancel,
             reducer: reducer
         )
     }

Sources/FreeCombine/Combinator/CombineLatestState.swift

@@ -0,0 +1,120 @@
+//
+//  CombineLatestState.swift
+//  
+//
+//  Created by Van Simmons on 6/4/22.
+//
+struct CombineLatestState<Left: Sendable, Right: Sendable>: CombinatorState {
+    typealias CombinatorAction = Self.Action
+    enum Action {
+        case setLeft(AsyncStream<Left>.Result, UnsafeContinuation<Demand, Swift.Error>)
+        case setRight(AsyncStream<Right>.Result, UnsafeContinuation<Demand, Swift.Error>)
+    }
+
+    let downstream: (AsyncStream<(Left?, Right?)>.Result) async throws -> Demand
+    let leftCancellable: Cancellable<Demand>
+    let rightCancellable: Cancellable<Demand>
+
+    var mostRecentDemand: Demand
+    var left: (value: Left?, continuation: UnsafeContinuation<Demand, Swift.Error>)? = .none
+    var right: (value: Right?, continuation: UnsafeContinuation<Demand, Swift.Error>)? = .none
+    var leftComplete = false
+    var rightComplete = false
+
+    init(
+        channel: Channel<CombineLatestState<Left, Right>.Action>,
+        downstream: @escaping (AsyncStream<(Left?, Right?)>.Result) async throws -> Demand,
+        mostRecentDemand: Demand = .more,
+        left: Publisher<Left>,
+        right: Publisher<Right>
+    ) async {
+        self.downstream = downstream
+        self.mostRecentDemand = mostRecentDemand
+        self.leftCancellable = await channel.consume(publisher: left, using: CombineLatestState<Left, Right>.Action.setLeft)
+        self.rightCancellable = await channel.consume(publisher: right, using: CombineLatestState<Left, Right>.Action.setRight)
+    }
+
+    static func create(
+        mostRecentDemand: Demand = .more,
+        left: Publisher<Left>,
+        right: Publisher<Right>
+    ) -> (@escaping (AsyncStream<(Left?, Right?)>.Result) async throws -> Demand) -> (Channel<CombineLatestState<Left, Right>.Action>) async -> Self {
+        { downstream in { channel in
+            await .init(channel: channel, downstream: downstream, left: left, right: right)
+        } }
+    }
+
+    static func complete(state: inout Self, completion: Reducer<Self, Self.Action>.Completion) async -> Void {
+        state.leftCancellable.cancel()
+        state.left?.continuation.resume(returning: .done)
+        state.rightCancellable.cancel()
+        state.right?.continuation.resume(returning: .done)
+        switch completion {
+            case .cancel:
+                _ = try? await state.downstream(.completion(.cancelled))
+            case .exit, .termination:
+                _ = try? await state.downstream(.completion(.finished))
+            case let .failure(error):
+                _ = try? await state.downstream(.completion(.failure(error)))
+        }
+    }
+
+    static func reduce(`self`: inout Self, action: Self.Action) async throws -> Reducer<Self, Action>.Effect {
+        guard !Task.isCancelled else { return .completion(.cancel) }
+        return try await `self`.reduce(action: action)
+    }
+
+    private mutating func reduce(
+        action: Self.Action
+    ) async throws -> Reducer<Self, Action>.Effect {
+        switch action {
+            case let .setLeft(leftResult, leftContinuation):
+                if mostRecentDemand == .done { leftContinuation.resume(returning: .done) }
+                else { return try await handleLeft(leftResult, leftContinuation) }
+            case let .setRight(rightResult, rightContinuation):
+                if mostRecentDemand == .done { rightContinuation.resume(returning: .done) }
+                else { return try await handleRight(rightResult, rightContinuation) }
+        }
+        return .none
+    }
+
+    private mutating func handleLeft(
+        _ leftResult: AsyncStream<Left>.Result,
+        _ leftContinuation: UnsafeContinuation<Demand, Error>
+    ) async throws -> Reducer<Self, Action>.Effect {
+        switch leftResult {
+            case let .value((value)):
+                left = (value, leftContinuation)
+                guard !Task.isCancelled else {
+                    leftContinuation.resume(returning: .done)
+                    return .completion(.cancel)
+                }
+                mostRecentDemand = try await downstream(.value((value, right?.value)))
+                leftContinuation.resume(returning: mostRecentDemand)
+                return .none
+            case .completion(_):
+                leftComplete = true
+                return leftComplete && rightComplete ? .completion(.exit) : .none
+        }
+    }
+
+    private mutating func handleRight(
+        _ rightResult: AsyncStream<Right>.Result,
+        _ rightContinuation: UnsafeContinuation<Demand, Error>
+    ) async throws -> Reducer<Self, Self.Action>.Effect {
+        switch rightResult {
+            case let .value((value)):
+                right = (value, rightContinuation)
+                guard !Task.isCancelled else {
+                    rightContinuation.resume(returning: .done)
+                    return .completion(.cancel)
+                }
+                mostRecentDemand = try await downstream(.value((left?.value, value)))
+                rightContinuation.resume(returning: mostRecentDemand)
+                return .none
+            case .completion(_) :
+                rightComplete = true
+                return leftComplete && rightComplete ? .completion(.exit) : .none
+        }
+    }
+}

Sources/FreeCombine/Filter/DropFirst.swift

@@ -0,0 +1,25 @@
+//
+//  File.swift
+//  
+//
+//  Created by Van Simmons on 6/4/22.
+//
+public extension Publisher {
+    func dropFirst(
+        _ count: Int = 1
+    ) -> Self {
+        .init { continuation, downstream in
+            let currentValue: ValueRef<Int> = ValueRef(value: count + 1)
+            return self(onStartup: continuation) { r in
+                let current = await currentValue.value - 1
+                await currentValue.set(value: max(0, current))
+                switch r {
+                case .value:
+                    guard current <= 0 else { return .more }
+                    return try await downstream(r)
+                case let .completion(value):
+                    return try await downstream(.completion(value))
+            } }
+        }
+    }
+}

Sources/FreeCombine/Publishers/Combinator.swift

@@ -12,13 +12,11 @@ public protocol CombinatorState {
 public func Combinator<Output: Sendable, State: CombinatorState, Action>(
     initialState: @escaping (@escaping (AsyncStream<Output>.Result) async throws -> Demand) -> (Channel<Action>) async -> State,
     buffering: AsyncStream<Action>.Continuation.BufferingPolicy,
-    onCancel: @escaping () -> Void,
     reducer: Reducer<State, Action>
 ) -> Publisher<Output> where State.CombinatorAction == Action {
     .init(
         initialState: initialState,
         buffering: buffering,
-        onCancel: onCancel,
         reducer: reducer
     )
 }
@@ -27,7 +25,6 @@ public extension Publisher {
     init<State: CombinatorState, Action>(
         initialState: @escaping (@escaping (AsyncStream<Output>.Result) async throws -> Demand) -> (Channel<Action>) async -> State,
         buffering: AsyncStream<Action>.Continuation.BufferingPolicy,
-        onCancel: @escaping () -> Void,
         reducer: Reducer<State, Action>
     ) where State.CombinatorAction == Action {
         self = .init { continuation, downstream in
@@ -37,10 +34,7 @@ public extension Publisher {
                     reducer: reducer
                 )
 
-                return try await withTaskCancellationHandler(handler: {
-                    stateTask.cancel()
-                    onCancel()
-                }) {
+                return try await withTaskCancellationHandler(handler: stateTask.cancel) {
                     continuation?.resume()
                     guard !Task.isCancelled else {
                         throw PublisherError.cancelled