We need to trigger a side effect on a publisher when the first subscriber subscribes and the last subscriber unsubscribes.
We're looking for the Combine equivalent of "doOnSubscribe" and "doOnUnsubscribe" familiar to ReactiveX users. We've used ".handleEvents", but we've been informed that it should only be used for debugging as it may not be thread safe.
Here's a solution for "doOnSubscribe" that utilizes "prepend". But to implement "doOnUnsubscribe", you cannot replace "prepend" with "append" as the upstream will never be cancelled without the side-effect - when the last subscriber unsubscribes, we need to trigger the side effect that ultimately tells the upstream publisher to complete.
extension Publishers {
public struct OnSubscribePublisher<P: Publisher>: Publisher where P.Failure == Never {
public typealias Output = P.Output
public typealias Failure = P.Failure
private let _transform: AnyPublisher<Output,Failure>
fileprivate init(
publisher: P,
work: @escaping () -> Void
)
{
self._transform = Deferred{ publisher }
.map { Optional($0) }
.prepend(
Deferred {
Just(
nil
)
.map { (_: Output?) in
work()
return nil
}
}
)
.compactMap { $0 }
.eraseToAnyPublisher()
}
public func receive<S>(
subscriber: S
) where
S: Subscriber,
S.Failure == Failure,
S.Input == Output
{
_transform
.receive(
subscriber: subscriber
)
}
}
}