share、makeConnectable、connect

Controlling Publishing with Connectable Publishers

share() を消すとどうなるか？
=> 購読が1本ではなく、2本発生するので、1も2も両方 valueとfinished を無事受け取れる。

Received data 1: 1256 bytes.
Received completion 1: finished.
Received data 2: 1256 bytes.
Received completion 2: finished.

これだとPublisherの生成コストがエコではないから、share()を使ってPublisherを struct => class化し、
自身を参照型とすることで、購読を1つにすることができる。
ただし、1つにすることで発生する注意点もある。その例が、1つのPubに対し、複数のSubがくっつく時だ。
「1つめのSubにくっついた時点で」Pubは値の放出を開始するので、
2つめのSubがくっつくのが遅く、PubがdataTask() でダウンロードを既に完了してしまっている場合(finishを流している場合)、
2つめのSubはValueを受け取れない(finishだけは受け取れる。)

Received data 1: 1256 bytes.
Received completion 1: finished.
Received completion 2: finished.

そこで makeConnectable() で PubをConnectablePublisher化する。

ConnectablePublisherにはconnect() が生えており、これを明示的に呼ばれるまで、
Subの購読が開始されていたとしても値の放出を行わない。
2つのSubへの接続完了がしっかり保証されたうえで、安心してconnect()を呼ぼう。

<注>
connect() は Cancellable を返す。

1. これに対し cancel() を呼ぶ
2. Cancellableをdeinitする

のどちらかで、値の発行を止めることができる。

(dataTaskPublisherで「通信がキャンセルされました」が出るのは、値が返ってくる前に、let _ 、もしくは返り値を無視することによって、
購読が破棄されていたから、なのだな...)


let url = URL(string: "https://example.com/")!
let connectable = URLSession.shared
  .dataTaskPublisher(for: url)
  .map() { $0.data }
  .catch() { _ in Just(Data() )}
  .share()
  .makeConnectable()

connectable
  .sink(receiveCompletion: { print("Received completion 1: \($0).") },
        receiveValue: { print("Received data 1: \($0.count) bytes.") })
  .store(in: &cancellables)

DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
  connectable
      .sink(receiveCompletion: { print("Received completion 2: \($0).") },
            receiveValue: { print("Received data 2: \($0.count) bytes.") })
      .store(in: &self.cancellables)
}

DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
  // connect() は、MakeConnectableというConnectablePublisherに生えている
  connectable.connect()
      .store(in: &self.cancellables)
}


いくつかのPubは、既にConnectablePublisherとして実装されている。
Multicast とか、TimerPublisher とか。
これらのSubに対し複数のSubがくっつかないのであれば、
こららのSubに 明示的な connect() のコールはわずらわしい。
そこで autoConnect()。
これをしておけば、Subが接続された時点で(connectを呼ばなくても)、値の発行が開始される。
     
Timer.publish(every: 1, on: .main, in: .default)
    .autoconnect()
    .sink() { date in
        print ("Date now: \(date)")
    }
    .store(in: &cancellables)

Q. そもそも share、multicast、makeConnectable による connectable publisher変換は、どういったときに使うのか？
(share は connectable publisherを返しているわけではないので、connectable publisher変換ではない気がするが)

A. 重い処理、コスト大の処理。API通信など。

ただし、そのPubを購読するSubが1つだけなのであれば、使う必要はないのかも、という気がする。。。

Q. share() って、なにをシェアするの？

A. 「本来なら複数になる購読を1つの購読にまとめ、その購読を複数のSubで」shareする、ということ。

購読自体は1つだけなのにもかかわらず、そのPubには複数のSub が紐付いており、
sinkで別の副作用を実行できたり、assignで値をセットしたりできる。

https://forums.swift.org/t/combine-what-are-those-multicast-functions-for/26677/31

There are three things to distinguish:

Subject

Multicast

Share

Subject is already a class so we get reference semantics. Thus a Subject is already a multicaster and there is no need to do anything else.

Multicast takes advantage of that fact; it takes as its parameter a function that produces a Subject and holds on to that, thus allowing us to multicast to any subscribers. It is a ConnectablePublisher so the upstream won't actually start publishing until you say connect.

Share is a convenience wrapper for Multicast; it appends autoconnect so the upstream starts publishing as soon as we are subscribed to.

I now have a free online tutorial with examples that may help clarify:

https://www.apeth.com/UnderstandingCombine/operators/operatorsSplitters/operatorssplitters.html 2

https://www.apeth.com/UnderstandingCombine/operators/operatorsSplitters/operatorsshare.html 2

https://www.apeth.com/UnderstandingCombine/operators/operatorsSplitters/operatorsmulticast.html 3