Сбор значений Publisher с помощью async

Мне кажется, проблема здесь в том, что вы ожидаете, что collectValues будет содержать «10» и «20», которые вы отправляете после установки константы. subject.send(10) добавляет значения к subject.values, а не к collectValues, поэтому тест не пройден.

Изменение такого порядка должно привести к успешному завершению теста.

// Send some data through `subject` and then `.finish` it.
subject.send(10)
subject.send(20)
subject.send(completion: .finished)

// I'm expecting this to only complete once `subject` finishes. I've used
// `async let` so I can poke some data through `subject` and then later on
// `await collectValues` to hopefully get back the stuff published by 
// `subject`. In the real test this is a property from the unit under test
// which runs various operators on `subject`.
async let collectValues = await subject.values.reduce(into: []) { $0.append($1) }

То, что происходит, довольно просто. Как это написать правильно, гораздо менее понятно, и моя рекомендация — «не делайте этого».

Во-первых, небольшая проблема, которая не является проблемой:

async let collectValues = await subject.values.reduce(into: []) { $0.append($1) }

Вы не должны использовать await здесь. Это, вероятно, было бы проблемой, если бы не было других проблем.

Основная проблема заключается в том, что PassthroughSubject отбрасывает сообщения, если нет подписчика. В вашем текущем коде это обязательно произойдет, но это также очень сложно исправить.

// Taking out the extra `await`
async let collectValues = subject.values.reduce(into: []) { $0.append($1) }

// That line is pretty close to:
    let collectValues = Task {
        var values: [Int] = []
        for await value in subject.values {
            values.append(value)
        }
        return values
    }

Проблема в том, что это запускает задачу, которая может не начаться сразу. Итак, ваша следующая строка кода subject.send(10) не имеет подписчика (она даже не дошла до строки for-await), и она просто выброшена.

Вы можете исправить это, добавив try await Task.sleep(for: .seconds(1)) после создания задачи, но это мало помогает. На PassthroughSubject нет буферизации. Пока ты звонишь append, тебя никто не слушает. Ценность будет выброшена, и вы сбросите 20.

Вы можете улучшить ситуацию с помощью буферизации, но вам все равно нужно спать (что неприемлемо для IMO). Тем не менее, для меня очень надежным является следующее:

func test_collectingPassthroughValues() async throws {
    // In the real test this is injected in to the unit under test.
    let subject = PassthroughSubject<Int, Never>()
    let readSubject = subject.buffer(size: 10, prefetch: .keepFull, whenFull: .dropOldest)

    async let collectValues = readSubject.values.reduce(into: []) { $0.append($1) }

    try await Task.sleep(for: .seconds(1))
    subject.send(10)
    subject.send(20)
    subject.send(completion: .finished)

    // Await the values so we can check we got what's expected.
    let values = await collectValues

    XCTAssertEqual(values, [10, 20])
}

Но ИМО, это полностью сломанный подход.

Я бы не стал смешивать PassthroughSubject с .values. Я просто не вижу способа сделать его прочным. В более широком смысле я рекомендую очень осторожно смешивать Combine и Structured Concurrency. У них, как правило, очень разные представления о том, как все должно работать.