How can I trigger a process after a returned publisher would be subscribed?

user_11837681 0
Viewed 446

I have a function that returns a publisher. This publisher gives the results of a background process. I only want to trigger the background process when the publisher would be subscribed, so that no results are lost. The background process can update its results many times, so the variant with Future is not suitable.

private let passthroughSubject = PassthroughSubject<Data, Error>()

// This function will be used outside.
func fetchResults() -> AnyPublisher<Data, Error> {
     return passthroughSubject
     .eraseToAnyPublisher()
     .somehowTriggerTheBackgroundProcess()
}

extension MyModule: MyDelegate {
     func didUpdateResult(newResult: Data) {
          self.passthroughSubject.send(newResult)
     }
}

What have I tried?

Future:

Future<Data, Error> { [weak self] promise in
     self?.passthroughSubject
          .sink(receiveCompletion: { completion in
               // My logic
          }, receiveValue: { value in
               // My logic    
          })
          .store(in: &self.cancellableSet)
      self?.triggerBackgroundProcess()
}.eraseToAnyPublisher()

Works the way I want but the subscriber is called only once (logical).

Deffered:

Deferred<AnyPublisher<Data, Error>>(createPublisher: { [weak self] in
   defer {
      self?.triggerBackgroundProcess()
   }
   return passthroughSubject.eraseToAnyPublisher()
}

Debugger shows that everything is correct: first return then trigger but the subscriber is not called for the first time.

receiveSubscription:

passthroughSubject
.handleEvents(receiveSubscription: { [weak self] subscription in 
   self?.triggerBackgroundProcess()
})
.eraseToAnyPublisher()

The same effect as with Deffered.

Is it even possible what I want to achieve? Or, it is better to create a public publisher subscribe it and receive results from background process. And the fetchResults() function doesn't return anything?

Thanks in advance for your help.

combine swift
2 Answers

It seems to me that your last bit of code is a perfectly viable solution: don't trigger the background process until you detect the subscription. Example:

let subject = PassthroughSubject<String, Never>()
var storage = Set<AnyCancellable>()
func start() {
    self.subject
        .handleEvents(receiveSubscription: {_ in
            print("subscribed")
            DispatchQueue.main.async {
                self.doSomethingAsynchronous()
            }
        })
        .sink { print("got", $0) }
        .store(in: &storage)
}
func doSomethingAsynchronous() {
    DispatchQueue.global(qos: .userInitiated).async {
        DispatchQueue.main.async {
            self.subject.send("bingo")
        }
    }
}
user_3419940

You can write your own type that conforms to Publisher and wraps a PassthroughSubject. In your implementation, you can start the background process when you get a subscription.

public struct MyPublisher: Publisher {
    public typealias Output = Data
    public typealias Failure = Error

    public func receive<Downstream: Subscriber>(subscriber: Downstream)
        where Downstream.Input == Output, Downstream.Failure == Failure
    {
        let subject = PassthroughSubject<Output, Failure>()
        subject.subscribe(subscriber)

        startBackgroundProcess(subject: subject)
    }

    private func startBackgroundProcess(subject: PassthroughSubject<Output, Failure>) {
        DispatchQueue.global(qos: .utility).async {
            print("background process running")
            subject.send(Data())
            subject.send(completion: .finished)
        }
    }
}

Note that this publisher starts a new background process for each subscriber. That is a common implementation. For example URLSession.DataTaskPublisher issues a new request for each subscriber. If you want multiple subscribers to share the output of a single request, you can use the .multicast operator, add multiple subscribers, and then .connect() the multicast publisher to start the background process once:

let pub = MyPublisher().multicast { PassthroughSubject() }
pub.sink(...).store(in: &tickets) // first subscriber
pub.sink(...).store(in: &tickets) // second subscriber
pub.connect().store(in: &tickets) // start the background process
user_775670
Related