iOS13 的合并流在操作员使用调度程序后不会流动

Question

iOS13 的发布者组合流在操作员使用调度程序后似乎没有流动。

这是我的代码：

import Foundation
import Combine
struct MyPublisher: Publisher {
    typealias Output = Int
    typealias Failure = Error
    func receive<S>(subscriber: S) where S : Subscriber,
        Failure == S.Failure,
        Output == S.Input {
            subscriber.receive(1)
            print("called 1")
            subscriber.receive(2)
            print("called 2")
            subscriber.receive(completion: .finished)
            print("called finish")
    }
}
MyPublisher()
//    .receive(on: RunLoop.main) // If this line removed, it will be fine.
//    .throttle(for: .milliseconds(1000), scheduler: RunLoop.main, latest: false)) // If this line removed, it will be fine.
//    .debounce(for: .milliseconds(1000), scheduler: RunLoop.main)) // If this line removed, it will be fine.
//    .delay(for: .milliseconds(1000), scheduler: DispatchQueue.main)) // If this line removed, it will be fine.
    .print()
    .sink(receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("finished")
        case .failure(let error):
            print("error:(error)")
        }
    }, receiveValue: { num in
        print("(num)")
    })

我预计输出是

1
2
finished

但实际输出什么都没有。

如果我不使用receive或throttle或debounce或delay.输出会很好。

是错误还是我的代码有问题？

我尝试了Playground(Xcode 11 beta3(。

Answer 1

订阅：

我不确定为什么它在单个线程的情况下有效，但您应该确保在subscriber上调用received(subscription:)。如果您不需要处理订阅者的需求，您可以使用Subscribers.empty：

struct MyPublisher: Publisher {
    typealias Output = Int
    typealias Failure = Never
    func receive<S>(subscriber: S) where S : Subscriber, Failure == S.Failure, Output == S.Input {
        subscriber.receive(subscription: Subscriptions.empty)
        _ = subscriber.receive(1)
        Swift.print("called 1")
        _ = subscriber.receive(2)
        Swift.print("called 2")
        _ = subscriber.receive(completion: .finished)
        Swift.print("called finish")
    }
}

任何可取消：

您应该注意到一条警告：

调用"sink(receiveCompletion：receiveValue：("的结果未使用

这应该出现，因为sink返回一个AnyCancellable：

func sink(receiveCompletion: @escaping ((Subscribers.Completion<Self.Failure>) -> Void), receiveValue: @escaping ((Self.Output) -> Void)) -> AnyCancellable

任何返回AnyCancellable的内容都将在解除分配AnyCancellable后立即被取消。

我的猜测是，如果您将其放在另一个线程上，那么当调用方法结束时，可取消将在收到订阅之前解除分配。但是当在当前线程上收到时，它似乎正在及时执行以显示订阅和输出。最有可能的是，当当前线程退出时，可取消的线程正在被释放。

Answer 2

使用可取消

例如：

class ImageLoader: ObservableObject {
    @Published var image: UIImage?
    private var cancellable: AnyCancellable?
func fetchImages() {
        guard let urlString  = urlString,
            let url =  URL(string: urlString) else { return }
        cancellable = URLSession.shared.dataTaskPublisher(for: url)
            .map { UIImage(data: $0.data) }
            .replaceError(with: nil)
            .receive(on: DispatchQueue.main)
            .sink { [weak self] in self?.image = $0 }
    }
}

Answer 3

使用下划线

可以通过下划线传递警告。我用了奈什塔回答中的例子。

例如

class ImageLoader: ObservableObject {
    @Published var image: UIImage?
    
    func fetchImages() {
        guard let urlString  = urlString,
            let url =  URL(string: urlString) else { return }
        _ = URLSession.shared.dataTaskPublisher(for: url)
            .map { UIImage(data: $0.data) }
            .replaceError(with: nil)
            .receive(on: DispatchQueue.main)
            .sink { [weak self] in self?.image = $0 }
    }
}