下面是一个Playground的源代码，它演示了我认为您正在尝试实现的功能，并有一些代码来练习该功能。
在代码中，publisher1和publisher2是PassthroughSubjects，这是一个简单的发布器，在示例中使用非常方便，因为您可以在任何时候通过它们发送值。
在subscribeToPipelines中，代码构建了一个新的管道withTimeout，它从pipeline2获取输入，如果在1秒内没有发布任何内容，则会超时。当pipeline1通过flatMap接收值时，该管道将进行控制
withTimeout的结果由sink运算符捕获。代码处理withTimeout的可能条件，它可能已经正常完成，它可能已经超时，并且它可能接收一个值。print语句涵盖了所有情况。
代码使用Task来执行管道，因为Task.sleep是一种简单的延迟方式。第一个练习展示了如果publisher1接收到一个值，而publisher2在半秒后接收到一个值，会发生什么。第二个练习展示了如果publisher2延迟发布值1.5秒（超时）会发生什么。

import UIKit
import Combine

enum MyErrors: Error {
    case publisher2Timedout
}

let publisher1 = PassthroughSubject<Int, Error>()
let publisher2 = PassthroughSubject<Int, Error>()

func subscribeToPipelines() -> AnyCancellable {
    let withTimeout = publisher2.timeout(1,
                                         scheduler: RunLoop.main,
                                         customError: { return MyErrors.publisher2Timedout })

    let subscription = publisher1.flatMap { value1 in
        print("The first publisher's result was \(value1)")
        return withTimeout
    }
        .sink { completion in
            switch completion {
                case .finished:
                    print("The entire pipeline finished")
                case .failure(let error):
                    if case MyErrors.publisher2Timedout = error {
                        print("The second operation timed out")
                    }
            }
        } receiveValue: { value2 in
            print("The second publisher's result is \(value2)")
        }

    return subscription
}

Task {
    var subscription: AnyCancellable? =  subscribeToPipelines()
    // Try a case where the operations succeed
    publisher1.send(3)
    try await Task.sleep(for: .milliseconds(500))
    publisher2.send(4)

    subscription = nil;

    try await Task.sleep(for: .milliseconds(500))

    subscription =  subscribeToPipelines()
    publisher1.send(5)
    try  await Task.sleep(for: .milliseconds(1500))
    publisher2.send(6)

    subscription = nil;
}