title: Not getting all expected events when subscribing to sequence with TestScheduler
body: <p>I'm trying to write an integration test for a <code>Reactor</code> in an app built with ReactorKit and Realm/RxRealm.</p>
  <p>I'm having trouble using <code>TestScheduler</code> to simulate user actions and test the expected emitted states.</p>
  <p>In a nutshell, my problem is this: I'm binding an action that will make my Reactor save an item to Realm, my Reactor also observes changes to this object in Realm, and I expect my Reactor to emit the new state of this item observed from Realm.</p>
  <p>What I'm seeing is that my test does not get the emission of the newly saved object in time to assert its value, it's emitted after my test assertion runs.</p>
  <p>There is a fair amount of code involved, but attempting to whittle it down into a self-contained example of what it all roughly looks like below:</p>
  <pre class="lang-swift prettyprint-override"><code>struct MyObject {
      var counter: Int = 0
  }
  
  class MyReactor: Reactor {
      enum Action {
          case load
          case mutateState
      }
  
      enum Mutation {
          case setObject(MyObject)
      }
  
      struct State {
          var object: MyObject?
      }
      
      func mutate(action: Action) -&gt; Observable&lt;Mutation&gt; {
          switch action {
          case .load:
              return service.monitorObject().map(Mutation.setObject)
          case .mutateState:
              guard var myObject = currentState.object else { return .empty() }
              myObject.counter += 1
              return service.save(myObject).andThen(.empty())
          }
      }
  
      func reduce(state: State, mutation: Mutation) -&gt; Observable&lt;State&gt; {
          var newState = state
          switch mutation {
          case let .setObject(object):
              // Should be called twice in tests, once on load, once after mutateState action
              newState.object = object
          }
      }
  }
  
  struct Service {
  
      // There is always at least one default instance of `MyObject` in Realm.
      func monitorObject() -&gt; Observable&lt;MyObject&gt; {
          return Observable
              .collection(from: realm.objects(MyObject.self))
              .map { $0.first! }
      }
  
      func save(_ object: MyObject) -&gt; Completable {
          return Completable.create { emitter in
              try! realm.write {
                  realm.add(object, update: .modified)
              }
              emitter(.completed)
              return Disposables.create()
          }
      }
  }
  
  class MyTest: QuickSpec {
  
      var scheduler: TestScheduler!
      var sut: MyReactor!
      var disposeBag: DisposeBag!
      var service: Service!
      var config: Realm.Configuration!
  
      override func spec() {
          beforeEach {
              config = Realm.Configuration(inMemoryIdentifier: UUID().uuidString)
              scheduler = TestScheduler(initialClock: 0)
              disposeBag = DisposeBag()
              sut = MyReactor()
              service = Service(realmConfig: config)
          }
  
          describe(&quot;when my reactor gets a mutateState action&quot;) {
              it(&quot;should mutate state&quot;) {
  
                  scheduler.createHotObservable([
                      .next(1, Action.load),
                      .next(2, Action.mutateState),
                  ])
                  .bind(to: sut.action)
                  .disposed(by: disposeBag)
  
                  let response = scheduler.start(created: 0, subscribed: 0, disposed: 1000) {
                      sut.state.map(\.object)
                  }
  
                  // Counter always equals 0
                  XCTAssertTrue(response.events.last!.value.element!!.counter == 1)
              }
          }
      }
  }
  </code></pre>
  <p>What I'm expecting to happen is my Reactor's state is set for a 2nd time, before the <code>XCTAssertTrue</code> is hit. What is actually happening is the assert is hit with the initially loaded state, and then, my reactor's state is set again.</p>
  <p>I thought my problem might be related to schedulers. Something I tried was injecting the test scheduler into my Service and doing <code>observeOn(testScheduler)</code> on my <code>monitorObject</code> function. But I'm still observing the assert get hit before the reactor's state is set for the 2nd time. I'm also not sure if a nuance of RxRealm/Realm change set notifications is the cause - not sure how to verify whether that might be the case.</p>
  <p>Hopefully the problem and question is clear. Thanks in advance for any help.</p>