rxswift_performance_test.swift

import RxSwift
import XCTest

 /// Measures the average time needed to create a `BehaviorSubject`, chain three `do` blocks on
 /// it and get into the last `subscribe` block.
func testRxSubscribeOnSerialQueue() {
  // Arrange.
  let expectation = self.expectation(description: "")
  expectation.expectedFulfillmentCount = Constants.iterationCount
  let queue = DispatchQueue(label: #function, qos: .userInitiated)
  let semaphore = DispatchSemaphore(value: 0)
  let scheduler = SerialDispatchQueueScheduler(queue: queue, internalSerialQueueName: #function)
  let disposeBag = DisposeBag()

  // Act.
  DispatchQueue.main.async {
    let time = dispatch_benchmark(Constants.iterationCount) {
      BehaviorSubject(value: true).observeOn(scheduler).do(onNext: { _ in }).do(onNext: { _ in
      }).subscribe(onNext: { _ in
        semaphore.signal()
        expectation.fulfill()
      }).disposed(by: disposeBag)
      semaphore.wait()
    }
    print(average: time)
  }

  // Assert.
  waitForExpectations(timeout: 10)
}

/// Measures the total time needed to observe a lot of `PublishSubject` with chained `subscribe`
/// blocks on them on a concurrent queue and wait for each of them to get into chained block.
func testRxSubscribeOnConcurrentQueue() {
  // Arrange.
  let queue = DispatchQueue(label: #function, qos: .userInitiated, attributes: .concurrent)
  let group = DispatchGroup()
  let scheduler = ConcurrentDispatchQueueScheduler(queue: queue)
  let disposeBag = DisposeBag()
  var subjects = [PublishSubject<Bool>]()
  for _ in 0..<Constants.iterationCount {
    group.enter()
    let subject = PublishSubject<Bool>()
    subject.observeOn(scheduler).subscribe(onNext: { _ in
      group.leave()
    }).disposed(by: disposeBag)
    subjects.append(subject)
  }
  let startDate = Date()

  // Act.
  for subject in subjects {
    subject.onNext(true)
  }

  // Assert.
  XCTAssert(group.wait(timeout: .now() + 1) == .success)
  let endDate = Date()
  print(total: endDate.timeIntervalSince(startDate))
}