Add withUnretained operator

CombineExt.xcodeproj/project.pbxproj

@@ -49,6 +49,8 @@
 		EA0D86D5287D19DC0085356E /* MapToResultTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = EA0D86D2287D19DC0085356E /* MapToResultTests.swift */; };
 		EA0D86D6287D19DC0085356E /* MapToValueTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = EA0D86D3287D19DC0085356E /* MapToValueTests.swift */; };
 		EAEAAC72287FB3C900288379 /* CombineSchedulers in Frameworks */ = {isa = PBXBuildFile; productRef = EAEAAC71287FB3C900288379 /* CombineSchedulers */; };
+		E17B23B526DFBFBD008E595F /* WithUnretained.swift in Sources */ = {isa = PBXBuildFile; fileRef = E17B23B426DFBFBD008E595F /* WithUnretained.swift */; };
+		E17B23B726DFFA56008E595F /* WithUnretainedTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = E17B23B626DFFA56008E595F /* WithUnretainedTests.swift */; };
 		OBJ_100 /* ZipMany.swift in Sources */ = {isa = PBXBuildFile; fileRef = OBJ_33 /* ZipMany.swift */; };
 		OBJ_101 /* CurrentValueRelay.swift in Sources */ = {isa = PBXBuildFile; fileRef = OBJ_35 /* CurrentValueRelay.swift */; };
 		OBJ_102 /* PassthroughRelay.swift in Sources */ = {isa = PBXBuildFile; fileRef = OBJ_36 /* PassthroughRelay.swift */; };
@@ -146,6 +148,8 @@
 		EA0D86D1287D19DC0085356E /* EnumeratedTests.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = EnumeratedTests.swift; sourceTree = "<group>"; };
 		EA0D86D2287D19DC0085356E /* MapToResultTests.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = MapToResultTests.swift; sourceTree = "<group>"; };
 		EA0D86D3287D19DC0085356E /* MapToValueTests.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = MapToValueTests.swift; sourceTree = "<group>"; };
+		E17B23B426DFBFBD008E595F /* WithUnretained.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = WithUnretained.swift; sourceTree = "<group>"; };
+		E17B23B626DFFA56008E595F /* WithUnretainedTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = WithUnretainedTests.swift; sourceTree = "<group>"; };
 		OBJ_10 /* Sink.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Sink.swift; sourceTree = "<group>"; };
 		OBJ_12 /* Optional.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Optional.swift; sourceTree = "<group>"; };
 		OBJ_14 /* Event.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Event.swift; sourceTree = "<group>"; };
@@ -289,6 +293,7 @@
 				1970A8A925246FBD00799AB6 /* FilterMany.swift */,
 				BFADDC8025BCE4C200465E9B /* FlatMapBatches.swift */,
 				63FEBC9227E9FCDB00E934AD /* FlatMapFirst.swift */,
+				E17B23B426DFBFBD008E595F /* WithUnretained.swift */,
 			);
 			path = Operators;
 			sourceTree = "<group>";
@@ -347,6 +352,7 @@
 				OBJ_61 /* ZipManyTests.swift */,
 				BFADDC8A25BCE91E00465E9B /* FlatMapBatchesTests.swift */,
 				63FEBC9427E9FE9000E934AD /* FlatMapFirstTests.swift */,
+				E17B23B626DFFA56008E595F /* WithUnretainedTests.swift */,
 			);
 			path = Tests;
 			sourceTree = SOURCE_ROOT;
@@ -590,6 +596,7 @@
 				OBJ_126 /* CreateTests.swift in Sources */,
 				OBJ_127 /* CurrentValueRelayTests.swift in Sources */,
 				C387777F24E6BF8F00FAD2D8 /* NwiseTests.swift in Sources */,
+				E17B23B726DFFA56008E595F /* WithUnretainedTests.swift in Sources */,
 				OBJ_128 /* DematerializeTests.swift in Sources */,
 				OBJ_129 /* FlatMapLatestTests.swift in Sources */,
 				OBJ_130 /* MapManyTests.swift in Sources */,
@@ -626,6 +633,7 @@
 				EA0D86CE287D19CC0085356E /* MapToValue.swift in Sources */,
 				OBJ_83 /* ObjectOwnership.swift in Sources */,
 				OBJ_84 /* Amb.swift in Sources */,
+				E17B23B526DFBFBD008E595F /* WithUnretained.swift in Sources */,
 				OBJ_85 /* AssignOwnership.swift in Sources */,
 				OBJ_86 /* AssignToMany.swift in Sources */,
 				BF3D3B5D253B83F300D830ED /* IgnoreFailure.swift in Sources */,

Package.resolved

@@ -6,17 +6,26 @@
         "repositoryURL": "https://github.com/pointfreeco/combine-schedulers",
         "state": {
           "branch": null,
-          "revision": "4cf088c29a20f52be0f2ca54992b492c54e0076b",
-          "version": "0.5.3"
+          "revision": "ec62f32d21584214a4b27c8cee2b2ad70ab2c38a",
+          "version": "0.11.0"
+        }
+      },
+      {
+        "package": "swift-concurrency-extras",
+        "repositoryURL": "https://github.com/pointfreeco/swift-concurrency-extras",
+        "state": {
+          "branch": null,
+          "revision": "479750bd98fac2e813fffcf2af0728b5b0085795",
+          "version": "0.1.1"
         }
       },
       {
         "package": "xctest-dynamic-overlay",
         "repositoryURL": "https://github.com/pointfreeco/xctest-dynamic-overlay",
         "state": {
           "branch": null,
-          "revision": "50a70a9d3583fe228ce672e8923010c8df2deddd",
-          "version": "0.2.1"
+          "revision": "50843cbb8551db836adec2290bb4bc6bac5c1865",
+          "version": "0.9.0"
         }
       }
     ]

Sources/Operators/WithUnretained.swift

@@ -0,0 +1,112 @@
+//
+//  WithUnretained.swift
+//  CombineExt
+//
+//  Created by Robert on 01/09/2021.
+//  Copyright © 2020 Combine Community. All rights reserved.
+//
+
+#if canImport(Combine)
+import Combine
+
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+public extension Publisher {
+    /**
+     Provides an unretained, safe to use (i.e. not implicitly unwrapped), reference to an object along with the events published by the publisher.
+
+     In the case the provided object cannot be retained successfully, the publisher will complete.
+
+     - parameter obj: The object to provide an unretained reference on.
+     - parameter resultSelector: A function to combine the unretained referenced on `obj` and the value of the observable sequence.
+     - returns: A publisher that contains the result of `resultSelector` being called with an unretained reference on `obj` and the values of the upstream.
+     */
+    func withUnretained<UnretainedObject: AnyObject, Output>(_ obj: UnretainedObject, resultSelector: @escaping (UnretainedObject, Self.Output) -> Output) -> Publishers.WithUnretained<UnretainedObject, Self, Output> {
+        Publishers.WithUnretained(unretainedObject: obj, upstream: self, resultSelector: resultSelector)
+    }
+
+    /**
+     Provides an unretained, safe to use (i.e. not implicitly unwrapped), reference to an object along with the events published by the publisher.
+
+     In the case the provided object cannot be retained successfully, the publisher will complete.
+
+     - parameter obj: The object to provide an unretained reference on.
+     - returns: A publisher that publishes a sequence of tuples that contains both an unretained reference on `obj` and the values of the upstream.
+     */
+    func withUnretained<UnretainedObject: AnyObject>(_ obj: UnretainedObject) -> Publishers.WithUnretained<UnretainedObject, Self, (UnretainedObject, Output)> {
+        Publishers.WithUnretained(unretainedObject: obj, upstream: self) { ($0, $1) }
+    }
+
+    /// Attaches a subscriber with closure-based behavior.
+    ///
+    /// Use ``Publisher/sink(unretainedObject:receiveCompletion:receiveValue:)`` to observe values received by the publisher and process them using a closure you specify.
+    /// This method creates the subscriber and immediately requests an unlimited number of values, prior to returning the subscriber.
+    /// The return value should be held, otherwise the stream will be canceled.
+    ///
+    /// - parameter obj: The object to provide an unretained reference on.
+    /// - parameter receiveComplete: The closure to execute on completion.
+    /// - parameter receiveValue: The closure to execute on receipt of a value.
+    /// - Returns: A cancellable instance, which you use when you end assignment of the received value. Deallocation of the result will tear down the subscription stream.
+    func sink<UnretainedObject: AnyObject>(unretainedObject obj: UnretainedObject, receiveCompletion: @escaping ((Subscribers.Completion<Self.Failure>) -> Void), receiveValue: @escaping ((UnretainedObject, Self.Output) -> Void)) -> AnyCancellable {
+        withUnretained(obj)
+            .sink(receiveCompletion: receiveCompletion, receiveValue: receiveValue)
+    }
+}
+
+// MARK: - Publisher
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+public extension Publishers {
+    struct WithUnretained<UnretainedObject: AnyObject, Upstream: Publisher, Output>: Publisher {
+        public typealias Failure = Upstream.Failure
+
+        private weak var unretainedObject: UnretainedObject?
+        private let upstream: Upstream
+        private let resultSelector: (UnretainedObject, Upstream.Output) -> Output
+
+        public init(unretainedObject: UnretainedObject, upstream: Upstream, resultSelector: @escaping (UnretainedObject, Upstream.Output) -> Output) {
+            self.unretainedObject = unretainedObject
+            self.upstream = upstream
+            self.resultSelector = resultSelector
+        }
+
+        public func receive<S: Combine.Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input {
+            upstream.subscribe(Subscriber(unretainedObject: unretainedObject, downstream: subscriber, resultSelector: resultSelector))
+        }
+    }
+}
+
+// MARK: - Subscriber
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+private extension Publishers.WithUnretained {
+    class Subscriber<Downstream: Combine.Subscriber>: Combine.Subscriber where Downstream.Input == Output, Downstream.Failure == Failure {
+        typealias Input = Upstream.Output
+        typealias Failure = Downstream.Failure
+
+        private weak var unretainedObject: UnretainedObject?
+        private let downstream: Downstream
+        private let resultSelector: (UnretainedObject, Input) -> Output
+
+        init(unretainedObject: UnretainedObject?, downstream: Downstream, resultSelector: @escaping (UnretainedObject, Input) -> Output) {
+            self.unretainedObject = unretainedObject
+            self.downstream = downstream
+            self.resultSelector = resultSelector
+        }
+
+        func receive(subscription: Subscription) {
+            if unretainedObject == nil { return }
+            downstream.receive(subscription: subscription)
+        }
+
+        func receive(_ input: Input) -> Subscribers.Demand {
+            guard let unretainedObject = unretainedObject else { return .none }
+            return downstream.receive(resultSelector(unretainedObject, input))
+        }
+
+        func receive(completion: Subscribers.Completion<Failure>) {
+            if unretainedObject == nil {
+                return downstream.receive(completion: .finished)
+            }
+            downstream.receive(completion: completion)
+        }
+    }
+}
+#endif

Tests/WithUnretainedTests.swift

@@ -0,0 +1,140 @@
+//
+//  WithUnretainedTests.swift
+//  CombineExtTests
+//
+//  Created by Robert on 02/09/2021.
+//
+
+#if !os(watchOS)
+import XCTest
+import Foundation
+import Combine
+import CombineExt
+
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+final class WithUnretainedTests: XCTestCase {
+    fileprivate var testClass: TestClass!
+    var subscription: AnyCancellable?
+    var values: [String] = []
+
+    enum WithUnretainedTestsError: Swift.Error {
+        case someError
+    }
+
+    override func setUp() {
+        super.setUp()
+
+        testClass = TestClass()
+        values = []
+    }
+
+    override func tearDown() {
+        subscription?.cancel()
+        subscription = nil
+    }
+
+    func testObjectAttached() {
+        let testClassId = testClass.id
+        var completed = false
+
+        let correctValues = [
+            "\(testClassId), 1",
+            "\(testClassId), 2",
+            "\(testClassId), 3",
+            "\(testClassId), 5",
+            "\(testClassId), 8"
+        ]
+
+        let inputArr = [1, 2, 3, 5, 8]
+
+        subscription = Publishers.Sequence<[Int], WithUnretainedTestsError>(sequence: inputArr)
+            .withUnretained(self.testClass)
+            .map { "\($0.id), \($1)" }
+            .sink(receiveCompletion: { _ in completed = true },
+                  receiveValue: { self.values.append($0) })
+
+        XCTAssertEqual(values, correctValues)
+        XCTAssertTrue(completed)
+    }
+
+    func testObjectDeallocatesWithEmptyPublisher() {
+        subscription = Empty<Int, WithUnretainedTestsError>()
+            .withUnretained(self.testClass)
+            .sink(receiveCompletion: { _ in }, receiveValue: { _ in })
+
+        // Confirm the object can be deallocated
+        XCTAssertTrue(testClass != nil)
+        testClass = nil
+        XCTAssertTrue(testClass == nil)
+    }
+
+    func testObjectDeallocates() {
+        let inputArr = [1, 2, 3, 5, 8]
+
+        subscription = Publishers.Sequence<[Int], WithUnretainedTestsError>(sequence: inputArr)
+            .withUnretained(self.testClass)
+            .sink(receiveCompletion: { _ in }, receiveValue: { _ in })
+
+        // Confirm the object can be deallocated
+        XCTAssertTrue(testClass != nil)
+        testClass = nil
+        XCTAssertTrue(testClass == nil)
+    }
+
+    func testObjectDeallocatesSequenceCompletes() {
+        let testClassId = testClass.id
+        var completed = false
+
+        let correctValues = [
+            "\(testClassId), 1",
+            "\(testClassId), 2",
+            "\(testClassId), 3"
+        ]
+
+        let inputArr = [1, 2, 3]
+        subscription = Publishers.Sequence<[Int], WithUnretainedTestsError>(sequence: inputArr)
+            .withUnretained(self.testClass)
+            .handleEvents(receiveOutput: { _, value in
+                // Release the object in the middle of the sequence
+                // to confirm it properly terminates the sequence
+                if value == 3 {
+                    self.testClass = nil
+                }
+            })
+            .map { "\($0.id), \($1)" }
+            .sink(receiveCompletion: { _ in completed = true },
+                  receiveValue: { self.values.append($0) })
+
+        XCTAssertEqual(values, correctValues)
+        XCTAssertTrue(completed)
+    }
+
+    func testResultsSelector() {
+        let testClassId = testClass.id
+        var completed = false
+
+        let inputArr = [(1, "a"), (2, "b"), (3, "c"), (5, "d"), (8, "e")]
+
+        let correctValues = [
+            "\(testClassId), 1, a",
+            "\(testClassId), 2, b",
+            "\(testClassId), 3, c",
+            "\(testClassId), 5, d",
+            "\(testClassId), 8, e"
+        ]
+
+        subscription = Publishers.Sequence<[(Int, String)], WithUnretainedTestsError>(sequence: inputArr)
+                .withUnretained(self.testClass) { ($0, $1.0, $1.1) }
+                .map { "\($0.id), \($1), \($2)" }
+            .sink(receiveCompletion: { _ in completed = true },
+                  receiveValue: { self.values.append($0) })
+
+        XCTAssertEqual(values, correctValues)
+        XCTAssertTrue(completed)
+    }
+}
+
+private class TestClass {
+    let id: String = UUID().uuidString
+}
+#endif