2. Publishers & Subscribers

Hello Publisher

publisher는 두가지 이벤트를 방출함

1. 값 (as element)
2. completion event

---

Hello Subscriber

Subscribing with sink(_: _:)

RxSwift의 subscribe(onNext: ) 와 동일

// Code
Just(["myung", "sub"])
    .sink(receiveCompletion: {
        print("오늘 스터디는?? \\($0)")
    }, receiveValue: {
        print("스터디 참가자: \\($0)")
    })

// Result
스터디 참가자: ["myung", "sub"]
오늘 스터디는?? finished

Subscribing with assign(to:on:), assign(to: )

KVO 방식으로 값을 할당

// 사람 객체
class Man {
    @Published var name: String = ""
    
    var didSetName: String = "" {
        didSet {
            print(didSetName)
        }
    }
}

========================================================================
// assign(to: on: )
let myung_nameTag = ["myung"].publisher
let myung = Man()

myung_nameTag
    .assign(to: \\.didSetName, on: ken)

// Result
[didSet] 새로운 제 이름은: myung

========================================================================
// assign(to: )
let sub_nameTag = ["sub"].publisher
let sub = Man()

sub.$name
    .filter { $0 != "" }
    .sink(receiveValue: {
        print("[Published & Sink] 새로운 제 이름은: \\($0)")
    })
sub_nameTag
    .assign(to: &sub.$name)

// Result
[Published & Sink] 새로운 제 이름은: sub

---

Hello Cancellable

RxSwift의 Dispose, DisposeBag 과 동일

구독(Subscription)의 결과로 AnyCancellable(== Disposalbe)을 반환함

cancel() 을 통해서 구독을 취소하여 더이상의 이벤트를 받지 않도록 하여 memory leak을 방지함

var cancellableBag = Set<AnyCancellable>()

sub.$name
    .filter { $0 != "" }
    .sink(receiveValue: { print("[Published & Sink] 새로운 제 이름은: \\($0)") })
		**.store(in: &cancellableBag)**

---

Creating a custom subscriber

Publisher의 구조

public protocol Publisher {
  // 1
  associatedtype Output

  // 2
  associatedtype Failure : Error

  // 4
  func receive<S>(subscriber: S)
    where S: Subscriber,
    Self.Failure == S.Failure,
    Self.Output == S.Input
}

extension Publisher {
  // 3
  public func subscribe<S>(_ subscriber: S)
    where S : Subscriber,
    Self.Failure == S.Failure,
    Self.Output == S.Input
}

Subscription의 구조

public protocol Subscriber: CustomCombineIdentifierConvertible {
  // 1
  associatedtype Input

  // 2
  associatedtype Failure: Error

  // 3
	// 구독을 받는 곳
  func receive(subscription: Subscription) 

  // 4
	// 구독으로 일어나는 각 이벤트들에 대한 값을 받는 곳
  func receive(_ input: Self.Input) -> Subscribers.Demand

  // 5
	// 구독이 어떠한 이유던 complete가 되었을 때
  func receive(completion: Subscribers.Completion<Self.Failure>)
}

---

Hello Future

Just와 동일함! 단지 Just는 곧바로 값을 방출하는 반면 Future 은 비동기적으로 값을 만들어 방출함

var cancellableBag = Set<AnyCancellable>()

let future = Future<DispatchTime, Never>({ promise in
    DispatchQueue.global().asyncAfter(deadline: .now() + 5, execute: {
        promise(.success(DispatchTime.now()))
    })
})

future
    .sink(receiveCompletion: {
        print("complete: \\($0)")
    }, receiveValue: {
        print("value: \\($0)")
    }).store(in: &cancellableBag)

---

Hello Subject

RxSwift의 Subject와 동일하며 “이벤트를 보낼수도, 구독할수도” 있는 객체

Subject 프로토콜로 구현된 Subject들을 말하며 Subject 프로토콜은 Publisher<Output, Failure> 프로토콜 또한 채택하고 있다 Subject는 send() 를 통해서 값을 주입할 수 있으며 이를 통해 이벤트를 방출한다

• PassTroughtSubject RxSwift의 PublishSubject에 대응하며 초깃값 or 버퍼가 없음
• CurrentValueSubject RxSwift의 BehaviorSubject에 대응하며 초깃값 and 퍼버가 존재함

---

Dynamically adjusting demand

final class IntSubscriber: Subscriber {
  typealias Input = Int
  typealias Failure = Never
  
	// 1
  func receive(subscription: Subscription) {
    subscription.request(.max(2))
  }
  
	// 2
  func receive(_ input: Int) -> Subscribers.Demand {
    print("Received value", input)
    
		switch input {
      case 1:
        return .max(2) // 1
      case 3:
        return .max(1) // 2
      default:
        return .none // 3
    }
  }
  
  func receive(completion: Subscribers.Completion<Never>) {
    print("Received completion", completion)
  }
}

let subscriber = IntSubscriber()

let subject = PassthroughSubject<Int, Never>()

subject.subscribe(subscriber)

(1...10).forEach { subject.send($0) }

/*
1. The new max is 4 (original max of 2 + new max of 2).
2. The new max is 5 (previous 4 + new 1).
3. max remains 5 (previous 4 + new 0).
*/

1에서는 Subscription이 이벤트를 얼마나 받을것인지에 대한 초기값을 설정
2에서는 매번 방출되는 값에 메소드가 호출되며, 여기에 정의된 로직을 통해 Subscription이 이벤트를 어떻게 다룰것인지에 대해 정의
즉 .max()는 이벤트를 얼마나 받을것인지 상수로 박아둔다는 개념 보다는 .max(숫자) 숫자 안에 있는 값 만큼 이벤트를 ‘더’ 받겠다는 덧셈으로 이해하는게 맞는듯.
그래서 메소드 설명보면 Create a demand 라고 적혀있는듯?

---

Type erasure

AnyPublisher<Output, Error> 타입으로 래핑하여 내부 operation들의 연산을 숨겨(지워) 타입을 반환함

let operationChaining = (1...10).publisher
    .print()
    .map { String($0) }
    .tryMap { try Int($0) }
    .compactMap { $0 }

print(type(of: operationChaining))
print(type(of: operationChaining.eraseToAnyPublisher()))

// Result
CompactMap<TryMap<Print<Sequence<ClosedRange<Int>, Never>>, Optional<Int>>, Int>
AnyPublisher<Int, Error>

특히나 API 요청들 같은 경우에는 위와 같은 operation 연산을 거칠 경우 복잡한 반환 타입을 가짐

이럴때 eraseToAnyPublisher()를 활용하면 AnyPublisher<Output, Error> 로 타입을 래핑해버리기에 추후 연산이 바뀌어도 반환 타입이 바뀔 일이 없음