swift - 我如何排队 URLSession.DataTaskPublisher 请求以便一次只发出一个请求？

Question

在下面的代码中，应用程序对象数组用于创建发布者数组，这些发布者数组被合并到发布对象数组中。

apps.map { latestRelease(app: $0) }.merge()

这是最新版本的完成方式。

func latestRelease(app: App) -> AnyPublisher<Release, Error> {
    do {
        let request = try requestFactory.make(.get, "apps/\(app.owner.name)/\(app.name)/releases/latest")

        return publisherFactory.make(for: request)
            .mapError{ $0 as Error }
            .map { data, _ in data }
            .decode(type: Release.self, decoder: decoder)
            .eraseToAnyPublisher()
    } catch {
        return Fail(error: error)
            .eraseToAnyPublisher()
    }
}

网络请求由工厂完成。

struct AppCenterPublisherFactory: DataTaskPublisherFactory {
    let session: URLSession

    init(session: URLSession = .shared) {
        session.configuration.httpMaximumConnectionsPerHost = 1
        self.session = session
    }

    func make(for request: URLRequest) -> URLSession.DataTaskPublisher {
        return session.dataTaskPublisher(for: request)
    }
}

问题是发布者立即发出网络请求。这会导致服务器返回 429 Too Many Requests。如何将 URLSession.DataTaskPublisher 请求排队，以便一次只发出一个请求每个请求之间有延迟？

Answer 1

如果您使用 append 而不是 merge 来组合发布者，它们将连续运行而不是同时运行。

对于延迟，您可以将 Empty().delay(for: cooldown, scheduler: whatever) 添加到除第一个发布者之外的每个发布者之前。

func releases<S: Scheduler>(of apps: [App], scheduler: S, cooldown: S.SchedulerTimeType.Stride)
    -> AnyPublisher<Release, Error>
{
    let singles = apps.map { latestRelease(app: $0) }
    guard let first = singles.first else { return Empty().eraseToAnyPublisher() }

    let combo: AnyPublisher<Release, Error> = singles.dropFirst()
        .reduce(first.eraseToAnyPublisher(), { combo, single in
            combo
                .append(Empty().delay(for: cooldown, scheduler: scheduler))
                .append(single)
                .eraseToAnyPublisher()
        })

    return combo
}

测试:

let testApps: [App] = [
    .init(name: "Facebook", owner: .init(name: "zuck")),
    .init(name: "Kindle", owner: .init(name: "bezos")),
    .init(name: "Crossword", owner: .init(name: "shortz")),
]

print("starting at \(Date())")
let ticket = releases(of: testApps, scheduler: DispatchQueue.global(qos: .utility), cooldown: .seconds(2))
    .sink(
        receiveCompletion: { print("got \($0) at \(Date())") },
        receiveValue: { print("got \($0) at \(Date())") })

输出:

starting at 2020-05-12 17:45:17 +0000
got Release(name: "Facebook") at 2020-05-12 17:45:17 +0000
got Release(name: "Kindle") at 2020-05-12 17:45:19 +0000
got Release(name: "Crossword") at 2020-05-12 17:45:21 +0000
got finished at 2020-05-12 17:45:21 +0000