swift - 订阅返回的发布者后如何触发流程？

Question

我有一个返回发布者的函数。该发布者给出了后台进程的结果。我只想在订阅发布者时触发后台进程，这样就不会丢失结果。后台进程可以多次更新其结果，因此带有 Future 的变体不适合。

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)
     }
}

我尝试了什么？

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()

按我想要的方式工作，但订阅者仅被调用一次(逻辑)。

延期:

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

调试器显示一切正确:首先返回，然后触发，但订阅者不是第一次调用。

接收订阅:

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

与Deffered效果相同。

我想要实现的目标是否可能实现？ 或者，最好创建一个公共(public)发布者订阅它并从后台进程接收结果。并且 fetchResults() 函数没有返回任何内容？

预先感谢您的帮助。

Answer 1

您可以编写自己的类型，该类型符合 Publisher 并包装 PassthroughSubject。在您的实现中，您可以在获得订阅时启动后台进程。

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)
        }
    }
}

请注意，该发布者会为每个订阅者启动一个新的后台进程。这是一个常见的实现。例如URLSession.DataTaskPublisher向每个订阅者发出新的请求。如果您希望多个订阅者共享单个请求的输出，可以使用.multicast运算符，添加多个订阅者，然后.connect()多播发布者启动后台进程一次:

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