ありがとうございます。 Subject等の状態を導入すればできそう、というのはイメージがつきます。ただ、Rx等だと、例えば以下の処理のように、監視の終了などを含め、1つのObservable内にencapsulateできるのが扱いやすくて便利だなぁと思っているのですが、CombineだとこのようにPublisher内に閉じ込めることは難しい（あまりやらない？）のでしょうか？

let observable = Observable<String>.create { observer in let token = webSocketClient.receiveMessage { message in observer.on(.next(message)) } return Disposables.create { // close connection when disposed token.invalidate() } }