为什么这个组合管道不让项目通过？

Question

我遇到了一个组合问题，我找不到合适的解决方案。

我的目标是监控一个队列，并处理队列中的项目直到它为空。如果然后有人在队列中添加更多项目，我会继续处理。物品需要一件一件处理，我不想丢失任何物品。

我在下面写了一个非常简化的队列来重现这个问题。为了简单起见，我的项目再次建模为字符串。

鉴于上述限制：

• 我在队列中使用changePublisher 来监控更改。
• 一个按钮让我可以向队列中添加一个新项目
• flatMap 运算符依赖于maxPublishers 参数，只允许一个进行中的处理。
• buffer 运算符可防止在 flatMap 忙时丢失项目。

此外，我使用combineLatest 运算符仅在某些条件下触发管道。为简单起见，我在这里使用Just(true) 发布者。

问题

如果我点击按钮，第一个项目会进入管道并被处理。 changePublisher 触发是因为队列被修改（项目被删除），并且管道在compactMap 处停止，因为peek() 返回nil。到现在为止还挺好。不过，之后，如果我再次点击该按钮，则会在管道中发送一个值，但 永远不会通过 buffer。

解决方案？

我注意到删除 combineLatest 可以防止问题发生，但我不明白为什么。

代码

import Combine
import UIKit

class PersistentQueue {
    let changePublisher = PassthroughSubject<Void, Never>()

    var strings = [String]()

    func add(_ s: String) {
        strings.append(s)
        changePublisher.send()
    }

    func peek() -> String? {
        strings.first
    }

    func removeFirst() {
        strings.removeFirst()
        changePublisher.send()
    }
}

class ViewController: UIViewController {

    private let queue = PersistentQueue()
    private var cancellables: Set<AnyCancellable> = []

    override func viewDidLoad() {
        super.viewDidLoad()
        start()
    }

    @IBAction func tap(_ sender: Any) {
        queue.add(UUID().uuidString)
    }

    /*
     Listen to changes in the queue, and process them one at a time. Once processed, remove the item from the queue.
     Keep doing this until there are no more items in the queue. The pipeline should also be triggered if new items are
     added to the queue (see `tap` above)
     */
    func start() {
        queue.changePublisher
            .print("Change")
            .buffer(size: Int.max, prefetch: .keepFull, whenFull: .dropNewest)
            .print("Buffer")
            // NOTE: If I remove this combineLatest (and the filter below, to make it compile), I don't have the issue anymore.
            .combineLatest(
                Just(true)
            )
            .print("Combine")
            .filter { _, enabled in return enabled }
            .print("Filter")
            .compactMap { _ in
                self.queue.peek()
            }
            .print("Compact")
            // maxPublishers lets us process one page at a time
            .flatMap(maxPublishers: .max(1)) { reference in
                return self.process(reference)
            }
            .sink { reference in
                print("Sink for \(reference)")

                // Remove the processed item from the queue. This will also trigger the queue's changePublisher,
                // which re-run this pipeline in case
                self.queue.removeFirst()
            }
            .store(in: &cancellables)
    }

    func process(_ value: String) -> AnyPublisher<String, Never> {
        return Future<String, Never> { promise in
            print("Starting processing of \(value)")
            DispatchQueue.global(qos: .userInitiated).asyncAfter(deadline: .now() + 2) {
                promise(.success(value))
            }
        }.eraseToAnyPublisher()
    }

}

输出

如果您点击按钮两次，以下是流水线的运行示例：

Change: receive subscription: (PassthroughSubject)
Change: request max: (9223372036854775807)
Buffer: receive subscription: (Buffer)
Combine: receive subscription: (CombineLatest)
Filter: receive subscription: (Print)
Compact: receive subscription: (Print)
Compact: request max: (1)
Filter: request max: (1)
Combine: request max: (1)
Buffer: request max: (1)
Change: receive value: (())
Buffer: receive value: (())
Combine: receive value: (((), true))
Filter: receive value: (((), true))
Compact: receive value: (3999C98D-4A86-42FD-A10C-7724541E774D)
Starting processing of 3999C98D-4A86-42FD-A10C-7724541E774D
Change: request max: (1) (synchronous)
Sink for 3999C98D-4A86-42FD-A10C-7724541E774D // First item went through pipeline
Change: receive value: (())
Compact: request max: (1)
Filter: request max: (1)
Combine: request max: (1)
Buffer: request max: (1)
Buffer: receive value: (())
Combine: receive value: (((), true))
Filter: receive value: (((), true))

// Second time compactMap is hit, value is nil -> doesn't forward any value downstream.

Filter: request max: (1) (synchronous)
Combine: request max: (1) (synchronous)
Change: request max: (1)

// Tap on button

Change: receive value: (())

// ... Nothing happens

[编辑] 这是一个更受限制的示例，它可以在 Playgrounds 中运行并且也演示了问题：

import Combine
import Foundation

import PlaygroundSupport

PlaygroundPage.current.needsIndefiniteExecution = true

func process(_ value: String) -> AnyPublisher<String, Never> {
    return Future<String, Never> { promise in
        print("Starting processing of \(value)")
        DispatchQueue.global(qos: .userInitiated).asyncAfter(deadline: .now() + 0.1) {
            promise(.success(value))
        }
    }.eraseToAnyPublisher()
}

var count = 3

let s = PassthroughSubject<Void, Never>()

var cancellables = Set<AnyCancellable>([])

// This reproduces the problem. Switching buffer and combineLatest fix the problem…

s
    .print()
    .buffer(size: Int.max, prefetch: .keepFull, whenFull: .dropNewest)
    .combineLatest(Just("a"))
    .filter { _ in count > 0 }
    .flatMap(maxPublishers: .max(1)) { _, a in process("\(count)") }
    .sink {
        print($0)
        count -= 1
        s.send()
    }
    .store(in: &cancellables)

s.send()

Thread.sleep(forTimeInterval: 3)

count = 1
s.send()

切换组合和缓冲区可以解决问题。

Answer 1

我不知道为什么管道被阻塞，但是当队列为空时没有理由发布。解决这个问题为我解决了问题。

func removeFirst() {
    guard !strings.isEmpty else {
        return
    }
    strings.removeFirst()
    if !self.strings.isEmpty {
        self.changePublisher.send(self.strings.first)
    }
}

Answer 2

刚刚尝试了您的示例。如果缓冲区放在 flatMap 之前，它会按预期工作。并根据下面 Paulw11 的回答更新 removeFirst

queue.changePublisher
            .print("Change")
            // NOTE: If I remove this combineLatest (and the filter below, to make it compile), I don't have the issue anymore.
            .combineLatest(Just(true))
            .print("Combine")
            .filter { _, enabled in return enabled }
            .print("Filter")
            .compactMap { _ in
                self.queue.peek()
            }
            .print("Compact")
            // maxPublishers lets us process one page at a time
            .buffer(size: Int.max, prefetch: .keepFull, whenFull: .dropNewest)
            .print("Buffer")
            .flatMap(maxPublishers: .max(1)) { reference in
                return self.process(reference)
            }
            .sink { reference in
                print("Sink for \(reference)")

                // Remove the processed item from the queue. This will also trigger the queue's changePublisher,
                // which re-run this pipeline in case
                self.queue.removeFirst()
                print("COUNT: " + self.queue.strings.count.description)
            }
            .store(in: &cancellables)