title: Dispose (cancel) observable. SubscribeOn and observeOn different schedulers
body: <p><strong>Reformed question</strong></p>
  
  <p>I have reformed my question. To the common case.</p>
  
  <p>I want to generate items with RxSwift in background thread (loading from disk, long-running calculations, etc.), and observe items in MainThread. And I want to be sure that no items will be delivered after dispose (from main thread).</p>
  
  <p>According to documentation (<a href="https://github.com/ReactiveX/RxSwift/blob/master/Documentation/GettingStarted.md#disposing" rel="nofollow noreferrer">https://github.com/ReactiveX/RxSwift/blob/master/Documentation/GettingStarted.md#disposing</a>):</p>
  
  <blockquote>
    <p>So can this code print something after the dispose call is executed? The answer is: it depends.</p>
    
    <ul>
    <li><p>If the scheduler is a serial scheduler (ex. MainScheduler) and dispose is called on the same serial scheduler, the answer is no.</p></li>
    <li><p>Otherwise it is yes.</p></li>
    </ul>
  </blockquote>
  
  <p>But in case of using subscribeOn and observerOn with different schedulers - we cannot guarantee that nothing will be emitted after dispose (manual or by dispose bag, it does not matter).</p>
  
  <p>How should I generate items (images, for example) in background and be sure that result will not be used after the dispose?</p>
  
  <p>I made workaround in real project, but I want to solve this problem and to understand how should we avoid it in the same cases.</p>
  
  <p>In my test project I have used small periods - they demonstrate the problem perfectly!</p>
  
  <pre><code>import RxSwift
  
  class TestClass {
      private var disposeBag = DisposeBag()
  
      private var isCancelled = false
  
      init(cancelAfter: TimeInterval, longRunningTaskDuration: TimeInterval) {
          assert(Thread.isMainThread)
  
          load(longRunningTaskDuration: longRunningTaskDuration)
  
          DispatchQueue.main.asyncAfter(deadline: .now() + cancelAfter) { [weak self] in
              self?.cancel()
          }
      }
  
      private func load(longRunningTaskDuration: TimeInterval) {
          assert(Thread.isMainThread)
  
          // We set task not cancelled
          isCancelled = false
  
          DataService
              .shared
              .longRunngingTaskEmulation(sleepFor: longRunningTaskDuration)
              // We want long running task to be executed in background thread
              .subscribeOn(ConcurrentDispatchQueueScheduler.init(queue: .global()))
              // We want to process result in Main thread
              .observeOn(MainScheduler.instance)
              .subscribe(onSuccess: { [weak self] (result) in
                  assert(Thread.isMainThread)
  
                  guard let strongSelf = self else {
                      return
                  }
  
                  if !strongSelf.isCancelled {
                      print("Should not be called! Task is cancelled!")
                  } else {
                      // Do something with result, set image to UIImageView, for instance
                      // But if task was cancelled, this method will set invalid (old) data
                      print(result)
                  }
  
                  }, onError: nil)
              .disposed(by: disposeBag)
      }
  
      // Cancel all tasks. Can be called in PreapreForReuse.
      private func cancel() {
          assert(Thread.isMainThread)
  
          // For test purposes. After cancel, old task should not make any changes.
          isCancelled = true
  
          // Cancel all tasks by creating new DisposeBag (and disposing old)
          disposeBag = DisposeBag()
      }
  }
  
  class DataService {
      static let shared = DataService()
  
      private init() { }
  
      func longRunngingTaskEmulation(sleepFor: TimeInterval) -&gt; Single&lt;String&gt; {
          return Single
              .deferred {
                  assert(!Thread.isMainThread)
  
                  // Enulate long running task
                  Thread.sleep(forTimeInterval: sleepFor)
  
                  // Return dummy result for test purposes.
                  return .just("Success")
          }
      }
  }
  
  class MainClass {
      static let shared = MainClass()
  
      private init() { }
  
      func main() {
  
          Timer.scheduledTimer(withTimeInterval: 0.150, repeats: true) { [weak self] (_) in
              assert(Thread.isMainThread)
  
              let longRunningTaskDuration: TimeInterval = 0.050
  
              let offset = TimeInterval(arc4random_uniform(20)) / 1000.0
              let cancelAfter = 0.040 + offset
  
              self?.executeTest(cancelAfter: cancelAfter, longRunningTaskDuration: longRunningTaskDuration)
          }
      }
  
      var items: [TestClass] = []
      func executeTest(cancelAfter: TimeInterval, longRunningTaskDuration: TimeInterval) {
          let item = TestClass(cancelAfter: cancelAfter, longRunningTaskDuration: longRunningTaskDuration)
          items.append(item)
      }
  }
  </code></pre>
  
  <p>Call MainClass.shared.main() somewhere to start.</p>
  
  <p>We call method to load some data and later we call cancel (all from Main Thread). After cancel we sometimes receive the result (in main thread too), but it is old already.</p>
  
  <p>In real project TestClass is a UITableViewCell subclass and cancel method is called in prepareForReuse. Then cell is being reused and new data is set to the cell. And later we get the result of OLD task. And old image is set to the cell!</p>
  
  <hr>
  
  <p><strong>ORIGINAL QUESTION (OLD):</strong></p>
  
  <p>I would like to load image with RxSwift in iOS. I want to load image in background, and to use it in main thread. So I subscribeOn background thread, and observeOn main thread. And function will look like this:</p>
  
  <pre><code>func getImage(path: String) -&gt; Single&lt;UIImage&gt; {
      return Single
          .deferred {
              if let image = UIImage(contentsOfFile: path) {
                  return Single.just(image)
              } else {
                  return Single.error(SimpleError())
              }
          }
          .subscribeOn(ConcurrentDispatchQueueScheduler(qos: .background))
          .observeOn(MainScheduler.instance)
  }
  </code></pre>
  
  <p>But I get problems with cancelation. Because different schedulers are used to create items and to call dispose (disposing from main thread), subscription event can be raised after dispose is called. So in my case of using in UITableViewCell I receive invalid (old) image.</p>
  
  <p>If I create item (load image) in the same scheduler that observes (Main thread), everything works fine!
  But I would like to load images in background and I want it will be canceled after disposing (in prepareForReuse method or in new path set method). What is the common template for this?</p>
  
  <p><strong>EDIT:</strong></p>
  
  <p>I have created a test project, where I can emulate the problem when the event is received after dispose.</p>
  
  <p>And I have one simple solution that works. We should emit items in the same scheduler. So we should capture scheduler and emit items there (after long running task completes).</p>
  
  <pre><code>func getImage2(path: String) -&gt; Single&lt;UIImage&gt; {
      return Single
          .create(subscribe: { (single) -&gt; Disposable in
              // We captrure current queue to execute callback in
              // TODO: It can be nil if called from background thread
              let callbackQueue = OperationQueue.current
  
              // For async calculations
              OperationQueue().addOperation {
                  // Perform any long-running task
                  let image = UIImage(contentsOfFile: path)
  
                  // Emit item in captured queue
                  callbackQueue?.addOperation {
                      if let result = image {
                          single(.success(result))
                      } else {
                          single(.error(SimpleError()))
                      }
                  }
              }
  
              return Disposables.create()
          })
          .observeOn(MainScheduler.instance)
  }
  </code></pre>
  
  <p>But it is not in Rx way. And I think this is not the best solution.</p>
  
  <p>May be I should use CurrentThreadScheduler to emit items, but I cannot understand how. Is there any tutorial or example of items generation with schedulers usage? I did not find any.</p>