Combine_Chap2_Publishers&Subscribers

Combine 2021. 11. 23. 00:57

이번 챕터에서는, Publishers 와 그것들을 subscribe 하는 여러가지 방법에 대해 실험할 거에요.

((앞으로 자주 테스트에 사용할 코드이니 어떻게 생겼는지만 파악해주세요 :)))

public func example(of description: String,
                    action: () -> Void) {
  print("\n——— Example of:", description, "———")
  action()
}

Combine 의 핵심은 Publisher protocol 에 있어요. 이 protocol 은 하나 또는 다중의 subscriber 로 sequence of values 를 보낼 수 있는 타입의 조건들을 정의합니다. 다시 말하면, publisher 는 values 를 포함하는 events 를 보냅니다 (publish or emit).

publisher 를 subscribing 하는 것과 NotificationCenter 의 특정 notification 을 subscribing 하는 것은 유사해요. NotificationCenter 에서 관심있는 events 를 쓰면, 새로운 events 가 들어올 때마다 비동기적으로 notification 을 받을 수 있습니다.

이것들은 매우 유사해서, NotificationCenter 는 publisher(for:object) method 를 가지고 있고, 이 method 는 notifications 를 보낼 수 있는 Publisher 를 제공합니다.

example(of: "Publisher") {
  // Notification 이름을 지어줍니다. 
  let myNotification = Notification.Name("MyNotification")

  // NotificationCenter 의 기본 instance 에 접근 후, publisher method 를 호출, 그리고 반환된 값을 할당해줍니다.
  let publisher = NotificationCenter.default
    .publisher(for: myNotification, object: nil)
}

publisher(for:object:) method 를 보면 아래와 같이 NotificationCenter.Publisher, 즉 Publisher 를 반환한다는 것을 알 수 있어요.

func publisher(for name: Notification.Name, object: AnyObject? = nil) -> NotificationCenter.Publisher

Summary를 보면 'notifications 를 broadcasting 할 때 events 를 보내는 publisher 를 반환한다' 고 합니다.

이러한 methods 는 이미 있는 older async API 과 새로운 Combine 사이의 연결고리 정도로 생각하면 될 것 같아요.

Publisher 는 두 종류의 events 를 보냅니다.

1. Values (elements)

2. completion event.

Publisher 는 0개 ~ 여러개의 values, 그리고 한개의 completion event (normal or error) 를 보냅니다. publisher 가 completion event 를 보내면 종료, 즉 그 후에는 어떠한 events 도 보낼 수 없습니다.

이 관점에서 보면, publisher 와 Swift iterator 는 비슷한 면이 있습니다. 가장 다른 점은 Publisher 의 completion 은 success 와 fail 이 있고, iteraotr 에서는 values 를 pull 해야 하는 반면 Publisher 는 consumers 에게 values 를 push 한다는 것이죠.

Notification 예제는 observe, post notification, 그리고 observer 를 unregister 하는 것으로 마치도록 하겠습니다.

example(of: "Publisher") {
  // Notification 이름을 지어줍니다. 
  let myNotification = Notification.Name("MyNotification")

  // NotificationCenter 의 기본 instance 에 접근 후, publisher method 를 호출, 그리고 반환된 값을 할당해줍니다.
  let publisher = NotificationCenter.default
    .publisher(for: myNotification, object: nil)
    
    
    let center = NotificationCenter.default

    // 이전에 생성한 것과 같은 이름으로 해당 notification 을 받을 observer 를 생성합니다.
    let observer = center.addObserver(
      forName: myNotification,
      object: nil,
      queue: nil) { notification in
        print("Notification received!")
    }

    // 해당 이름으로 notification 을 보냅니다.
    center.post(name: myNotification, object: nil)

    // center 에서 observer 를 제거합니다.
    center.removeObserver(observer)
}

Subscriber 는 publisher 로 부터 input 을 받을 수 있는 타입의 조건들을 정의한 protocol 입니다.

example(of: "Subscriber") {
  let myNotification = Notification.Name("MyNotification")
  let center = NotificationCenter.default

  let publisher = center.publisher(for: myNotification, object: nil)
}

여기서 만약 notification 을 post 하려고 한다면, publisher 는 notification 을 받은 subscription 이 없기 때문에 값을 보내지 않을 거에요.

이제 코드를 추가시킵니다.

example(of: "Subscriber") {
    let myNotification = Notification.Name("MyNotification")
    let center = NotificationCenter.default
    
    let publisher = center.publisher(for: myNotification, object: nil)
    
    // publisher 에 sink 를 호출함으로써 subscription 을 생성합니다. 
    let subscription = publisher
        .sink { _ in
            print("Notification received from a publisher!")
        }
    
    // notification 을 post 합니다. 
    center.post(name: myNotification, object: nil)
    
    // subscription 을 취소합니다.
    subscription.cancel()
    
    
}

.sink 를 살펴보면 다음과 같습니다.

func sink(receiveValue: @escaping ((Notification) -> Void)) -> AnyCancellable

(Notification) -> Void type 의 escaping closure 를 받고 AnyCancellable 을 반환하네요.

즉, closure 를 실행 시키고 subscription 에 AnyCancellable 을 반환합니다.

위 예시에서는 단순히 notifiaction 을 받았다는 것을 표시하기 위해 출력하고 있습니다.

——— Example of: Publisher ———
Notification received!

sink operator 는 가능한 한 많은 values 를 publisher 가 값을 보낼 때 마다 받을 것입니다. ( - unlimited demand 라 불립니다.)

그리고, sink 는 두가지의 closures 를 제공합니다. 하나는 completion 을 , 다른 하나는 value 을 다루기 위한 closure 입니다.

example(of: "Just") {
  // Just 를 이용해서 publisher 를 생성합니다. (하나의 값으로부터 publisher 를 생성하도록 함)
  let just = Just("Hello world!")
  
  // publisher 에 대해 subscription 을 생성, 받은 각 event 에 대해 메시지를 출력
  _ = just
    .sink(
      receiveCompletion: {
        print("Received completion", $0)
      },
      receiveValue: {
        print("Received value", $0)
    })
}

출력 결과는 아래와 같습니다.

——— Example of: Just ———
Received value Hello world!
Received completion finished

Just 는 각 subscriber 에 output 을 한번 보낸 후 종료하는 publisher 입니다.

built-in assign(to:on:) operator 는 받은 값을 object 의 KVO-compliant property 에 할당하는 operator 입니다.

(KVO-compliant : Key-Value Observing 을 따르는 )

example(of: "assign(to:on:)") {
  // (새로운 값을 출력하는) didSet 을 가진 property 를 property로 가진 class 선언 
  class SomeObject {
    var value: String = "" {
      didSet {
        print(value)
      }
    }
  }
  
  // 위에서 만든 class 의 instance 생성.
  let object = SomeObject()
  
  // [String] 으로부터 publisher 생성
  let publisher = ["Hello", "world!"].publisher
  
  // publisher 를 subscribe, 그리고 받은 각 값을 object 의 'value' property 에 할당
  _ = publisher
    .assign(to: \.value, on: object)
}

.assign(to:on:) 의 정의는 아래와 같다.

func assign<Root>(to keyPath: ReferenceWritableKeyPath<Root, Self.Output>, on object: Root) -> AnyCancellable

위를 실행하면 다음과 같이 출력된다.

——— Example of: assign(to:on:) ———
Hello
world!

assign(to:on:) 은 UIKit 또는 AppKit 에서 사용할 때 labels, text views, checkboxes, 그 외 다른 UIComponents 에 직접적으로 값을 할당할 수 있기 때문에 매우 유용하다.

assign operator 중에 publisher 로 부터 보내진 값을 @Published wrapper 가 씌인 property 에 republish 할 수 있는 assign(to:) 도 있다.

example(of: "assign(to:)") {
  
  // @Published wrapper 가 씌인 property 를 가진 class 선언 및 instance 생성
  // 일반적인 property 임과 동시에 value 에 대해 publisher 를 생성한다. 
  
  class SomeObject {
    @Published var value = 0
  }
  
  let object = SomeObject()
  
  // underlying publisher 에 접근, subscribe, 그리고 출력하기 위해 $ prefix 를  @Published property 에 붙여준다. 
  object.$value
    .sink {
      print($0)
    }
  
  // 숫자들인 publisher 를 생성, object 의 value publisher 에 각 값을 보낸다. 
  // 여기서 & 은 property 에 대한 input reference 을 나타내기 위해 사용. 
  (0..<10).publisher
    .assign(to: &object.$value)
}

assign(to:) operator 는 AnyCancellable 을 반환하지 않는데, 이는 내부에서 lifecycle 을 관리하고, @Published property 가 deinitialize 될 때 subscription 을 취소하기 때문이다.

func assign(to published: inout Published<Self.Output>.Publisher)

아래 MyObject class 에서 assign(to:\.word, on:self) 사용과 결과로 나오는 AnyCancellable 을 저장하는 것은 strong reference cycle 을 만든다. 이때, assign(to:on:) 을 assign(to: &$word) 로 바꾸면 이러한 문제를 해결할 수 있다.

class MyObject {
  @Published var word: String = ""
  var subscriptions = Set<AnyCancellable>()

  init() {
    ["A", "B", "C"].publisher
      .assign(to: \.word, on: self)
      .store(in: &subscriptions)
  }
}

Subscriber 가 작업을 끝내고, 더이상 publisher 로부터 값을 받는것을 원하지 않을 때, subscription 을 취소하고 관련 resources 의 메모리를 해제, 그리고 다른 것들이 발생하지 않도록 멈추게 하는게 좋다.

Subscriptions 는 AnyCancellable instance 를 반환하고, 따라서 작업이 끝난 후 subscription 을 취소할 수 있다. AnyCancellable 은 Cancellable protocol 을 따르며, 이 protocol 은 cancel() method 를 바로 위 목적을 위해 필요로 한다.

Subscription protocol 은 Cancellable 을 상속받았기 때문에 subscription 에 cancel() 호출이 가능하다.

만약 subscription 에 cancel() 을 명시적으로 해주지 않으면 publisher 가 끝나거나, 메모리 관리하는 곳에서 stored subscription 을 deinit 시킬 때까지 계속될 것이다.

위 그래프를 서술하면 아래와 같다.

Subscriber 가 publisher 를 subscribe 한다.
Publisher 가 subscription 을 만든 후, subscriber 에게 준다.
Subscriber 가 values 를 요청한다.
Publisher 가 values 을 보낸다.
Publisher 가 completion 을 보낸다.

이제, Publisher protocol 을 살펴보자.

public protocol Publisher {
  // Publisher 가 생성할 수 있는 값들의 type
  associatedtype Output

  // publisher 가 생성할 수도 있는 error type, 생성을 절대 안할 경우 Never 
  associatedtype Failure : Error

  // The implementation of subscribe(_:) will call receive(subscriber:) to attach the subscriber
  // to the publisher, i.e., create a subscription.

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

extension Publisher {
  // A subscriber calls subscribe(:) on a publisher to attach to it.
  public func subscribe<S>(_ subscriber: S)
    where S : Subscriber,
    Self.Failure == S.Failure,
    Self.Output == S.Input
}

The associated types are the publisher’s interface that a subscriber must match in order to create a subscription. Now, look at the Subscriber protocol:

public protocol Subscriber: CustomCombineIdentifierConvertible {
  // subscriber 가 받을 수 있는 값들의 type
  associatedtype Input

  // subscriber 가 받을 수 있는 error type, error 을 받지 않을 경우 Never.
  associatedtype Failure: Error

  // The publisher calls receive(subscription:) on the subscriber 
  // to give it the subscription.
  func receive(subscription: Subscription)

  // The publisher calls receive(_:) on the subscriber 
  // to send it a new value that it just published.
  func receive(_ input: Self.Input) -> Subscribers.Demand

  // The publisher calls receive(completion:) on the subscriber 
  // to tell it that it has finished producing values, either normally or due to an error.
  func receive(completion: Subscribers.Completion<Self.Failure>)
}

publisher 와 subscriber 사이의 연결은 subscription 이다.

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

The subscriber calls request(_:) to indicate it is willing to receive more values, up to a max number or unlimited.

import Combine
import Foundation

var subscriptions = Set<AnyCancellable>()

public func example(of description: String, action: () -> Void) {
    print("\n-----------Example of:", description, "------------")
    action()
}


example(of: "Publisher") {
    let myNotification = Notification.Name("MyNotification")

    let publisher = NotificationCenter.default.publisher(for: myNotification, object: nil)
    // func publisher(for name: Notification.Name, object: AnyObject? = nil) -> NotificationCenter.Publisher
    // Summary: Returns a publisher that emits events when broadcasting notifications

    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)
}
//The example's title is a little misleading
//because the output is not actually coming from a publisher.
//For that to happen, you need a subscriber.






/*
 Subscriber is a protocol that defines the requirements for a type to be able to receive input from a publisher.

 */

example(of: "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()

}



example(of: "Just") {
    // 1 Create a publisher using Just, which lets you create a publisher from a primitive value type.
    let just = Just("Hello world!")
    // Just: A publisher that emits an output to each subscriber just once, and then finishes.

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


example(of: "assign(to:on:)") {

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

    let object = SomeObject()

    let publisher = ["Hello", "World"].publisher

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


}



public protocol Publisher2 {
//    1. The type of values that the publisher can produce.
    associatedtype Output

//    2. The type of error a publisher may produce, or Never if the publisher is guaranteed to not produce an error.
    associatedtype Failure: Error

//    4. The implementation of subscribe(_:) will call receive(subscriber:) to
//    attach the subscriber to the publisher, i.e., create a subscription.
    func receive<S>(subscriber: S) where S: Subscriber, Self.Failure == S.Failure, Self.Output == S.Input


}

extension Publisher {
//    3. A subscriber calls subscribe(_:) on a publisher to attach to it. ???
    public func subscribe<S>(_ subscriber: S) where S: Subscriber, Self.Failure == S.Failure, Self.Output == S.Input {}
}




public protocol Subscriber2: CustomCombineIdentifierConvertible {
//    1. The type of values a subscriber can receive.
    associatedtype Input

//    2. The type of error a subscriber can receive; or Never if the subscriber won’t receive an error.
    associatedtype Failure: Error

//    3. The publisher calls receive(subscription:) on the subscriber to give it the subscription.  ???
    func receive(subscription: Subscription)

//    4. The publisher calls receive(_:) on the subscriber to send it a new value that it just published.
    func receive(_ input: Self.Input) -> Subscribers.Demand
    // Demand: A requested number of items, sent to a publisher from a subscriber through the subscription.

//    5. The publisher calls receive(completion:) on the subscriber.
//    to tell it that it has finished producing values, either normally or due to an error.
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

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

// Cancellable: A protocol indicating that an activity or action supports cancellation.


example(of: "Custom Subscriber") {
    let publisher = (1...6).publisher
//    let publisher = ["A","B","C","D"]
//    let publisher = [1,2,3,4,5,6].publisher
    final class IntSubscriber: Subscriber {
        typealias Input = Int
        typealias Failure = Never

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

        func receive(_ input: Int) -> Subscribers.Demand {
            print("Received value: ", input)
//            return .none
            return .unlimited
//            return .max(1)
        }

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

    let subscriber = IntSubscriber()
    publisher.receive(subscriber: subscriber)


}


// Future
//Much like you can use Just to create a publisher that emits a single value to a subscriber and then complete,
//a Future can be used to asynchronously produce a single result and then complete.


example(of: "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))
            }
        }
    }

//    1. Create a future using the factory function you created earlier,
//    specifying to increment the integer you passed after a three-second delay.
    let future = futureIncrement(integer: 1, afterDelay: 3)


//    2. Subscribe to and print the received value and completion event,
//    and store the resulting subscription in the subscriptions set.

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


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


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

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

    let subscriber = StringSubscriber()

    let subject = PassthroughSubject<String, MyError>()

//    6. Subscribes the subscriber to the subject.
    subject.subscribe(subscriber)

//    7. Creates another subscription using sink.
    let subscription = subject
        .sink { 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(.test))
    subject.send(completion: .finished)
    subject.send("How about another one?")
}

// Passthrough subjects enable you to publish new values on demand. They will happily pass along those values and a completion event.





example(of: "CurrentValueSubject") {
//    1. Create a CurrentValueSubject of type Int and Never.
//    This will publish integers and never publish an error, with an initial value of 0.


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

//    Create a subscription to the subject and print values received from it.
    subject
        .print()
        .sink(receiveValue: {print($0)})
        .store(in: &subscriptions)

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

//        Unlike a passthrough subject, you can ask a current value subject for its value at any time.
//        print(subject.value)




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

    subject.value = 3
    print(subject.value)

    subject.send(completion: .finished)

}




example(of: "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)
    
}


example(of: "Type erasure") {
    let subject = PassthroughSubject<Int, Never>()
    
    let publisher = subject.eraseToAnyPublisher()
    
    publisher
        .sink(receiveValue: {print($0)})
        .store(in: &subscriptions)
    
    subject.send(0)
}

// One example of when you would want to use type erasure for a publisher is when you want to use a pair of public and private properties, to allow the owner of those properties to send values on the private publisher, and let outside callers only access the public publisher for subscribing but not be able to send values.






/*
Key points
 
• Publishers transmit a sequence of values over time to one or more subscribers, either synchronously or asynchronously.
 
• A subscriber can subscribe to a publisher to receive values; however, the subscriber’s input and failure types must match the publisher’s output and failure types.
 
• There are two built-in operators you can use to subscribe to publishers: sink(_:_:) and assign(to:on:).
 
• A subscriber may increase the demand for values each time it receives a value, but it cannot decrease demand.
 
• To free up resources and prevent unwanted side effects, cancel each subscription when you’re done.
 
• You can also store a subscription in an instance or collection of AnyCancellable to receive automatic cancelation upon deinitialization.
 
• A future can be used to receive a single value asynchronously at a later time.
 
• Subjects are publishers that enable outside callers to send multiple values asynchronously to subscribers, with or without a starting value.
 
• Type erasure prevents callers from being able to access additional details of the underlying type.
 
• Use the print() operator to log all publishing events to the console and see what’s going on.
 
*/

import Combine
import Foundation

var subscriptions = Set<AnyCancellable>()

public func example(of description: String, action: () -> Void) {
    print("\n-----------Example of:", description, "------------")
    action()
}

Publisher

example(of: "Publisher") {
    let myNotification = Notification.Name("MyNotification")

    let publisher = NotificationCenter.default.publisher(for: myNotification, object: nil)
    // func publisher(for name: Notification.Name, object: AnyObject? = nil) -> NotificationCenter.Publisher
    // Summary: Returns a publisher that emits events when broadcasting notifications

    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)
}

The example's title is a little misleading because the output is not actually coming from a publisher.
For that to happen, you need a subscriber.

Subscriber

Subscriber is a protocol that defines the requirements for a type to be able to receive input from a publisher.

example(of: "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()

}

example(of: "Just") {
    // 1 Create a publisher using Just, which lets you create a publisher from a primitive value type.
    let just = Just("Hello world!")
    // Just: A publisher that emits an output to each subscriber just once, and then finishes.

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

example(of: "assign(to:on:)") {

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

    let object = SomeObject()

    let publisher = ["Hello", "World"].publisher

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

public protocol Publisher2 {
//    1. The type of values that the publisher can produce.
    associatedtype Output

//    2. The type of error a publisher may produce, or Never if the publisher is guaranteed to not produce an error.
    associatedtype Failure: Error

//    4. The implementation of subscribe(_:) will call receive(subscriber:) to
//    attach the subscriber to the publisher, i.e., create a subscription.
    func receive<S>(subscriber: S) where S: Subscriber, Self.Failure == S.Failure, Self.Output == S.Input
}

extension Publisher2 {
//    3. A subscriber calls subscribe(_:) on a publisher to attach to it. ???
    public func subscribe<S>(_ subscriber: S) where S: Subscriber, Self.Failure == S.Failure, Self.Output == S.Input {}
}

public protocol Subscriber2: CustomCombineIdentifierConvertible {
//    1. The type of values a subscriber can receive.
    associatedtype Input

//    2. The type of error a subscriber can receive; or Never if the subscriber won’t receive an error.
    associatedtype Failure: Error

//    3. The publisher calls receive(subscription:) on the subscriber to give it the subscription.  ???
    func receive(subscription: Subscription)

//    4. The publisher calls receive(_:) on the subscriber to send it a new value that it just published.
    func receive(_ input: Self.Input) -> Subscribers.Demand
    // Demand: A requested number of items, sent to a publisher from a subscriber through the subscription.

//    5. The publisher calls receive(completion:) on the subscriber.
//    to tell it that it has finished producing values, either normally or due to an error.
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

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

example(of: "Custom Subscriber") {
    let publisher = (1...6).publisher
//    let publisher = ["A","B","C","D"]
//    let publisher = [1,2,3,4,5,6].publisher
    final class IntSubscriber: Subscriber {
        typealias Input = Int
        typealias Failure = Never

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

        func receive(_ input: Int) -> Subscribers.Demand {
            print("Received value: ", input)
//            return .none
            return .unlimited
//            return .max(1)
        }

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

    let subscriber = IntSubscriber()
    publisher.receive(subscriber: subscriber)
}

Future

Much like you can use Just to create a publisher that emits a single value to a subscriber and then complete,

a Future can be used to asynchronously produce a single result and then complete.

example(of: "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))
            }
        }
    }

//    1. Create a future using the factory function you created earlier,
//    specifying to increment the integer you passed after a three-second delay.
    let future = futureIncrement(integer: 1, afterDelay: 3)


//    2. Subscribe to and print the received value and completion event,
//    and store the resulting subscription in the subscriptions set.

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


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

// Passthrough subjects enable you to publish new values on demand. They will happily pass along those values and a completion event.

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

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

    let subscriber = StringSubscriber()

    let subject = PassthroughSubject<String, MyError>()

//    6. Subscribes the subscriber to the subject.
    subject.subscribe(subscriber)

//    7. Creates another subscription using sink.
    let subscription = subject
        .sink { 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(.test))
    subject.send(completion: .finished)
    subject.send("How about another one?")
}

example(of: "CurrentValueSubject") {
//    1. Create a CurrentValueSubject of type Int and Never.
//    This will publish integers and never publish an error, with an initial value of 0.


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

//    Create a subscription to the subject and print values received from it.
    subject
        .print()
        .sink(receiveValue: {print($0)})
        .store(in: &subscriptions)

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

//        Unlike a passthrough subject, you can ask a current value subject for its value at any time.
//        print(subject.value)



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

    subject.value = 3
    print(subject.value)

    subject.send(completion: .finished)
}

example(of: "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)
    
}

example(of: "Type erasure") {
    let subject = PassthroughSubject<Int, Never>()
    
    let publisher = subject.eraseToAnyPublisher()
    
    publisher
        .sink(receiveValue: {print($0)})
        .store(in: &subscriptions)
    
    subject.send(0)
}

One example of when you would want to use type erasure for a publisher is when you want to use a pair of public and private properties, to allow the owner of those properties to send values on the private publisher, and let outside callers only access the public publisher for subscribing but not be able to send values.

'Combine' 카테고리의 다른 글

Combine-Chap7_Sequence Operator (0)	2021.12.10
Combine_Chap5_CombiningOperators (0)	2021.11.26
Combine_Chap4_FilteringOperators (0)	2021.11.21
Combine_Chap3_TransformingOperators (0)	2021.11.21
Swift - Combine (0)	2021.11.16

ABOUT ME

iOS 를 매우매우 좋아하는 개발자의 노트 iOS 를 매우매우 좋아하는 개발자의 노트

'Combine' 카테고리의 다른 글

티스토리툴바

ABOUT ME

'Combine' 카테고리의 다른 글

관련글 관련글 더보기

티스토리툴바