start in on Connectable support

Documentation/Bibliography.md

@@ -34,6 +34,8 @@ https://github.com/bitemyapp/papers/blob/master/Stream%20Fusion%20on%20Haskell%2
 
 ### Stream Libraries
 
+[Akka Streams](https://qconnewyork.com/ny2015/system/files/presentation-slides/AkkaStreamsQconNY.pdf)
+
 ### Combine
 
 [Conversation on Combine](https://iosdevelopers.slack.com/archives/C0AET0JQ5/p1623102144192300)

README.md

@@ -2,20 +2,16 @@
 
 # FreeCombine
 
-## Combine. Only free. And concurrent.
-
-- Protocol-free.
-- Race-free.
-
-  - Yield-free.
-  - Sleep-free.
-
-- Leak-free.
-
-  - Unbounded queue-free.
-
-- Lock-free.
-- Dependency-free.
+## Like Combine. Only free. And concurrent.
+
+* Protocol-free.
+* Race-free.
+  * Yield-free.
+  * Sleep-free.
+* Leak-free.
+  * Unbounded queue-free.
+* Lock-free.
+* Dependency-free.
 
 ## Salient features
 

Sources/FreeCombine/Cancellable/Cancellable.swift

@@ -5,6 +5,8 @@
 //  Created by Van Simmons on 5/18/22.
 //
 
+// Can't be a protocol bc we have to implement deinit
+
 public final class Cancellable<Output: Sendable>: Sendable {
     private let _cancel: @Sendable () -> Void
     private let _isCancelled: @Sendable () -> Bool
@@ -63,3 +65,43 @@ public extension Cancellable {
         self.init(task: Task.init(operation: task))
     }
 }
+public extension Cancellable {
+    static func join<B>(_ outer: Cancellable<Cancellable<B>>) -> Cancellable<B> {
+        .init(
+            cancel: { outer.cancel() },
+            isCancelled: { outer.isCancelled },
+            value: {
+                let inner = try await outer.value
+                return try await inner.value
+            },
+            result: {
+                switch await outer.result {
+                    case let .success(value): return await value.result
+                    case let .failure(error): return .failure(error)
+                }
+            }
+        )
+    }
+
+    func map<B>(_ f: @escaping (Output) async -> B) -> Cancellable<B> {
+        .init(
+            cancel: self.cancel,
+            isCancelled: { self.isCancelled },
+            value: { try await f(self.value) },
+            result: {
+                switch await self.result {
+                    case let .success(value): return await .success(f(value))
+                    case let .failure(error): return .failure(error)
+                }
+            }
+        )
+    }
+
+    func join<B>() -> Cancellable<B> where  Output == Cancellable<B> {
+        Self.join(self)
+    }
+
+    func flatMap<B>(_ f: @escaping (Output) async -> Cancellable<B>) -> Cancellable<B> {
+        self.map(f).join()
+    }
+}

Sources/FreeCombine/Combinator/CombineLatestState.swift

@@ -51,7 +51,7 @@ struct CombineLatestState<Left: Sendable, Right: Sendable> {
         switch completion {
             case .cancel:
                 _ = try? await state.downstream(.completion(.cancelled))
-            case .exit, .termination:
+            case .exit, .finished:
                 _ = try? await state.downstream(.completion(.finished))
             case let .failure(error):
                 _ = try? await state.downstream(.completion(.failure(error)))

Sources/FreeCombine/Combinator/MergeState.swift

@@ -60,7 +60,7 @@ struct MergeState<Output: Sendable> {
         guard state.mostRecentDemand != .done else { return }
         do {
             switch completion {
-                case .termination:
+                case .finished:
                     state.mostRecentDemand = try await state.downstream(.completion(.finished))
                 case .cancel:
                     state.mostRecentDemand = try await state.downstream(.completion(.cancelled))

Sources/FreeCombine/Combinator/ZipState.swift

@@ -47,7 +47,7 @@ struct ZipState<Left: Sendable, Right: Sendable> {
         switch completion {
             case .cancel:
                 _ = try? await state.downstream(.completion(.cancelled))
-            case .exit, .termination:
+            case .exit, .finished:
                 _ = try? await state.downstream(.completion(.finished))
             case let .failure(error):
                 _ = try? await state.downstream(.completion(.failure(error)))

Sources/FreeCombine/Decombinator/DistributorState.swift

@@ -30,7 +30,7 @@ public struct DistributorState<Output: Sendable> {
 
     static func complete(state: inout Self, completion: Reducer<Self, Self.Action>.Completion) async -> Void {
         switch completion {
-            case .termination:
+            case .finished:
                 await state.process(currentRepeaters: state.repeaters, with: .completion(.finished))
             case .exit:
                 fatalError("Distributor should never exit")

Sources/FreeCombine/Decombinator/MulticasterState.swift

@@ -53,7 +53,7 @@ public struct MulticasterState<Output: Sendable> {
 
     static func complete(state: inout Self, completion: Reducer<Self, Self.Action>.Completion) async -> Void {
         switch completion {
-            case .termination:
+            case .finished:
                 await state.distributor.process(currentRepeaters: state.distributor.repeaters, with: .completion(.finished))
             case .exit:
                 fatalError("Multicaster should never exit")
@@ -156,8 +156,8 @@ public struct MulticasterState<Output: Sendable> {
                 return .none // FIXME: Need to handle this
             case let .completion(completion):
                 switch completion {
-                    case .termination:
-                        return .completion(.termination)
+                    case .finished:
+                        return .completion(.finished)
                     case .exit:
                         return .completion(.exit)
                     case let .failure(error):

Sources/FreeCombine/Decombinator/RepeaterState.swift

@@ -31,7 +31,7 @@ public struct RepeaterState<ID: Hashable & Sendable, Output: Sendable>: Identifi
     static func complete(state: inout Self, completion: Reducer<Self, Self.Action>.Completion) async -> Void {
         do {
             switch completion {
-                case .termination:
+                case .finished:
                     _ = try await state.downstream(.completion(.finished))
                 case .exit:
                     ()

Sources/FreeCombine/Multicast/Share.swift

@@ -4,35 +4,51 @@
 //
 //  Created by Van Simmons on 6/7/22.
 //
+
+// Need join on Cancellable
 public extension Publisher {
-//    func share() async -> Self {
-//        let multicaster: ValueRef<Multicaster<Output>?> = ValueRef(value: .none)
-//        return .init { continuation, downstream in
-//            return self(onStartup: continuation) { r in
-//                if await multicaster.value == nil {
-//                    let m: Multicaster<Output> = await .init(
-//                        stateTask: Channel.init().stateTask(
-//                            initialState: MulticasterState<Output>.create(upstream: self),
-//                            reducer: Reducer(
-//                                onCompletion: MulticasterState<Output>.complete,
-//                                reducer: MulticasterState<Output>.reduce
-//                            )
-//                        )
-//                    )
-//                    await multicaster.set(value: m)
-//                    try await m.connect()
-//                }
-//                guard let m = await multicaster.value else {
-//                    fatalError("Lost connection")
-//                }
-//                switch r {
-//                    case .value(let a):
-//                        return try await downstream(.value(a))
-//                    case let .completion(value):
-//                        try await m.disconnect()
-//                        return .done
-//                }
-//            }
-//        }
-//    }
+
+    func share() async -> Self {
+        let multicaster: ValueRef<Multicaster<Output>?> = ValueRef(value: .none)
+        @Sendable func lift(
+            _ downstream: @Sendable @escaping (AsyncStream<Output>.Result) async throws -> Demand
+        ) -> @Sendable (AsyncStream<Output>.Result) async throws -> Demand {
+            { r in
+                switch r {
+                    case .value:
+                        return try await downstream(r)
+                    case .completion:
+                        let finalValue = try await downstream(r)
+                        await multicaster.set(value: .none)
+                        return finalValue
+                }
+            }
+        }
+        return .init { continuation, downstream in
+            Cancellable<Cancellable<Demand>>.join(.init {
+                if await multicaster.value == nil {
+                    let m: Multicaster<Output> = await .init(
+                        stateTask: Channel.init().stateTask(
+                            initialState: MulticasterState<Output>.create(upstream: self),
+                            reducer: Reducer(
+                                onCompletion: MulticasterState<Output>.complete,
+                                reducer: MulticasterState<Output>.reduce
+                            )
+                        )
+                    )
+                    await multicaster.set(value: m)
+                    let cancellable = await m.publisher().sink(lift(downstream))
+                    try await m.connect()
+                    continuation?.resume()
+                    return cancellable
+                }
+                guard let m = await multicaster.value else {
+                    return await Empty(Output.self).sink(downstream)
+                }
+                continuation?.resume()
+                let cancellable = await m.publisher().sink(lift(downstream))
+                return cancellable
+            } )
+        }
+    }
 }

Sources/FreeCombine/State/Reducer.swift

@@ -17,7 +17,7 @@ public struct Reducer<State, Action> {
     }
 
     public enum Completion {
-        case termination
+        case finished
         case exit
         case failure(Swift.Error)
         case cancel

Sources/FreeCombine/State/StateTask.swift

@@ -81,11 +81,11 @@ extension StateTask {
                             case .published(_): continue // FIXME: Handle future mutation
                             case .completion(.exit): throw PublisherError.completed
                             case let .completion(.failure(error)): throw error
-                            case .completion(.termination): throw PublisherError.internalError
+                            case .completion(.finished): throw PublisherError.internalError
                             case .completion(.cancel): throw PublisherError.cancelled
                         }
                     }
-                    await reducer(&state, .termination)
+                    await reducer(&state, .finished)
                 } catch {
                     channel.finish()
                     for await action in channel { reducer(action, .failure(error)); continue }

Tests/FreeCombineTests/CancellableTests.swift

@@ -0,0 +1,54 @@
+//
+//  CancellableTests.swift
+//  
+//
+//  Created by Van Simmons on 6/8/22.
+//
+
+import XCTest
+@testable import FreeCombine
+
+final class CancellableTests: XCTestCase {
+
+    override func setUpWithError() throws { }
+
+    override func tearDownWithError() throws { }
+
+    func testCancellable() async throws {
+        let expectation1 = await CheckedExpectation<Void>()
+        let expectation2 = await CheckedExpectation<Void>()
+        let expectation3 = await CheckedExpectation<Void>()
+
+        let expectation1a = await CheckedExpectation<Bool>()
+        let expectation2a = await CheckedExpectation<Bool>()
+        let expectation3a = await CheckedExpectation<Bool>()
+
+        _ = await Cancellable(task: Task<(Cancellable<Void>, Cancellable<Void>, Cancellable<Void>), Swift.Error> {
+            let t1 = Cancellable(task: Task<Void, Swift.Error> {
+                try await expectation1.value
+                try await expectation1a.complete(Task.isCancelled)
+            })
+            let t2 = Cancellable(task: Task<Void, Swift.Error> {
+                try await expectation2.value
+                try await expectation2a.complete(Task.isCancelled)
+            })
+            let t3 = Cancellable(task: Task<Void, Swift.Error> {
+                try await expectation3.value
+                try await expectation3a.complete(Task.isCancelled)
+            })
+            return (t1, t2, t3)
+        }).result
+
+        try await expectation1.complete()
+        try await expectation2.complete()
+        try await expectation3.complete()
+
+        let r1 = try await expectation1a.value
+        let r2 = try await expectation2a.value
+        let r3 = try await expectation3a.value
+
+        XCTAssert(r1, "Inner task 1 not cancelled")
+        XCTAssert(r2, "Inner task 2 not cancelled")
+        XCTAssert(r3, "Inner task 3 not cancelled")
+    }
+}

Tests/FreeCombineTests/CancellationTests.swift

@@ -99,7 +99,7 @@ class CancellationTests: XCTestCase {
                     let count2 = await counter2.count
                     XCTAssertTrue(count2 == 26, "Incorrect count: \(count2)")
                     do { try await expectation2.complete() }
-                    catch { XCTFail("Multiple terminations sent: \(error)") }
+                    catch { XCTFail("Multiple finishes sent: \(error)") }
                     return .done
                 case .completion(.cancelled):
                     XCTFail("Should not have cancelled")
@@ -128,7 +128,7 @@ class CancellationTests: XCTestCase {
                     case .completion(.finished):
                         XCTFail("Got to end of task that should have been cancelled")
                         do { try await expectation.complete() }
-                        catch { XCTFail("Multiple terminations sent: \(error)") }
+                        catch { XCTFail("Multiple finishes sent: \(error)") }
                         return .done
                     case .completion(.cancelled):
                         try await expectation.complete()

Tests/FreeCombineTests/CombineLatestTests.swift

@@ -184,7 +184,7 @@ class CombineLatestTests: XCTestCase {
                         let count = await counter.count
                         XCTAssert(count == 254, "wrong number of values sent: \(count)")
                         do { try await expectation.complete() }
-                        catch { XCTFail("Multiple terminations sent: \(error)") }
+                        catch { XCTFail("Multiple finishes sent: \(error)") }
                         return .done
                     case .completion(.cancelled):
                         XCTFail("Should not have cancelled")

Tests/FreeCombineTests/MulticastTests.swift

@@ -103,14 +103,12 @@ class MulticastTests: XCTestCase {
                     XCTFail("Got an error? \(error)")
                     return .done
                 case .completion(.finished):
-                    print("received: \(result)")
                     let count = await counter1.count
                     if count != 100 {
                         XCTFail("Incorrect count: \(count) in subscription 1")
                     }
                     return .done
                 case .completion(.cancelled):
-                    print("received: \(result)")
                     XCTFail("Should not have cancelled")
                     return .done
             }