How to properly manage a collection of `AnyCancellable`

Question

I'd like all publishers to execute unless explicitly cancelled. I don't mind AnyCancellable going out of scope, however based on docs it automatically calls cancel on deinit which is undesired.

I've tried to use a cancellable bag, but AnyCancelable kept piling up even after the publisher fired a completion.

Should I manage the bag manually? I had impression that store(in: inout Set) was meant to be used for convenience of managing the cancellable instances, however all it does is push AnyCancellable into a set.

var cancelableSet = Set<AnyCancellable>()

func work(value: Int) -> AnyCancellable {
    return Just(value)
        .delay(for: .seconds(1), scheduler: DispatchQueue.global(qos: .default))
        .map { $0 + 1 }
        .sink(receiveValue: { (value) in
            print("Got value: \(value)")
        })
}

work(value: 1337).store(in: &cancelableSet)

DispatchQueue.main.asyncAfter(deadline: .now() + .seconds(5)) {
    print("\(cancelableSet)")
}

What I came up with so far, which works fine but makes me wonder if something is missing in the Combine framework or it was not meant to be used in such fashion:

class DisposeBag {
    private let lock = NSLock()
    private var cancellableSet = Set<AnyCancellable>()

    func store(_ cancellable: AnyCancellable) {
        print("Store cancellable: \(cancellable)")

        lock.lock()
        cancellableSet.insert(cancellable)
        lock.unlock()
    }

    func drop(_ cancellable: AnyCancellable) {
        print("Drop cancellable: \(cancellable)")

        lock.lock()
        cancellableSet.remove(cancellable)
        lock.unlock()
    }
}

extension Publisher {

    @discardableResult func autoDisposableSink(disposeBag: DisposeBag, receiveCompletion: @escaping ((Subscribers.Completion<Self.Failure>) -> Void), receiveValue: @escaping ((Self.Output) -> Void)) -> AnyCancellable {
        var sharedCancellable: AnyCancellable?

        let disposeSubscriber = {
            if let sharedCancellable = sharedCancellable {
                disposeBag.drop(sharedCancellable)
            }
        }

        let cancellable = handleEvents(receiveCancel: {
            disposeSubscriber()
        }).sink(receiveCompletion: { (completion) in
            receiveCompletion(completion)

            disposeSubscriber()
        }, receiveValue: receiveValue)

        sharedCancellable = cancellable
        disposeBag.store(cancellable)

        return cancellable
    }

}

Answer 1

The subscriptions in Apple Combine are scoped in a RAII compliant fashion. I.e. the event of deinitialization is equivalent to the event of automatic disposal of the observable. That is contrary to RxSwift Disposable where this behavior is sometimes reproduced, but not strictly so.

Even in RxSwift if you lose a DisposeBag your subscriptions will be disposed and this is a feature. If you would like your subscription to live through the scope, it means that it belongs to an outer scope.

And none of these implementations get busy actually tossing out the Disposables out of the retention tree once the subscriptions are done.