Ch2. Publishers & Subscribers

Hello Publisher

let myNotification =  Notification.Name("MyNotification")
let center = NotificationCenter.default

let observer = center.addObserver(forName: myNotification, object: nil, queue: nil) { (notification) in
    print("Notification received!")
}

center.post(name: myNotification, object: nil)
center.removeObserver(observer)

Notification received!

Hello Subscriber

let myNotification =  Notification.Name("MyNotification")
let publisher = NotificationCenter.default.publisher(for: myNotification, object: nil)
let center = NotificationCenter.default

let subscription = publisher.sink { _ in
    print("Notification received from a publisher!")
}

center.post(name: myNotification, object: nil)
subscription.cancel()

Notification received from a publisher!

let just = Just("Hello world!")

_ = just
    .sink(
        receiveCompletion: {
            print("Received completion", $0)
        },
        receiveValue: {
            print("Received value", $0)
        })

_ = just
    .sink(
        receiveCompletion: {
            print("Received completion (another)", $0)
        },
        receiveValue: {
            print("Received value (another)", $0)
        })

Received value Hello world!
Received completion finished
Received value (another) Hello world!
Received completion (another) finished

Subscribing with assign(to:on:)

class SomeObject {
    var value: String = "" {
        didSet {
            print(value)
        }
    }
}

let object = SomeObject()

let publisher = ["Hello", "world!"].publisher

_ = publisher
    .assign(to: \\.value, on: object)

Hello
world!

Hello Cancellable

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
}

1. Publisher가 전달하는 값의 타입
2. Publisher가 전달하는 에러의 타입. 에러가 발생하지 않는다면 Never
3. Subscriber는 subscribe(_:)를 사용하여 Publisher에 연결된다.
4. subscribe(_:)를 구현하면 receive(subscriber:)가 호출되어 subscriber가 publisher에 연결된다. (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
  func receive(completion: Subscribers.Completion<Self.Failure>)
}

1. Subscriber가 받는 값의 타입
2. Subscriber가 받는 에러의 타입. 에러가 발생하지 않는다면 Never
3. Publisher가 receive(subscription:)을 호출하면 subsciber는 이 subscription을 전달받는다.
4. Publisher가 receive(_:)를 호출하면 subscriber는 값을 전달받는다.
5. Publisher가 receive(completion:)를 호출하여 값을 보내는 것이 끝났음을 알려준다.

public protocol Subscription: Cancellable, CustomCombineIdentifierConvertible {
  func request(_ demand: Subscribers.Demand)
}

Publisher와 Subscriber간의 연결은 Subscription이다.

Subscriber가 request(_:)를 호출하여 난 값을 더 받을거에요~~ 라고 말하는 것임

Creating a custom subscriber

let publisher = (1...6).publisher

final class IntSubscriber: Subscriber {
    
    typealias Input = Int
    typealias Failure = Never

    func receive(subscription: Subscription) {
				// 야 나 값 3개만 받을거임 ㅇㅋ?
        subscription.request(.max(3))
    }

    func receive(_ input: Int) -> Subscribers.Demand {
        print("Received value", input)
				// 보내준 값 잘 받았고 더 받을생각 없음. 진짜 3개까지만!!
        return .none
    }

    func receive(completion: Subscribers.Completion<Never>) {
        print("Received completion", completion)
    }
}

let subscriber = IntSubscriber()

publisher.subscribe(subscriber)

Received value 1
Received value 2
Received value 3

Hello Future

func futureIncrement(
    integer: Int,
    afterDelay delay: TimeInterval) -> Future<Int, Never> {
    Future<Int, Never> { promise in
        print("Original")
        DispatchQueue.global().asyncAfter(deadline: .now() + delay) {
            promise(.success(integer + 1))
        }
    }
}

let future = futureIncrement(integer: 1, afterDelay: 3)

future
    .sink(receiveCompletion: { print($0) },
          receiveValue: { print($0) })
    .store(in: &subscriptions)

future
    .sink(receiveCompletion: { print("Second", $0) },
          receiveValue: { print("Second", $0) })
    .store(in: &subscriptions)

Just와 유사하다. 하지만 보시다시피 딜레이가 있음

Received value 1
Received value 2
Received value 3

Hello Subject

이거 RxSwift의 PublishSubject, BehaviorSubject 떠올리면 비슷함

PassthroughSubject

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)
        // 3
        return input == "World" ? .max(1) : .none
    }

    func receive(completion: Subscribers.Completion<MyError>) {
        print("Received completion", completion)
    }
}

let subscriber = StringSubscriber()

let subject = PassthroughSubject<String, MyError>()

subject.subscribe(subscriber)

let subscription = subject
    .sink(
        receiveCompletion: { completion in
            print("Received completion (sink)", completion)
        },
        receiveValue: { value in
            print("Received value (sink)", value)
        }
    )

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?")

Received value Hello
Received value (sink) Hello
Received value World
Received value (sink) World
Received value Still there?
Received completion failure(__lldb_expr_13.(unknown context at $109b37024).(unknown context at $109b370f8).(unknown context at $109b37100).MyError.test)

CurrentValueSubject

var subscriptions = Set<AnyCancellable>()

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

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

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

print(subject.value)
subject.value = 3
print(subject.value)

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

subject.send(completion: .finished)

receive subscription: (CurrentValueSubject)
request unlimited
receive value: (0)
0
receive value: (1)
1
receive value: (2)
2
2
receive value: (3)
3
3
receive subscription: (CurrentValueSubject)
request unlimited
receive value: (3)
Second subscription: 3
receive finished
receive finished

초기값 필수

subject.value = 3 와 같은 코드로 이벤트를 발생할 수도 있음

Dynamically adjusting demand

final class IntSubscriber: Subscriber {
    typealias Input = Int
    typealias Failure = Never

    func receive(subscription: Subscription) {
        subscription.request(.max(2))
    }

    func receive(_ input: Int) -> Subscribers.Demand {
        print("Received value", input)

        switch input {
        case 1:
            return .max(2)
        case 3:
            return .max(1)
        default:
            return .none
        }
    }

    func receive(completion: Subscribers.Completion<Never>) {
        print("Received completion", completion)
    }
}

let subscriber = IntSubscriber()

let subject = PassthroughSubject<Int, Never>()

subject.subscribe(subscriber)

subject.send(1)
subject.send(2)
subject.send(3)
subject.send(4)
subject.send(5)
subject.send(6)

Received value 1
Received value 2
Received value 3
Received value 4
Received value 5

Type erasure

let subject = PassthroughSubject<Int, Never>()

let publisher = subject.eraseToAnyPublisher()

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

subject.send(0)

0

publisher.send(??)가 가능할까?

노노!

publisher의 타입은 AnyPublisher<Int, Never> 이고 send를 사용할 수 없음.

Any가 좀 익숙함?

앞에서 AnyCancellable을 배웠잖슴 ㅋ

Key points

• Publisher는 값의 시퀀스를 subscriber에게 전달한다. sync 또는 async하게
• Subscriber는 publisher를 subscribe하여 값을 받을 수 있다. 이 값의 타입과 failure type이 완전히 일치해야 한다.
• Combine에 build-in된 2개의 오퍼레이터는 sink(_:_:)와 assign(to:on:)이 있다.
• Subscriber는 전달받고자 하는 값의 개수(demand)를 증가시킬 수 있다. 감소시키는건 불가능
• Subscription객체를 따로 저장하거나 AnyCancellable의 Set에 저장할 수 있다. 취소하거나 해제하여 메모리확보하자
• future는 지연된 시간에 하나의 밸류를 전달받기에 좋은 놈이다
• Subject는 Publisher이고 caller외부에서 값을 전달하기에 좋다. 초기값이 없으면 PassthroughSubject 초기값이 있으면 CurrentValueSubject
• Type erasure는 caller가 publisher에 디테일하게 접근하지 않도록 해준다. (래퍼느낌)
• print()를 사용하면 전달받는 이벤트를 출력할 수 있다.