订阅在源发布者完成时取消主题

Question

我有一个函数，它接收一个 Publisher 并创建一个用于两者的 PassthroughSubject：

1. • 订阅源发布者
2. • send 手动值

例如：

class Adapter<T>{
  let innerSubject=PassThroughSubject<T,Never>()
  let scope = Scope() // custom type, Array<AnyCancellable>
  func init(_ source:Publisher<T,Never>){
    source.register(innerSubject).in(scope)
    innerSubject.sink(receiveValue:{debugPrint($0)}).in(scope)
  }
  func adapt(_ T val){
    innerSubject.send(val)
  }
}

fn usage(){
  let adapter=Adapter(Empty()) // change to Empty(completeImmediately:false) to fix
  adapter.adapt(42) // should print 42,
}

内部主题似乎被取消了，因为Empty calls complete。 我希望内部主题的 subscription 被取消，而不是主题本身。在 .Net 中至少是这样的行为——我是不是遗漏了什么？我需要将其包装在广播主题中吗？我认为发布者可以接收多个接收者并向每个接收者进行多播？

Answer 1

在您当前的实现中，内部 PassThroughSubject 将在 Empty 发布者完成时取消。这是因为您使用的 in(scope) 修饰符将内部主题的生命周期附加到范围数组的生命周期，而当 Empty 发布者完成时，它又被取消。因此，内部对象的订阅将被取消，后续调用适配将没有效果。

如果你想在 Empty 发布者完成后保持内部 PassThroughSubject 存活，你可以从 register 和 sink 调用中删除 in(scope) 修饰符，而是将它们返回的可取消对象存储在你的 Adapter 实例中。这样，内部 PassThroughSubject 的生命周期将不会与范围数组的生命周期相关联。

class Adapter<T> {
  let innerSubject = PassthroughSubject<T, Never>()
  var cancellables = Set<AnyCancellable>()

  init(_ source: Publisher<T, Never>) {
    source.subscribe(innerSubject).store(in: &cancellables)
    innerSubject.sink(receiveValue: { debugPrint($0) }).store(in: &cancellables)
  }

  func adapt(_ val: T) {
    innerSubject.send(val)
  }
}

这里的 subscribe 和 sink 调用 return AnyCancellable objects，它们被添加到 cancellables 集合中。即使在 Empty 发布者完成后，这也会使内部 PassThroughSubject 保持活动状态，并且后续调用 adapt 仍将具有预期的效果。

另请注意，可以使用store(in:)修饰符将可取消的对象存储在cancellables集合中，这样可以简化代码并确保在cancellable实例被释放时正确释放Adapter对象。