[iOS/Swift] Combine 알아보기 viii

Combine 스케줄러 이해하기

이번 장에선 Scheduler 프로토콜과 주로 사용하는 연산자인 receive(on:)과 subscriber(on:)을 알아본다.

Scheduler 프로토콜 탐구하기

Combine은 여러 비동기 작업을 수행한다.

• 그러므로 Combine은 스레드를 블로킹하지 않고 작업을 스케줄하고 실행하는 좋은 방법을 가져야 한다.
• 이를 위해 Combine은 Combine의 publishers가 작업을 수행할 수 있는 공간인 Scheduler라는 프로토콜을 정의했다.

대표적인 Scheduler 프로토콜

• RunLoop
• DispatchQueue
  • OS_dispatch_queue_main(DispatchQueue.main)
  • OS_dispatch_queue_global(DispatchQueue.global())
• OperationQueue

Combinedms 자신의 스케줄러인 ImmediateScheduler도 제공한다.

• ImmediateScheduler에 작업이 들어오면 해당 작업은 바로 실행된다.
• delay된 작업을 ImmediateScheduler에 스케줄해도 delay 되지 않고 바로 실행된다.

Scheduler가 사용된 예시

$searchQuery
	.debounce(for: 0.3, scheduler: DispatchQueue.main)
	.assign(to: \\.text, on: label)
	.store(in: &cancellables)

Combine needs a scheduler to set up some timed work to make the debouncing work (이건 번역을 어케 해야 될지 모르겠다)

• 타이머와 delayed task(ex/ debounce)는 본질적으로 스케줄러와 묶여있기 때문에 debounce 연산자에 scheduler를 전달하지 않고 debouncer를 스케줄할 수 없다.

DispatchQueue를 사용해도 되지만 main Queue에 여유가 없다면 타이머는 중단된다.

• ex/ 일반적인 타이머를 실행시키는 도중에 테이블 뷰를 스크롤 하면 스크롤이 끝날 때까지 타이머가 중단될 것이다.

→ 다른 스케줄러를 사용하는 것이 적합해 보인다.

$searchQuery
	.debounce(for: 0.3, scheduler: DispatchQueue.global())
	.assign(to: \\.text, on: label)
	.store(in: &cancellables)

Scheduler가 프로토콜이기 때문에 손쉽게 DispatchQueue.global()로 변경할 수 있다.

• OperaptionQueue, RunLoop 모두 가능
• 어떤 것을 골라도 궁극적으로 동일하게 동작해야 한다.

OperationQueue와 Scheduler

scheduler에 OperationQueue 인스턴스를 사용하는 것은 조금 복잡하다.

Combine에서 모든 스케줄러는 serial queue에서 운영되는 것을 기대한다.

• 병렬로 진행되는 것이 아닌 작업을 하나씩 진행한다.
• Combine은 이를 강제하지 않으나 병렬 처리로 인해 문제가 발생하는 경우가 더러 있다.

Receive(on:)

receive(on:)이 사용된 예시는 아래와 같다.

var cancellables = Set<AnyCancellable>()

UNUserNotificationCenter.current().getNotificationSettings()
    .flatMap({ settings -> AnyPublisher<Bool, Never> in
        switch settings.authorizationStatus {
        case .notDetermined:
            return UNUserNotificationCenter.current().requestAuthorization(options: [.alert, .sound, .badge])
                .replaceError(with: false)
                .eraseToAnyPublisher()
        case .denied:
            return Just(false).eraseToAnyPublisher()
        default:
            return Just(true).eraseToAnyPublisher()
        }
    })
    .receive(on: DispatchQueue.main)
    .sink(receiveValue: { hasPermissions in
        if hasPermissions == false {
            // user setting
        } else {
            // permission granted
        }

    })
    .store(in: &cancellables)

receive(on:)을 사용하면 receive(on:)이 사용된 downstream부터 receive(on:)으로부터 지정된 스케줄러에서 실행된다.

위의 예시는 sink가 main 스레드에서 동작하는 것을 확실히 하기 위해 receive(on: DispatchQueue.main)을 적용했다.

Combine은 일이 시작된 지점을 기본 스케줄러로 사용한다.

• 기본 스케줄러는 값이 생성된 스레드에서 값을 downstream에게 방출한다.

import PlaygroundSupport

PlaygroundPage.current.needsIndefiniteExecution = true

var cancellables = Set<AnyCancellable>()

let intSubject = PassthroughSubject<Int, Never>()

intSubject
    .sink(receiveValue: { value in
        print(value)
        print(Thread.current)
    })
		.store(in: &cancellables)

intSubject.send(1)

DispatchQueue.global().async {
    intSubject.send(2)
}

let queue = OperationQueue()
queue.maxConcurrentOperationCount = 1
queue.underlyingQueue = DispatchQueue(label: "com.Combine.queue")

queue.addOperation {
    intSubject.send(3)
}

/* Output
1
<_NSMainThread: 0x600001b881c0>{number = 1, name = main}
2
<NSThread: 0x600001b81900>{number = 7, name = (null)}
3
<NSThread: 0x600001b94640>{number = 9, name = (null)}
*/

위의 예시를 보면 intSubject.send가 실행된 스레드에서 동작하는 것을 알 수 있다.

• 아무런 조치 없이 PassthroughSubject를 사용해 UI를 그리는 것은 문제로 이어질 수 있다.
• receive(on:)을 사용해 해결할 수 있다.

import PlaygroundSupport

PlaygroundPage.current.needsIndefiniteExecution = true

var cancellables = Set<AnyCancellable>()

let intSubject = PassthroughSubject<Int, Never>()

intSubject
    .receive(on: DispatchQueue.main)
    .sink(receiveValue: { value in
        print(value)
        print(Thread.current)
    })
		.store(in: &cancellables)

intSubject.send(1)

DispatchQueue.global().async {
    intSubject.send(2)
}

let queue = OperationQueue()
queue.maxConcurrentOperationCount = 1
queue.underlyingQueue = DispatchQueue(label: "com.Combine.queue")

queue.addOperation {
    intSubject.send(3)
}

/*
1
<_NSMainThread: 0x600000cfc1c0>{number = 1, name = main}
2
<_NSMainThread: 0x600000cfc1c0>{number = 1, name = main}
3
<_NSMainThread: 0x600000cfc1c0>{number = 1, name = main}
*/

receive(on:) 연산자는 sink 바로 앞에 쓰거나 어떤 스레드에서 동작하길 확신할 때만 사용하는 것이 좋다.

subscribe(on:)

publisher 스트림의 upstream에 적용된다.

• publisher와 operator의 전체적인 체인에 영향을 미친다.

subscribe(on:)을 적용했지만 의도한대로 동작하지 않을 수도 있다.

• publisher가 어떤 queue에 값을 방출할 것인지 결정할 수 있다.

URLSession.shared.dataTaskPublisher(for: URL(string: "https://...")
	.subscribe(on: DispatchQueue.main)
	.map({ result in
		print(Thread.current.isMainThread)
	})
	.sink(
		receiveCompletion: { _ in },
		receiveValue: { value in
			print(Thread.current.isMainThread)
		})
	

subscribe를 사용해 전체적인 체인이 main thread에서 돌아가게끔 했지만 결과를 보면 main thread에서 호출되지 않음을 알 수 있다.