[Combine] Map Operator 로직 딥다이브 (ft. OpenCombine)

 

Combine, Functional Reactive Programming의 꽃이라고도 할 수 있는 Operator의 로직이 궁금해져 이를 딥다이브해보고자 합니다.

Operator의 종류가 너무 많고 모든 Operator의 동작이 다 달라서 모든 것을 다룰 수는 없고 대표적으로 Map Operator를 가지고 Operator의 동작을 엿보고자 합니다. 

 

Operator에도 결국 중요한 것은 upstream, downstream그리고 Operator를 거쳐서 바뀌는 value들이겠죠.

한번 집중해서 어떤 흐름으로 굴러가는지 살펴봅시다. 이번에도 OpenCombine을 통해서 알아보겠습니다.

https://github.com/OpenCombine/OpenCombine

GitHub - OpenCombine/OpenCombine: Open source implementation of Apple's Combine framework for processing values over time.

 

이전 글을 보고 오신다면 이해가 더 수월합니다 !

[Combine] Publisher와 Subscriber간 연결 과정 딥다이브 (ft. OpenCombine)

 

결론을 미리 공유한다면 아래 그림과 같습니다.

 

그림의 과정을 정리하기까지의 과정이 글의 주된 내용입니다. 그림을 보고도 이해가 잘가신다면 아래 내용은 볼 필요 없..지 않을까요..? 아마도 ?

---

예제 코드

이번에는 저번 글에서 map이 추가된 형태의 코드를 가지고 살펴보겠습니다.

0. PassthorughSubject.map()

extension Publisher

Publisher의 extension으로 제공되는 map메서드는 사실 Publishers.Map이라는 구조체를 리턴해주는 메서드입니다.

 

구조체 생성시에 self, 와 transform이라는 @escaping closure를 넘겨주고 있는데 여기서 self는 Publisher의 extension이기 때문에 map()을 호출한 Publihser가 곧 self에 대응됩니다.

 

예제코드상으로는 PassthroughSubject가 self가 됩니다.

@escaping closure인 transform은 바로 예제 코드에서 map 뒤에 쓴 후행 클로저에 대응됩니다.

• Publisher의 extension에 있는 map()메서드를 호출하게 되면, Publishers.Map이라는 구조체를 생성해서 반환하는데
• 이 때 "map 메서드를 호출한 Publisher(PassthroughSubject) self"와 "값을 변환하고자 하는 작업의 의미를 가진 @escaping closure, transform"를 가지고 Publishers.Map이라는 구조체를 만들고 있구나~

 

넘겨받은 self는 upstream 변수로, transform 클로저는 transform이라는 변수에 할당하는 것을 확인할 수 있습니다.

 

이제 그 인자들의 의미를 조금 보기 위해 Map의 type placeholder부분을 보자면 upstream과 Output이 존재합니다.

 

upstream을 Self(self가 PassthroughSubject 객체에 해당하고 Self는 이것의 메타타입), 그리고 Output은 transform의 return 타입이 Output으로 들어가는 것을 볼 수 있습니다.

 

즉, Publisher protocol을 채택하는 Publishers.Map이라는 구조체는 upstream을 map메서드를 호출한 Publisher로 두고, "값을 변환하는 작업"을 가지고 있구나 !

 

정도로 해석할 수 있을 것 같습니다. 

 

Publisher를 반환하고 있으니 operator chaining이 가능했던 이유도 이제 설명이 되네요😃

1. Map.subscribe(Sink)

이전글에서 이야기했지만 그래도 이야기 안하고 가긴 서운하니까 살짝 이야기하면 sink메서드를 호출하게 되면 내부에서 Sink Subscriber를 만들어서 구독시켜주기 때문에 우리가 별도의 subscriber를 만들고 구독하지 않아도 된다는 것을 알 수 있었습니다.

 

subscribe메서드 또한 Publisher의 extension이기 때문에 여기서 subscribe메서드를 호출하는 주체는 Publisher, 예제 코드 상으로는 Map Publisher가 됩니다.

 

이제 subscribe메서드의 구현부를 살펴보러 가봅시다.

 

1-1. Map.receive(Sink)

subscribe메서드에는 무조건 실행되는 메서드가 존재하는데 subscriber를 accept하는 과정의 시작을 알리는 receive(subscriber) 메서드입니다. 흰색 줄로 쳐진 subscriber는 위에서 넘겨받은 Sink Subscriber입니다.

 

여전히 Publisher주체는 Map Publihser이기 때문에 다음으로 살펴봐야할 메서드는 Map에 있는 receive(subscriber:)의 구현부입니다.

2. PassthroughSubject.subscribe(Inner)

 

처음 Map 구조체를 생성할 때 upstream변수에 "map 메서드를 호출한 Publisher(upstream)"를 저장했었던 것을 기억하시나요 ?

 

해당 upstream에. "자신(Map)이 받은 Subscriber(Sink)를 가지고 새로 생성한 Inner 객체"를 구독시켜주고 있습니다.

 

여기서 Inner객체를 한번 살펴보겠습니다.

2-1. Inner: Subscriber

Inner

Inner는 여기서 downstream이라는 네이밍으로 Subscriber를 가지고 있고, 값을 변경하는 작업을 의미하는 클로저를 map이라는 변수에 저장하고 있습니다. 그리고 Inner또한 Subscriber protocol을 채택하고 있기 때문에 Subscriber입니다.

 

Map 구조체에서도 분명 transform이라는 이름으로 변경에 대한 클로저를 가지고 있었지만, 그 목적은 그저 Inner객체를 생성할 때 넣어주기 위함이었고, 실제로 값을 바꿔주는 작업을 수행하는 객체는 Inner객체입니다. 이 부분은 뒤에서 다시 한번 살펴보도록 합시다 !

 

결국 Map을 구독하면 Inner객체를 생성하고, Inner객체를 자신의 upstream에 구독시키는 것을 볼 수 있습니다.

3. 다시 Subscribe: PassthroughSubject.receive(Inner)

 

조금 전의 subscribe과정과 완전히 동일하지만 이번에는 넘겨받는 subscriber가 Inner Subscriber라는 점이 차이점입니다.

 

이전 과정과 동일하게 receive의 메서드의 구현부를 살펴봅시다.

4-1. Inner.receive(subscription:) 호출

 

subscriber를 accept하는 과정이 시작되었으니, 제일 먼저 구독권(subscription)을 만들어야합니다. 

이 때 초록박스를 보시면 conduit을 만들 때 downstream으로 넣는 subscriber는 여기서는 Sink가 아닌 Inner입니다.

(초록박스 유의 ! 이 부분은 뒤에 또 한번 언급됩니다..!)

 

왜냐하면 위에서 Map이 멋대로 Sink Subscriber를 담은 Inner객체를 만들어서 자신의 upstream에 구독시켰기 때문에, 이 때 구독요청을 보낸 Subscriber는 Sink가 아닌 Inner Subscriber입니다.

 

이 후 생성한 구독권을 Inner에게 전달하고 있습니다.

 

다음으로 봐야할 메서드는 Inner에 있는 receive(subscription:) 메서드입니다 !

4-2. Sink.receive(subscription:)

 

Map Publisher에게 구독요청을 보낸 Subscriber을 이용해서 Inner를 생성하는데, 그 때 해당 Subscriber를 Inner의 downstream이라는 변수에 담았었습니다. 예제코드상으로는 Inner의 downstream변수에 들어있는 Subscriber는 Sink가 될 것 입니다.

 

Inner가 subscription을 받으면 하는 일은, 그저 자신이 받은 subscription을 그냥 downstream, Sink에게 넘기고 있습니다 😂

 

즉, 구독 요청을 한 사람은 Inner인데, 애써 받은 구독권은 그냥 자신이 가지고 있는 downstream, Subscriber에게 넘겨주고 있는 것 입니다.

제가 생각하는 그 이유는 뒤에서 언급하겠습니다 !

 

다음으로 살펴볼 메서드는 Sink가 구독권을 받는 것이므로, Sink의 receive(subscription:) 메서드입니다.

4-3. Conduit.request(.unlimited)

 

Sink Subscriber는 구독권을 받으면 자신의 status를 subscribed로 바꾸고 연관값으로 구독권을 가지고 있습니다.

이제 마지막으로 얼만큼 값을 받을지에 대한 요청을 subscription에게 하고 있습니다.

 

4-4. Conduit의 demand값 증가

여기서 demand의 unlimited를 보시면 Demand를 UInt.max로 생성하는 것과 같은 같은 효과를 가집니다.

Unsigned Integar의 max면 2^32 = 4,294,967,296(약 43억)에 가깝기 때문에 무제한이라고 할 수 있겠네요.

위에서 구한 거의 무제한에 가까운 수를 demand에 더해줍니다.

 

여기까지 오면 Subject와 Sink간의 data stream이 연결된 것 입니다 ! 🥳

(하지만 이 말이 제 착각을 불러일으킵니다..🥲)

값을 보내는 상황 가정

subject.send(1)

 

5 ~ 6. Conduit.offer(input)

Subject가 가지고 있는 subscription list를 순회하면서 conduit에 offer메서드를 호출하고 있습니다.

 

여기서 Subject와 연결된 것이 Sink이기에 (구독권을 가지고 있다는 이유로) Sink로 바로 값을 보내주고 있구나 라고 생각했습니다.

 

흠..뭔가 이상합니다.. 🤔

 

어라..? downstream이 Sink면 Subject에서 값을 바로 Sink로 넘겨주면 map 과정은 어디서 하는거지..?

 

 

아까 조금전에 다시 언급하기로 했던 메서드를 다시 가져와봤습니다.

 

분명 구독권은 Inner객체가 받자마자 Sink Subscriber에게 넘겨주었지만, Conduit자체를 만들 때에 downstream은 Inner입니다..!

 

그리고 Publisher(Subject)가 값을 보낼 때는 conduit만 바라보고 값을 보내고 있습니다.

 

이 Conduit을 만들 때 넣어준 downstream subscriber가 곧 Inner이기 때문에 !

 

subject는 값이 들어오면 Inner로 보내고

 

 

값을 받은 Inner가 값을 변환시켜서 Sink에게 보내고 있던 것이었습니다.

 

그러면 어차피 값도 본인(Inner)가 받을텐데 subscription은 뭣하러 downstream인 Sink에게 넘길까 ? 🤔

+ 왜 Inner는 subscription을 받아서 저장하지 않고 바로 downstream으로 보낼까?

제 추측이지만, Map에서 만드는 Inner는 일회용입니다. 한번 쓰고 사라지는 객체입니다.

 

근데 Inner가 subscription을 가진 채로 할당 해제된다면 Subject와 Sink간의 연결고리가 끊어지기 때문에 subscription은 Sink에게 양도하는 것이 아닐까 싶습니다.

Wrap up

지금까지의 과정을 정리하면 위의 그림과 같습니다. 최대한 직관적으로 정리했는데 이해가 잘가면 좋겠습니다..!

 

0번부터 숫자를 따라가면서 보시면 됩니다. 혹시 이해가 안가시는 부분이 있다면 위에서 해당 번호로 시작되는 제목으로 찾아가셔서 다시 보셔도 좋습니다 !

---

마무리

operator에서 upstream과 downstream이 어떻게 유지되는지를 보기 위해서 Operator중 map을 대표로 살펴보았습니다.

아마 다른 Operator들을 다 이렇게까지 보긴 어렵기 때문에, Operator 딥다이브는 더 없을 것 같습니다.

사실 이거 이외에도 다른 딥다이브할 것들도 많고 아직 써야할 것들도 많아서...ㅎㅎ

 

틀린부분이나 궁금한 부분 있으면 편하게 댓글 남겨주시면 감사하겠습니다 !