如何使用一次性发布器包装代理模式

Question

通常，我们可以通过使用Future:将异步代码封装在单次发布服务器中来桥接异步代码和Combine

func readEmail() -> AnyPublisher<[String], Error> {
Future { promise in
self.emailManager.readEmail() { result, error in
if let error = error {
promise(.failure(error))
} else {
promise(.success(result))
}
}
}.eraseToAnyPublisher()
}

另一方面，如果我们包装委托模式(而不是异步回调(，建议使用PassthroughSubject，因为这些方法可能会被多次激发：

final class LocationHeadingProxy: NSObject, CLLocationManagerDelegate {
private let headingPublisher: PassthroughSubject<CLHeading, Error> 
override init() {
headingPublisher = PassthroughSubject<CLHeading, Error>()
// ...
}
func locationManager(_ manager: CLLocationManager, didUpdateHeading newHeading: CLHeading) {
headingPublisher.send(newHeading) 
}
}

然而，我正在尝试创建一个一次性发布器，它封装了现有的Delegate模式。原因是我正在启动像connect()这样的方法，我希望成功或失败会立即发生。我不希望将来的更新影响管道。

例如，假设我使用WKExtendedRuntimeSession，并将.start()方法封装在下面的startSession()中。如果我成功地包装了这个，我应该可以这样使用它：

manager.startSession()
.sink(
receiveCompletion: { result in
if result.isError {
showFailureToStartScreen()
}
},
receiveValue: { value in
showStartedSessionScreen()
})
.store(in: &cancellables)

一次性发布器之所以有用，是因为我们预计在调用方法后不久会调用以下两个方法之一：

• 成功：extendedRuntimeSessionDidStart(_:)
• 失败：extendedRuntimeSession(_:didInvalidateWith:error:)

此外，当会话暂停(或我们自己终止(时，我们不希望showFailureToStartScreen()等副作用随机发生。我们希望在代码的其他地方显式地处理它们。因此，在这里有一个一次性管道是有益的，所以我们可以保证sink只被调用一次。

---

我意识到，实现这一点的一种方法是使用Future，存储对Promise的引用，并在稍后调用promise，但这充其量看起来很棘手：

class Manager: NSObject, WKExtendedRuntimeSessionDelegate {
var session: WKExtendedRuntimeSession?
var tempPromise: Future<Void, Error>.Promise?
func startSession() -> AnyPublisher<Void, Error> {
session = WKExtendedRuntimeSession()
session?.delegate = self
return Future { promise in
tempPromise = promise
session?.start()
}.eraseToAnyPublisher()
}
func extendedRuntimeSessionDidStart(_ extendedRuntimeSession: WKExtendedRuntimeSession) {
tempPromise?(.success(()))
tempPromise = nil
}
func extendedRuntimeSession(_ extendedRuntimeSession: WKExtendedRuntimeSession, didInvalidateWith reason: WKExtendedRuntimeSessionInvalidationReason, error: Error?) {
if let error = error {
tempPromise?(.failure(error))
}
tempPromise = nil
}
}

这真的是与代表+一次性发布者合作的最优雅的方式吗？还是在Combine中有更优雅的方式？

---

作为参考，PromiseKit也有类似于Future.init的API。即Promise.init(resolver:)。然而，PromiseKit似乎也原生地支持我上面描述的pending()功能(示例(：

func startSession() -> Promise {
let (promise, resolver) = Promise.pending()
tempPromiseResolver = resolver
session = WKExtendedRuntimeSession()
session?.delegate = self
session?.start()
return promise
}

Answer 1

您可以确保使用.first()运算符的一次性发布器：

let subject = PassthroughSubject<Int, Never>()
let publisher = subject.first()
let c = publisher.sink(receiveCompletion: {
print($0)
}, receiveValue: {
print($0)
})
subject.send(1)
subject.send(2)

输出为：

1
finished

Answer 2

可能与任何感兴趣的人相关：

import Combine
import GoogleCast
struct GCKRequestPublisher: Publisher {
typealias Output = Void
typealias Failure = Error
let request: GCKRequest
init(request: GCKRequest) {
self.request = request
}
func receive<S>(subscriber: S) where S : Subscriber, Self.Failure == S.Failure, Self.Output == S.Input {
let subscription = GCKRequestSubscription(request: request, subscriber: subscriber)
subscriber.receive(subscription: subscription)
}
}
class GCKRequestSubscription<S: Subscriber>: NSObject, GCKRequestDelegate, Subscription where S.Input == Void, S.Failure == Error {
var subscriber: S?
let request: GCKRequest
enum GCKRequestError: Error {
case error(GCKError)
case abort(GCKRequestAbortReason)
}

init(request: GCKRequest, subscriber: S) {
self.subscriber = subscriber
self.request = request
super.init()
self.request.delegate = self
}
func request(_ demand: Subscribers.Demand) { }
func cancel() {
request.cancel()
subscriber = nil
}
// MARK: - GCKRequestDelegate
func requestDidComplete(_ request: GCKRequest) {
subscriber?.receive(completion: .finished)
}
func request(_ request: GCKRequest, didFailWithError error: GCKError) {
subscriber?.receive(completion: .failure(GCKRequestError.error(error)))
}
func request(_ request: GCKRequest, didAbortWith abortReason: GCKRequestAbortReason) {
subscriber?.receive(completion: .failure(GCKRequestError.abort(abortReason)))
}
}