结合 uncollect 运算符？

Question

我正在寻找与collect 运算符相反的运算符：

• 允许您从一个发射值映射到多个发射值，并在流上发射它们。例如，如果我在流中有元素[1,2,3]，我想将其转换为发出1 然后2，然后3 的流。
• 不更改流的完成状态。诚然，这个uncollect 运算符不会与collect 完全相反，因为collect 一直等到流完成。对于uncollect，它应该只在流未完成时处理结果，而不是尝试完成/取消流。

例如，我想这是 uncollect 运算符的运作方式：

func fibonacci(_ number: Int) -> AnyPublisher<Int, Never> {
  Future { ... }.eraseToAnyPublisher()
}

let serverEventStream: AnyPublisher<[Int], Never> = ...

serverEventStream              // AnyPublisher<[Int], Never>  // Ex. 2 values: `[12, 24]`, `[1, 10, 50]`
  .uncollect { $0 }            // AnyPublisher<Int, Never>    // Ex. 5 values: `12`, `24`, `1`, `10`, `50`
  .flatMap { fibonacci($0) }   // AnyPublisher<Int, Never>    // Ex. 5 values: `144`, `46368`, `1`, `55`, `12586269025`
  .sink { print($0) }

我一直在寻找 explode、splat 或 uncollect 之类的名称，但无济于事。共享一个有希望的名称的最接近的运算符是flatMap，因为Sequence.flatMap 在非组合世界中是等价的。但是，Combine's flatMap is the way to chain promises together。

有没有办法取消收集元素，或者将单个发射映射到任意数量的发射？

---

我能够让它与我的 custom JustSeveral Publisher 一起工作，但这似乎很笨拙：

class Tests: XCTestCase {
  func testUncollect() {
    func fibonacci(_ number: Int) -> AnyPublisher<Int, Never> {
      switch number {
      case 1: return Just(1).eraseToAnyPublisher()
      case 10: return Just(55).eraseToAnyPublisher()
      case 12: return Just(144).eraseToAnyPublisher()
      case 24: return Just(46368).eraseToAnyPublisher()
      case 50: return Just(12586269025).eraseToAnyPublisher()
      default: fatalError("Should actually build this function.")
      }
    }

    let serverEventStream: AnyPublisher<[Int], Never> = JustSeveral([[12, 24], [1, 10, 50]]).eraseToAnyPublisher()

    serverEventStream              // AnyPublisher<[Int], Never>  // Ex. 2 values: `[12, 24]`, `[1, 10, 50]`
      .uncollect { $0 }            // AnyPublisher<Int, Never>    // Ex. 5 values: `12`, `24`, `1`, `10`, `50`
      .flatMap { fibonacci($0) }   // AnyPublisher<Int, Never>    // Ex. 5 values: `144`, `46368`, `1`, `55`, `12586269025`
      .sink { print($0) }
  }
}

extension Publisher {
  func uncollect<T>(_ transform: @escaping (Output) -> [T]) -> AnyPublisher<T, Failure> {
    self
      .flatMap { output -> AnyPublisher<T, Failure> in
        JustSeveral(transform(output))
          .setFailureType(to: Failure.self)
          .eraseToAnyPublisher()
      }
      .eraseToAnyPublisher()
  }
}

Answer 1

您可以将flatMap 与新发布者一起使用

[1,2,3,4,5,6].publisher
    .collect()
    .flatMap { $0.publisher }
    .sink { print($0) }

1
2
3
4
5
6