Scheduling operators

• Combine introduces the Scheduler protocol
• ... adopted by DispatchQueue, RunLoop and others
• ... lets you determine the execution context for subscription and value delivery

本小节演示了线程调用，receive(on:) 能够改变其下游数据的接收线程， subscribe(on:)跟 receive(on:) 刚好相反，它影响上游调用的线程。

let firstStepDone = DispatchSemaphore(value: 0)

/*:
## `receive(on:)`
- determines on which scheduler values will be received by the next operator and then on
- used with a `DispatchQueue`, lets you control on which queue values are being delivered
*/
print("* Demonstrating receive(on:)")

let publisher = PassthroughSubject<String, Never>()
let receivingQueue = DispatchQueue(label: "receiving-queue")
let subscription = publisher
	.receive(on: receivingQueue)
	.sink { value in
		print("Received value: \(value) on thread \(Thread.current)")
		if value == "Four" {
			firstStepDone.signal()
		}
}

for string in ["One","Two","Three","Four"] {
	DispatchQueue.global().async {
		publisher.send(string)
	}
}

firstStepDone.wait()

/*:
## `subscribe(on:)`
- determines on which scheduler the subscription occurs
- useful to control on which scheduler the work _starts_
- may or may not impact the queue on which values are delivered
*/
print("\n* Demonstrating subscribe(on:)")
let subscription2 = [1,2,3,4,5].publisher
	.subscribe(on: DispatchQueue.global())
	.handleEvents(receiveOutput: { value in
		print("Value \(value) emitted on thread \(Thread.current)")
	})
	.receive(on: receivingQueue)
	.sink { value in
		print("Received value: \(value) on thread \(Thread.current)")
}

receive(on:) 比较常用，例如在子线程发起网络请求，当数据回来后，在主线程来刷新UI。

/*:
 写法一 在回调中指定主线程 DispatchQueue.main.async
 */
publisher
    .sink(receiveCompletion: {
        DispatchQueue.main.async {
            /// 刷新UI
        }
    }, receiveValue: {
        print($0)
    })

/*:
 写法二 使用`receive(on:)`指定主线程接收数据
 */
publisher
    .receive(on: RunLoop.main)
    .sink(receiveCompletion: {
        /// 刷新UI
    }, receiveValue: {
        print($0)
    })

subscribe(on:) 没有典型的使用场景，故而不谈