swift - 创建一个Publisher，一一通知订阅者，互相等待

Question

我有这个发布者和订阅者(​​示例代码):

import Combine

let publisher = PassthroughSubject<ComplexStructOrClass, Never>()
let sub1 = publisher.sink { (someString) in
    // Async work...
}

let sub2 = publisher.sink { (someString) in
    // Async work, but it has to wait until sub1 has finished his work
}

因此 publisher 常量有 2 个订阅者。当我在 publisher 常量上使用 send 方法时，它应该首先将值发送到 sub1 并之后 sub1 完成处理(通过回调或类似的方式)，publisher 应该通知 sub2。

因此在评论中指出，Combine 就是为此而设计的。我需要使用哪个出版商？ PassthroughSubject 可能是错误的决定。

用例

我需要在应用程序的整个生命周期中向几个不同的发布者的动态数量的订阅者发布值(我希望我可以制定一个协议(protocol))。因此，可以在任何给定时间向发布者添加或删除订阅者。订阅者如下所示:

• 它有一个NSPercientContainer

• 当新值到达时，发布者应进行回调。该过程如下所示:

  1. 发布者将创建订阅者容器的backgroundContext，因为它知道订阅者有一个容器
  2. 发布者将上下文连同新发布的值一起发送给订阅者
  3. 发布者等待，直到收到订阅者的回调。订阅者不应保存上下文，但发布者必须保存对上下文的引用。订阅者给出一个枚举的回调，其中有一个正常的情况和一些错误的情况。
  4. 当订阅者发出带有错误枚举情况的回调时，发布者必须回滚它为每个订阅者创建的上下文。
  5. 当订阅者发出 ok 情况的回调时，发布者会为每个订阅者重复步骤 1 到 5
  6. 仅当没有订阅者给出错误枚举情况或没有订阅者时才会执行此步骤。发布者将保存订阅者创建的所有上下文。

当前代码，没有合并这是一些未使用 Combine 的代码:

// My publisher
protocol NotiPublisher {

    // Type of message to send
    associatedtype Notification

    // List of subscribers for this publisher
    static var listeners: Set<AnyNotiPublisher<Notification>> { get set }
}

// My subscriber
protocol NotificationListener: Hashable {
    associatedtype NotificationType

    var container: NSPersistentContainer { get }
    // Identifier used to find this subscriber in the list of 'listeners' in the publisher
    var identifier: Int32 { get }
    var notify: ((_ notification: NotificationType, _ context: NSManagedObjectContext, @escaping CompletionHandlerAck) -> ()) { get }
}

// Type erased version of the NotificationListener and some convience methods here, can add them if desired

// In a extension of NotiPublisher, this method is here
static func notify(queue: DispatchQueue, notification: Notification, completionHander: @escaping CompletionHandlerAck) throws {
    let dispatchGroup = DispatchGroup()
    var completionBlocks = [SomeCompletionHandler]()
    var contexts = [NSManagedObjectContext]()
    var didLoop = false

    for listener in listeners {
        if didLoop {
            dispatchGroup.wait()
        } else {
            didLoop = true
        }

        dispatchGroup.enter()

        listener.container.performBackgroundTask { (context) in
            contexts.append(context)

            listener.notify(notification, context, { (completion) in                    
                completionBlocks.append(completion)

                dispatchGroup.leave()
            })
        }
    }

    dispatchGroup.notify(queue: queue) {
        let err = completion.first(where: { element in
            // Check if an error has occured
        })

        if err == nil {
            for context in contexts {
                context.performAndWait {
                    try! context.save()
                }
            }
        }

        completionHander(err ?? .ok(true))
    }
}

这是相当复杂的代码，我想知道是否可以利用合并的强大功能来使代码更具可读性。

Answer 1

我编写了以下代码来使用 flatMap 链接来自发布者的异步操作，允许您返回另一个发布者。我不是粉丝，它可能无法满足您动态更改潜艇的需求，但它可能会对某人有所帮助:

let somePublisher = Just(12)

let anyCancellable = somePublisher.flatMap{ num in
    //We can return a publisher from flatMap, so lets return a Future one because we want to do some async work
    return Future<Int,Never>({ promise  in

        //do an async thing using dispatch
        DispatchQueue.main.asyncAfter(deadline: .now() + 3, execute: {
            print("flat map finished processing the number \(num)")

            //now just pass on the value
            promise(.success(num))
        })
    })
}.flatMap{ num in
    //We can return a publisher from flatMap, so lets return a Future one because we want to do some async work
    return Future<Int,Never>({ promise  in

        //do an async thing using dispatch
        DispatchQueue.main.asyncAfter(deadline: .now() + 3, execute: {
            print("second flat map finished processing the number \(num)")

            //now just pass on the value
            promise(.success(num))
        })
    })
}.sink { num in
    print("This sink runs after the async work in the flatMap/Future publishers")
}