Swift Combine：检查主题是否有观察者？

Question

在 RxSwift 中，我们可以检查 *Subject 是否有任何观察者，使用 hasObserver，我如何在组合中执行此操作，例如PassthroughSubject?

Answer 1

在发布我的问题后的一段时间，我写了这个简单的扩展。比@Asperi 的解决方案简单得多。除了简单（我的）之外，不确定这两种解决方案之间的缺点/优点。

private enum CounterChange: Int, Equatable {
    case increased = 1
    case decreased = -1
}

extension Publisher {
    func trackNumberOfSubscribers(
        _ notifyChange: @escaping (Int) -> Void
    ) -> AnyPublisher<Output, Failure> {

        var counter = NSNumber.init(value: 0)
        let nsLock = NSLock()

        func updateCounter(_ change: CounterChange, notify: (Int) -> Void) {
            nsLock.lock()
            counter = NSNumber(value: counter.intValue + change.rawValue)
            notify(counter.intValue)
            nsLock.unlock()
        }

        return handleEvents(
            receiveSubscription: { _ in updateCounter(.increased, notify: notifyChange) },
            receiveCompletion: { _ in updateCounter(.decreased, notify: notifyChange) },
            receiveCancel: { updateCounter(.decreased, notify: notifyChange) }
        ).eraseToAnyPublisher()
    }
}

这里有一些测试：

import XCTest
import Combine

final class PublisherTrackNumberOfSubscribersTest: TestCase {

    func test_four_subscribers_complete_by_finish() {
        doTest { publisher in
            publisher.send(completion: .finished)
        }
    }

    func test_four_subscribers_complete_by_error() {
        doTest { publisher in
            publisher.send(completion: .failure(.init()))
        }
    }

}

private extension PublisherTrackNumberOfSubscribersTest {
    struct EmptyError: Swift.Error {}
    func doTest(_ line: UInt = #line, complete: (PassthroughSubject<Int, EmptyError>) -> Void) {
        let publisher = PassthroughSubject<Int, EmptyError>()

        var numberOfSubscriptions = [Int]()
        let trackable = publisher.trackNumberOfSubscribers { counter in
            numberOfSubscriptions.append(counter)
        }

        func subscribe() -> Cancellable {
            return trackable.sink(receiveCompletion: { _ in }, receiveValue: { _ in })
        }

        let cancellable1 = subscribe()
        let cancellable2 = subscribe()
        let cancellable3 = subscribe()
        let cancellable4 = subscribe()

        XCTAssertNotNil(cancellable1, line: line)
        XCTAssertNotNil(cancellable2, line: line)
        XCTAssertNotNil(cancellable3, line: line)
        XCTAssertNotNil(cancellable4, line: line)

        cancellable1.cancel()
        cancellable2.cancel()

        complete(publisher)
        XCTAssertEqual(numberOfSubscriptions, [1, 2, 3, 4, 3, 2, 1, 0], line: line)
    }
}

Answer 2

没有人需要这个...Apple 没有通过 API 提供这个，实际上，我不推荐这样的东西，因为它就像在 ARC 之前的 Objective-C 中手动检查 retainCount 的值。代码中的决定。

无论如何，这是可能的。让我们将其视为实验室练习。希望有人觉得这有帮助。

免责声明：以下代码未经过所有发布者的测试，对于某些实际项目并不安全。这只是方法演示。

所以，由于发布者有很多种，而且它们都是最终的和私有的，而且可能通过类型擦除器来，我们需要通用的东西适用于任何发布者，因此需要操作符

extension Publisher {
    public func countingSubscribers(_ callback: ((Int) -> Void)? = nil)
        -> Publishers.SubscribersCounter<Self> {
            return Publishers.SubscribersCounter<Self>(upstream: self, callback: callback)
    }
}

Operator 让我们可以在发布者链的任何地方注入并通过回调提供有趣的值。在我们的例子中，有趣的价值是订阅者的数量。

由于在上游和下游都注入了操作符，我们需要双向自定义管道实现，即。自定义发布者、自定义订阅者、自定义订阅。在我们的例子中，它们必须是透明的，因为我们不需要修改流……实际上它将是组合代理。

可能的用法：
1）SubscribersCounter发布者在链中最后，numberOfSubscribers属性可以直接使用

let publisher = NotificationCenter.default
    .publisher(for: UIApplication.didBecomeActiveNotification)
    .countingSubscribers()
...
publisher.numberOfSubscribers

2) 当它在链的中间某处，然后接收关于改变订阅者数量的回调

let publisher = URLSession.shared
        .dataTaskPublisher(for: URL(string: "https://www.google.com")!)
        .countingSubscribers({ count in print("Observers: \(count)") })
        .receive(on: DispatchQueue.main)
        .map { _ in "Data received" }
        .replaceError(with: "An error occurred")

这里是实现：

import Combine

extension Publishers {

    public class SubscribersCounter<Upstream> : Publisher where Upstream : Publisher {

        private(set) var numberOfSubscribers = 0

        public typealias Output = Upstream.Output
        public typealias Failure = Upstream.Failure

        public let upstream: Upstream
        public let callback: ((Int) -> Void)?

        public init(upstream: Upstream, callback: ((Int) -> Void)?) {
            self.upstream = upstream
            self.callback = callback
        }

        public func receive<S>(subscriber: S) where S : Subscriber,
            Upstream.Failure == S.Failure, Upstream.Output == S.Input {
                self.increase()
                upstream.receive(subscriber: SubscribersCounterSubscriber<S>(counter: self, subscriber: subscriber))
        }

        fileprivate func increase() {
            numberOfSubscribers += 1
            self.callback?(numberOfSubscribers)
        }

        fileprivate func decrease() {
            numberOfSubscribers -= 1
            self.callback?(numberOfSubscribers)
        }

        // own subscriber is needed to intercept upstream/downstream events
        private class SubscribersCounterSubscriber<S> : Subscriber where S: Subscriber {
            let counter: SubscribersCounter<Upstream>
            let subscriber: S

            init (counter: SubscribersCounter<Upstream>, subscriber: S) {
                self.counter = counter
                self.subscriber = subscriber
            }

            deinit {
                Swift.print(">> Subscriber deinit")
            }

            func receive(subscription: Subscription) {
                subscriber.receive(subscription: SubscribersCounterSubscription<Upstream>(counter: counter, subscription: subscription))
            }

            func receive(_ input: S.Input) -> Subscribers.Demand {
                return subscriber.receive(input)
            }

            func receive(completion: Subscribers.Completion<S.Failure>) {
                subscriber.receive(completion: completion)
            }

            typealias Input = S.Input
            typealias Failure = S.Failure
        }

        // own subcription is needed to handle cancel and decrease
        private class SubscribersCounterSubscription<Upstream>: Subscription where Upstream: Publisher {
            let counter: SubscribersCounter<Upstream>
            let wrapped: Subscription

            private var cancelled = false
            init(counter: SubscribersCounter<Upstream>, subscription: Subscription) {
                self.counter = counter
                self.wrapped = subscription
            }

            deinit {
                Swift.print(">> Subscription deinit")
                if !cancelled {
                    counter.decrease()
                }
            }

            func request(_ demand: Subscribers.Demand) {
                wrapped.request(demand)
            }

            func cancel() {
                wrapped.cancel()
                if !cancelled {
                    cancelled = true
                    counter.decrease()
                }
            }
        }
    }
}