[Swift] Combine의기본 Publisher이란?

예전에 Published를 스유를 기준으로 다뤘다면

이번에는 비동기를 기준으로 다룰 Publisher에 대해 알아보려고해요

둘이 다른건 아니에요!

 

 

스위프트 API에서 이미 많은 publisher를 지원하고있어요

이렇게 배열을 만들고 publisher로 변환할 수도 있구요

 

 

그래서 publisher가 뭔데?

 

Combine프레임워크에 속한 타입이구요

Combine을 이용하면 delegate callback, completion handler closure를 사용하는대신

이벤트에대한 단일처리 체인을 만들 수 있어요

 

 

Combine의 핵심요소로 publisher, operator, subscriber가 있구요

그중 하나가 바로 publisher이죠

 

애플문서에는 

Declares that a type can transmit a sequence of values over time.

이라고 정의되어있구요

구독자와 같은 하나이상의 관계자에게 시간이 지남에 따라 값을 보낼 수 있는 타입

이라고 설명할 수 있겠네요

관심 있는 값이나 이벤트를 (publish)게시할 수 있죠

 

간단하게 동작을 보자면

1. 관심있는 값에 구독을 걸고

2. 그 값에 새 이벤트가 발생하면(값이 변경되면)

3. 비동기식으로 이벤트(값)을 전달 받을 수 있습니다.

 

이 떄 publisher는 2가지종류의 이벤트를 발생시키는데

값 또는 완료 이벤트를 발생시킵니다.

값은 0개이상 방출할 수 있고

완료 이벤트는 정상적인 .finished이벤트와, 오류일 수 있습니다.

 

이렇게만든 게시자를

구독해서 사용하는데

두가지 방법으로 구독할 수 있어요

 

1. assign을 이용해서 UI컴포넌트에 직접 값을 할당할 수 있는 방법

2. sink를 이용해서 수신 값을 처리하는것과 완료 이벤트를 수신처리하는 클로저로 값을 처리하는 방법

 

 

그리고 이렇게 게시자를 구독하고 작업이 끝나면

자원을 확보하기위해 구독을 취소해줘야하는데

Cancellable타입을 이용해서 처리해요

 

구독을하면 AnyCancellable인스턴스를 반환하고

AnyCancellable은 cancel()메서드를 필요로하는 Cancellable프로토콜을 준수하고있어요

cancel()을 명시적으로 호출하지않으면

게시자가 완료될때까지

또는

메모리 관리로 인해 저장된 구독이 해제될때 까지 계속 살아있어요

 

cancel()을 사용하는 방법말고 또다른 방법으로

.store()함수를 이용해서

AnyCancellable를 저장해놓고 해당 변수가 해제될때 구독을 취소하는 방식이 있어요

// cancel이용
let cancellable2 = [1,2,3].publisher
    .sink { print($0) }

cancellable2.cancel()


// store이용
var cancellable = Set<AnyCancellable>()
[1,2,3].publisher
    .sink { print($0) }
    .store(in: &cancellable)

 

게시자가 구독되는 원리를 알아보면

1. 구독자가 게시자를 구독

2. 게시자는 구독을 생성해서 구독자에게 제공

3. 구독자가 값을 요청함

4. 게시자가 값을 보냄

5. 게시자가 완료를 보냄

순으로 이뤄지는데요

 

// MARK: - 구독
public protocol Subscription : Cancellable, CustomCombineIdentifierConvertible {

    /// Tells a publisher that it may send more values to the subscriber.
    func request(_ demand: Subscribers.Demand)
}

// MARK: - 게시자
public protocol Publisher {
    associatedtype Output
    associatedtype Failure : Error
    
    func receive<S>(subscriber: S)
    where S: Subscriber,
          Self.Failure == S.Failure,
          Self.Output == S.Input
}
extension Publisher {
    public func subscribe<S>(_ subscriber: S)
    where S : Subscriber,
          Self.Failure == S.Failure,
          Self.Output == S.Input
}

// MARK: - 구독자
public protocol Subscriber: CustomCombineIdentifierConvertible {
  associatedtype Input
  associatedtype Failure: Error

  func receive(subscription: Subscription)

  func receive(_ input: Self.Input) -> Subscribers.Demand

  func receive(completion: Subscribers.Completion<Self.Failure>)
}

위에서말한 과정을 세부과정으로 살펴보면

1. 구독자가 게시자를 구독

Publisher의 인스턴스함수 subscribe()를 호출하면서 구독자를 담아서 전달하고

 

2. 게시자는 구독을 생성해서 구독자에게 제공

Publisher은 내부적으로 receive()함수를 호출하고

게시자가 받은 구독자에게 Subscriber의 인스턴스 함수 receive(subscription: )을 통해 구독을 전달 합니다.

이 작업을 통해 Publisher(게시자)와 Subscriber(구독자)가 attach(결합)됩니다.

 

3. 구독자가 값을 요청함

게시자로부터 구독을 받았으니

(구독자가 게시자의 receive를 호출하며 구독자 자신을 전달하고

게시자가 subscriber.receive(subscription: )을 호출하고 구독을 전달하고

구독자가 전달된 subscription으로 subscription.request를 호출)

Subscription(구독)의 인스턴스함수인 request()를 호출해서 값을 요청합니다.

이때

requet함수는 Subscribers.Demand타입을 매개변수로 받는데

게시자가 받을 수 있는 값의 최대 수를 정해줍니다.

.unlimited - 제한없음

.none - 받지않음

.max(value: ) - 최대 value개 만큼 받음

를 지정할수 있습니다.

 

4. 게시자가 값을 보냄

이벤트가 발생하면 게시자가 구독자에게 값을 전달합니다.

Subscriber의 인스턴스 메소드 receive(_ input)를 이용해서 전달하며

구독자는 값을 받습니다.

반환값으로 Subscribers.Demand타입을 설정 할 수 있는데

값을 받을때 그 뒤로 몇개의 값을 더 받을지 설정할 수 있습니다.

만약 최초로 max(2)를 설정했다면 2개를 받을 수 있었고

이 함수에서 max(3)을 반환한다면

2 + 3 -> 게시자로부터 5개의 값을 받을 수 있습니다.

만약 값이 또들어온다면

2 + 3 + 3으로 8개의 값을 받을 수 있습니다.

 

위에서 정한 갯수만큼 받았다면 더이상 값을 받지않습니다.

이때 모든 이벤트를 받지않았기 때문에 완료이벤트는 발생하지않습니다.

 

5. 게시자가 완료를 보냄

게시자가 모든 값을 보냈다면 혹은 완료 이벤트를 

receive(completion: )함수를 통해서 구독자에게 완료이벤트를 전달합니다.

 

 

sink를 사용하는경우

This method creates the subscriber and immediately requests an unlimited number of values, prior to returning the subscriber.

즉시 .unlimited 타입으로 구독자를 생성해서 게시자와 연결시켜준다고해요

구독자를 직접 설정해서 생성하지않고도 편리하게 사용할 수 있죠

 

 

 

Publisher을 다르게 사용하는 방법에대해 간단하게 알아보고

추후에 자세히 게시글을 작성해보게씁니다! 

 

 

@Published 프로퍼티 레퍼

지금까지 

[1,2,3].publisher이런식으로 게시자를 만들었는데

@Published 프로퍼티 레퍼로 정의하여 생성하면

일반 프로퍼티로 접근할 수 있을 뿐만아니라, 값에대한 publisher가 생성되요

$로 접근하여 publisher에 접근할 수 있구요

프로퍼티가 해제될때 내부적으로 라이프사이클을 관리하고 구독을 취소해요

.assign(to: )함수와 함께 사용하기 좋죠

 

 

Just

단일 값Publisher을 생성할 수 있는타입

구독자에게 출력을 한번 내보낸다음 완료하는 게시자

public struct Just<Output> : Publisher {

    /// The kind of errors this publisher might publish.
    ///
    /// Use `Never` if this `Publisher` does not publish errors.
    public typealias Failure = Never
}

 

Future

단일 결과를 비동기적으로 생성한다음 완료할 수 있는 타입

final public class Future<Output, Failure> : Publisher
  where Failure: Error {
  public typealias Promise = (Result<Output, Failure>) -> Void
  ...
}

Promise는 단일 값 또는 오류가 포함된 Result타입을 수신하는 클로져 타입이에요

promise를 다시 실행하지않고 결과를 공유하거나 재사용합니다.

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

var subscriptions = Set<AnyCancellable>()
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)
  
//print 
Original
2
finished
Second 2
Second finished

위 코드를 보듯이

Original이라는 출력문이 구독될때마다 실행되지않고

공유되서 한번만 실행되는걸 볼 수 있죠

 

 

Subject프로토콜

외부에서 값을 주입할 수 있는 publisher

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

 

PassthroughSubject

초기값이 필요하지않음

이벤트가 발생하면 요소를 broadcast방식으로 다운스트림 구독자에게 전달하는 Subject

let pass = PassthroughSubject<Int, Never>()

pass.sink {
    print("1:",$0)
}

pass.sink {
    print("2:",$0)
}

pass.send(1)
pass.send(2)
pass.send(3)

//print
1: 1
2: 1
1: 2
2: 2
1: 3
2: 3

 

 

 

CurrentValueSubject

초기값이 필요하고

구독시 최근에 게시한 값 또는 초기값을 얻습니다.

값이 변경될때마다 새 요소를 publish하는 subject입니다.

가장 최근에 publish된 요소를 버퍼에 유지합니다.

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

current.sink {
    print("1:",$0)
}

current.sink {
    print("2:",$0)
}

current.send(1)
current.value = 2

//print
1: 0
2: 0
1: 1
2: 1
1: 2
2: 2

send()와 .value 프로퍼티를 통해서 새로운 값을 전달할 수 있습니다. 

.value프로퍼티를 통해 값에 접근할 수도 있습니다.

print("현재값:", current.value)

하지만 완료이벤트는 send()를 통해서만 전달 할 수 있습니다.

current.value = .finished // error
current.send(completion: .finished)