[Combine] Subject(PassthroughSubject, CurrentValueSubject) 정리

Subject는 non-Combine 코드로부터 값을 받아서 Subscriber에게 값을 방출시키는 Publisher입니다.

PassthroughtSubject로 value를 전달하는 방식은 명령형 코드를 Combine을 통한 선언형 코드로 연결하는 좋은 방법입니다.

(@published를 마킹한 프로퍼티와 유사한 징검다리 역할입니다. non-Combine 코드로부터 값을 받아, sink를 통해 값을 내보낼 수 있죠)

 

PassthroughSubject

( = RxSwift의 PublishSubject)

Subject의 가장 기본형입니다.

아래는 우선 커스텀 Subscriber의 예시입니다(Custom Subscriber는 이전글 참조)

// 1. Error 타입을 정의
enum MyError: Error {
    case test
}

// 2. String, MyError를 받는 Custom Subscriber를 정의
final class StringSubscriber: Subscriber {
    typealias Input = String
    typealias Failure = MyError
    
    func receive(subscription: Subscription) {
        subscription.request(.max(2))
    }
    
    func receive(_ input: String) -> Subscribers.Demand {
        print("Received value", input)
        // 3. 받은 값을 기준으로 demand를 조절함
        return input == "World" ? .max(1) : .none
    }
    
    func receive(completion: Subscribers.Completion<MyError>) {
        print("Received completion", completion)
    }
}


// 4. Custom Subscriber의 인스턴스를 생성
let subscriber = StringSubscriber()

String, MyError를 받는 StringSubscriber라는 커스텀 subscriber입니다.
subscribe시 최대 2개값을 받고, "world"라는 값을 받으면 최대값 2개에서 +1씩 더 받습니다.

이어서 아래에 PassthroughtSubject를 생성하고, Subscription을 StringSubscriber와 sink를 통해 만들어봅니다.
당연하게도 PassthroughtSubject는 <String, MyError>타입으로 정의되어야 StringSubscriber가 받을 수 있습니다.

 

enum MyError: Error {
    case test
}

final class StringSubscriber: Subscriber {
    typealias Input = String
    typealias Failure = MyError
    
    func receive(subscription: Subscription) {
        subscription.request(.max(2))
    }
    
    func receive(_ input: String) -> Subscribers.Demand {
        print("Received value", input)
        return input == "World" ? .max(1) : .none
    }
    
    func receive(completion: Subscribers.Completion<MyError>) {
        print("Received completion", completion)
    }
}

let subscriber = StringSubscriber()

// 추가
// 5. String과 MyError가 정의된 타입의 PassthroughSubject인스턴스를 생성
let subject = PassthroughSubject<String, MyError>()

// 6. subscriber를 subject에 subscribe등록
subject.subscribe(subscriber)

// 7. sink로 또다른 subscription을 생성
let subscription = subject
    .sink(
        receiveCompletion: { completion in
            print("Received completion (sink)", completion)
        },
        receiveValue: { value in
            print("Received value (sink)", value)
        }
    )

 

subscription들도 설정되었으니, subject에 값을 넣고 방출시키기 위한 코드를 작성합니다.

subject로 값은 .send()를 통해 전달합니다.

subject.send("Hello")
subject.send("World")

각 subscriber가 publish될 때마다 value를 받아서 출력합니다.

// 출력
Received value Hello
Received value (sink) Hello
Received value World
Received value (sink) World

 

아래에 subscription을 취소시키는 코드를 작성하고 한번 더 값을 전달해봅니다.

subject.send("Hello")
subject.send("World")

// 추가
// 8. 두번째 구독을 cancel
subscription.cancel()

// 9. 새로운 값을 내보냄
subject.send("Still there?")

첫번째 subscriber만 값을 받아 출력되는 것을 확인할 수 있습니다.

Received value (sink) Hello
Received value Hello
Received value (sink) World
Received value World
Received value Still there?

 

이어서 subject로 complete 이벤트를 전달 후 값을 다시 전달해봅니다.

subject.send("Hello")
subject.send("World")

subscription.cancel()

subject.send("Still there?")

// (추가)
// finish 이벤트
subject.send(completion: .finished)
subject.send("How about another one?")

두번째 subscription은 이전에 cancel되었기 때문에 completion을 받을 수 없고,

첫번째 subscriber는 "How about another one?" 메세지를 받기전에 complete 이벤트를 받았기 때문에 값을 받지 못합니다.

Received value (sink) Hello
Received value Hello
Received value (sink) World
Received value World
Received value Still there?
Received completion finished

 

마지막으로 subject.send(completion: .finished) 바로 윗줄에 failure 이벤트를 전달하는 코드를 작성해봅니다.

subject.send("Hello")
subject.send("World")

subscription.cancel()

subject.send("Still there?")

subject.send(completion: .failure(MyError.test)) // 추가

subject.send(completion: .finished)
subject.send("How about another one?")

Publisher가 failure로 error를 방출하여 Complete되고, 더이상 값을 내보내지 않는 것을 첫번째 subscriber를 통해 확인할 수 있습니다.

Received value (sink) Hello
Received value Hello
Received value (sink) World
Received value World
Received value Still there?
Received completion failure(__lldb_expr_47.MyError.test)

 

💡정리
PassthroughtSubject는 Subject의 기본형으로,
non-Combine 코드로부터 값을 받아서 Subscriber에게 값을 방출시키는 Publisher.

 

---

CurrentValueSubject

( = RxSwift의 BehaviorSubject)
때에 따라 명령형 코드에서 Publisher의 현재(마지막) value를 확인해야 할 필요가 생기는데,
이런 경우를 위해 CurrentValueSubject가 있습니다.
용법은 PassthroughtSubject와 동일하지만, 초기화시 초기값이 필요하다는 차이점이 있습니다.

아래는 CurrentValueSubject의 예시입니다.

// 1. subscription을 저장할 Set 생성
var subscriptions = Set<AnyCancellable>()

// 2. Int, Never 타입의 CurrentValueSubject 생성. 초기값은 0
let subject = CurrentValueSubject<Int, Never>(0)

// 3. subscription 생성. subject에게 값을 받아 print
subject
    .sink(receiveValue: { print($0) })
    .store(in: &subscriptions) // 4. subscriptions에 store

출력을 보면? subscribe하자마자 초기값을 받아오는 것이 확인됩니다.

// 출력
0

 

PassthroughSubject처럼 .send()를 통해 값을 전달할 수 있습니다.

아래에 코드를 추가합니다.

var subscriptions = Set<AnyCancellable>()

let subject = CurrentValueSubject<Int, Never>(0)

subject
    .sink(receiveValue: { print($0) })
    .store(in: &subscriptions)

// 추가
subject.send(1)
subject.send(2)

출력.

// 출력
0
1
2

 

현재(마지막) 값을 받아오는 코드를 아래에 추가합니다.

.value를 통해서 받아올 수 있습니다.

var subscriptions = Set<AnyCancellable>()

let subject = CurrentValueSubject<Int, Never>(0)

subject
    .sink(receiveValue: { print($0) })
    .store(in: &subscriptions)

subject.send(1)
subject.send(2)

// 추가
print("value: ", subject.value)

출력.

// 출력
0
1
2
value:  2

 

value에 직접 값을 전달해보는 코드를 아래에 추가해봅니다.

var subscriptions = Set<AnyCancellable>()

let subject = CurrentValueSubject<Int, Never>(0)

subject
    .sink(receiveValue: { print($0) })
    .store(in: &subscriptions)

subject.send(1)
subject.send(2)

print("value: ", subject.value)

// 추가
subject.value = 3
print("value: ", subject.value)

출력을 보면, value로 값을 변경시켜도 .send()로 전달하는 것과 동일한 효과인 것을 알 수 있습니다.

0
1
2
value:  2
3
value:  3

 

이어서 subscription을 추가해보겠습니다. 두번째 subscription에는 sink 위에 .print()가 추가되었습니다.

그리고 첫번째 subscription에도 .print()를 추가했습니다.

.print()를 사용할 땐 prefix값을 넣어서 구분시켜줄 수 있습니다.

var subscriptions = Set<AnyCancellable>()

let subject = CurrentValueSubject<Int, Never>(0)

subject
    .print("[1st subscription]") // 추가
    .sink(receiveValue: { print($0) })
    .store(in: &subscriptions)

subject.send(1)
subject.send(2)

print("value: ", subject.value)

subject.value = 3
print("value: ", subject.value)

// 추가
subject
    .print("[2nd subscription]")
    .sink(receiveValue: { print("2nd subscription:", $0) })
    .store(in: &subscriptions)

 

출력이 조금 복잡해지겠지만 실행시켜보면,

subscription의 변경이 생길 때마다 상세하게 볼 수 있는 print가 되는 것을 볼 수 있습니다.

[1st subscription]: receive subscription: (CurrentValueSubject)
[1st subscription]: request unlimited
[1st subscription]: receive value: (0)
0
[1st subscription]: receive value: (1)
1
[1st subscription]: receive value: (2)
2
value:  2
[1st subscription]: receive value: (3)
3
value:  3
[2nd subscription]: receive subscription: (CurrentValueSubject)
[2nd subscription]: request unlimited
[2nd subscription]: receive value: (3)
2nd subscription: 3

 

마지막으로 complete이벤트를 전달하면

// 추가
subject.send(completion: .finished)

두 subscription이 모두 종료를 전달받습니다.

[2nd subscription]: receive finished
[1st subscription]: receive finished

 

💡정리
CurrentValueSubject는 PassthroughSubject의 변형으로,
.value 프로퍼티를 통해 현재(마지막) 방출된 값을 받아올 수 있음.
이런 특성이 있기 때문에 초기화 시에 초기값을 넣어주어야 함.
.value 프로퍼티로 직접 값을 넣어도 .send() 함수로 값을 전달하는 것과 동일한 효과(subscriber들에게 value를 방출).

💡추가
Publisher나 Subject를 sink하는 구문 전에 .print("prefix")를 넣어주면, subscription의 변경이 있을 때마다 출력을 통해 정보를 알려줌.