Combine - Remove pipeline from container

Question

Assume the following simplified code-snippet

import Foundation
import Combine

public class NetworkFetch {
  fileprivate var networkPipelines : Set<AnyCancellable> = []
  
  public func loadDataFor(url : URL)
  { 
    URLSession.shared.dataTaskPublisher(for: url)
      .map { $0.data }
      .decode(type: City.self, decoder: JSONDecoder())
      .eraseToAnyPublisher()
      .sink(receiveCompletion: {_ in print("Finish")},
            receiveValue: { v in
              print("\(c)\n")
            }
      )
      .store(in: &networkPipelines)
  }
}

For each call of loadDataFor a new combine-pipeline is generated and added to the networkPipelines container. This container grows over time.

What is the correct way to remove such an URLSession-pipeline from this container as soon as all data is fetched by the URLSession-pipeline?

Related: https://stackoverflow.com/questions/60685810/remove-from-array-of-anycancellable-when-publisher-finishes — Cristik, Sep 03 '21 at 19:00

New Dev · Answer 1 · 2021-04-16T23:47:20.243

1

One thing you could do is remove your own subscription from inside sink:

But perhaps the better approach is to subscribe to a PassthroughSubject one time and send the requested URL and the callback through it:

private let subject = PassthroughSubject<(URL, (City) -> Void)), Never>()
private var c: Set<AnyCancellable> = []

init() {
   subject
      .flatMap { (url, callback) in
          URLSession.shared.dataTaskPublisher(for: url)
             .map(\.data)
             .decode(type: City.self, decoder: JSONDecoder())
             .zip(Just(callback).setFailureType(to: Error.self))
      }
      .sink(receiveCompletion: {_ in print("Finish")},
            receiveValue: { (city, callback) in
              callback(city)
            }
      )
      .store(in: &c)
}

public func loadDataFor(url : URL, callback: @escaping (City) -> Void) {
   subject.send(url, callback)
}

So, a single subscription can handle multiple requests by sending a pair of the requested URL and the callback through the subject.

edited Apr 16 '21 at 23:47

answered Apr 16 '21 at 16:43

New Dev

48,427
12
87
129

There are several calls to loadDataFor() in parallel, using just one "subject" variable is not a feasible solution. The action performed in in the .sink()-call is more complex (and race condition free). The print statement is just an example. – Patrick Bauers Apr 16 '21 at 19:22
@PatrickBauers, the missing detail in your question is what you want to do after each request? Just to make them, you can use a subject and `send` multiple request through it. But if each request needs its own a callback or something else, then it's more complicated. The way you phrased the question simplified it to the point where just using a subject should work – New Dev Apr 16 '21 at 19:24
Okay, I see. The real live problem has a closure that is passed to the sink()-call. – Patrick Bauers Apr 16 '21 at 20:54
So, for each request, you want to invoke a callback? Sort of something like this: `loadDataFor(url : URL, callback: (City) -> Void)`? @PatrickBauers – New Dev Apr 16 '21 at 21:33

Combine - Remove pipeline from container

1 Answers1