Intermediate Combine

In Combine, Publishers are usually structs, and therefore passed by value throughout your code, copying the publisher each time it passed around. Once it recieves a subscription, those individual publishers will do the only thing they can - start its work and deliver values.

But what if you want to pass around the publisher by reference instead? This is where the share() operator comes into play; it returns an instance of Publisher.Share and shares the upstream publisher with any subscribers that come after it. It will subscribe to the upstream publisher once with the first incoming subscriber; all other subscribers get values relayed to it by the Publisher.Share object.

Note that there is no sense of buffering or replay involved; new subscribers will only get values emitted after they have subscribed. If a new subscriber comes in after the publisher has completed, it only receives the completion event. Let’s take a look at that in an example.

Just like in the previous examples, make a subscription on a URLSession.shared instance and use a dataTaskPublisher(for:) operator to get data from raywenderlich.com

example(of: "shared") {

let shared = URLSession.shared
  .dataTaskPublisher(for: rwUrl)

Use the map operator to grab the data part of the tuple from dataTaskPublisher(for:), add a print operator to output “shared” to the console, and then add the share() operator

  .map(\.data)
  .print("shared")
  .share()

At this point the pipeline is waiting for the initial subscription to share() to take place. Add that via a sink subscriber, and have it output that subscription1 was received.

print("subscribing first")

shared.sink(
  receiveCompletion: { _ in },
  receiveValue: { print("subscription1 received: '\($0)'") }
)
.store(in: &subscriptions)

This sink subscriber is now subscribed to the upstream publisher via the share() operator. Add another sink subscriber.

print("subscribing second")

var subscription2 = shared.sink(
  receiveCompletion: { _ in },
  receiveValue: { print("subscription2 received: '\($0)'") }
)
.store(in: &subscriptions)

Run this playground. subscription1 is actually what triggers work, the network request, to take place in the publisher. The second subscriber however just gets information relayed to it by the share() operator.

Note that in the console output only one instance of “shared: received value” is seen - this is where subscription1 connects to the publisher. There is no corresponding request from subscription2, since it simply gets the relayed information and doesn’t make a formal connection.

What if the second subscriber comes after the request has completed? You could simulate this case by delaying the second subscription.

var subscription2: AnyCancellable? = nil

DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
  print("subscribing second")

  subscription2 = shared.sink(
    receiveCompletion: { print("subscription2 completion \($0)") },
    receiveValue: { print("subscription2 received: '\($0)'") }
  )
}

The share operator you just learned about is great for cases where subscribers get connected to the pipeline over time and don’t care about the data that was emitted before it connected. But what if you want to wait until all of your subscribers are in place before starting to publish? That’s where the multicast operator can help.

The unique quality of multicast is that the publisher it returns is a ConnectablePublisher, which means it won’t subscribe to the upstream publisher until you call connect(). This leaves you time to setup all the subscribers you need before letting it connect to the publisher. Let’s look at this in an example.

Let’s modify the example you wrote when learning about share(). Prepare a PassthroughSubject that takes in Data and emits URLError if things go wrong.

// 1
example(of: "multicast") {
let subject = PassthroughSubject<Data, URLError>() // 2

Prepare a multicasted publisher, using the shared property of URLSession. Use a dataTaskPublisher to connect to raywenderlich.com, use map to grab the data property from the tuple, add a print operator to see what the pipeline is doing, and then call the multicast operator with the subject you just made.

// 2
let multicasted = URLSession.shared
  .dataTaskPublisher(for: rwUrl)
  .map(\.data)
  .print("shared")
  .multicast(subject: subject)

Now make 2 subscriptions to the shared, or multicasted, publisher via sink subscribers. Use the receivedValue closure to output a statement to the console stating that the subscription has been recevied.

// 3
multicasted
  .sink(
    receiveCompletion: { _ in },
    receiveValue: { print("subscription1 received: '\($0)'") }
  )

multicasted
  .sink(
    receiveCompletion: { _ in },
    receiveValue: { print("subscription2 received: '\($0)'") }
  )

Now with the subscribers in place, instruct the multicasted publisher to connect to its upstream publisher with connect().

// 4
multicasted.connect()

Finally, tell the subject to send empty data as a test to see that both subscriptions receive data.

// 5
subject.send(Data())

Run this in the playground. You can see that both subscribers get the empty data.