Publishers and Subscribers (Combine)

publisher는 데이터를 publish하고 subscriber를 구현하여 값의 변화를 감지합니다

 

 

Combine 이란?

• Combine 프레임워크는 시간에 따른 값의 변화를 처리하기 위한 선언적 Swift API 를 제공합니다.
• 여기서 말하는 값은 여러 비동기 이벤트를 의미할 수 있습니다. 
• Combine 은 publishers 를 만들어 값의 변화를 subscriber 들에게 노출시킵니다
• subscriber 들은 publisher 로부터 값을 받습니다

Combine의 원리

publisher는 말 그대로 publish(출판,계시)하다는 의미며 publisher가 publish 한 이벤트, 값을 subscribe(구독)하여 값의 변화를 인지하고 대응할 수 있게 합니다. 이 publish 와 subscriber 가 Combine의 핵심입니다.

 

subscriber 는 publisher 에서 발생되는 값을 기다리다 publisher 가 특정 값을 전파하면 subscriber 는 값이 전파되었음을 인식하여 원하는 특정 동작을 수행할 수 있게 됩니다.

 

Foundation 에서 사용되는 Combine

• Timer
• NotificationCenter
• URLSession

 

Combine을 사용해야되는 이유

🟢Combine 을 사용하게 된다면 데이터 흐름을 조작할 수 있으며 코드를 더 읽기 쉽고 유지보수하기 쉽게 만들 수 있습니다

 

왜 @State를 놨두고 굳이 복잡한 Publisher, Subscriber 를 사용해야되나 생각이 오랫동안 들었습니다. 알아낸 것은 Combine을 사용하게 되면 data flow 즉 복잡한 데이터의 흐름을 좀 더 조작하고 또는 결합할 수 있다는 것입니다.

 

예를 들어 다음과 같은 TextField 눌렀을 때 text1의 값과 바뀐 값을 text2에 넣는 코드가 있습니다. 

struct ContentView: View {
    @State var text1 = ""
    @State var text2 = ""
    
    var body: some View {
        VStack {
            TextField("1번째", text: $text1)
            Text(text2)
        }
        .onChange(of: text1) { oldValue, newValue in
            text2 = newValue
        }
    }
}

이때 만약 textfield의 값을 다른 변수에 넣는 시간을 바꾸고 싶다면 데이터의 흐름을 조작해야 합니다.

이때 Combine 의 debounce 를 사용하면 구현할 수 있습니다.

import SwiftUI
import Combine

class TextFieldViewModel: ObservableObject {
    @Published var text1 = ""
    @Published var text2 = ""
    
    var cancellables = Set<AnyCancellable>()
    
    init() {
        addTextFieldSubcriber()
    }
    
    func addTextFieldSubcriber() {
        $text1
            .debounce(for: .seconds(0.5), scheduler: DispatchQueue.main)
            .sink { [weak self] value in
                self?.text2 = value
            }
            .store(in: &cancellables)

    }
}

struct ContentView: View {
    @StateObject var vm = TextFieldViewModel()

    var body: some View {
        VStack {
            TextField("1번째", text: $vm.text1)
            Text(vm.text2)
        }
    }
}

위에서 Publisher는 직접 cancle()를 사용하여 취소하기 전까지 계속 살아있다는 특징이 있습니다. 다시 말해 유튜버가 계정을 삭제하기 전까지 유튜버가 유튜브 활동을 계속 한다는 의미, 영상을 올리면 구독자한테 알림이 가서 구독자가 영상을 확인하고..

 

그러면 이제 Publisher의 종류를 알아보겠습니다

Publisher의 종류

✓ Combine 프레임워크에서 제공하는 Publisher는 여러개의 종류가 있으며 역할도 제각각입니다.

✓ Publisher들은 데이터의 publish 를 담당하며 Subscriber 에게 값을 전달하는 역할을 합니다.

 

Just 로 Subscriber 에 값을 한 번만  전달

Just는 주어진 값으로 Publisher를 생성합니다. 이 값은 단일 값으로 고정되며, Subscriber에게 값이 한 번만 전달되는 특징이 있습니다.

class TestSubscriber: Subscriber {
    typealias Input = String
    typealias Failure = Never
    
    func receive(subscription: any Subscription) {
        print("구독 시작했을 때 실행되는 함수로 item을 요청할 수 있다")
        subscription.request(.unlimited)
    }
    
    func receive(_ input: String) -> Subscribers.Demand {
        print("받은 값:\(input)")
        return .none // publisher에 요청할 item의 개수 0
//      return .unlimited // publisher에 요청할 item의 개수 제한 없음
//      return .max(0) // publisher에 요청할 item의 최대 개수 제한
    }
    
    func receive(completion: Subscribers.Completion<Never>) {
        print("값 받기 완료")
    }
}

let publisher = Just("저스트")
publisher.subscribe(TestSubscriber()) // Just Publisher에 Subscriber 구독시키기

 

Publishers.Sequence

Publishers.Sequence는 시퀀스에 포함된 요소들을 Publish하는 Publisher입니다. 배열이나 시퀀스와 같은 컬렉션을 기반으로 합니다.

사용할 때는 Publish할 데이터의 타입과 에러 타입을 <data type, error type>지정합니다. Never는 절대 에러가 발생하지 않는다는 의미입니다.

let publisher = Publishers.Sequence<[String], Never>(sequence: ["1", "2", "3", "4", "5"])
publisher.subscribe(TestSubscriber())

 

Publisher.Future 비동기 작업 결과 처리하기

Future는 비동기 작업의 결과를 담고 있는 Publisher입니다. Future를 생성할 때 클로저를 전달하며, 클로저는 값을 생성하고 해당 값을 Future Publisher가 Publish합니다. 비동기 작업 데이터를 처리하므로 네트워크 요청의 응답을 처리하는 과정에서 사용될 수 있습니다.

let publisher = Future<String, Never> { completion in
    completion(.success("Future Publisher에서 publish한 값"))
}
publisher.subscribe(TestSubscriber())

// Combine을 사용하여 MKDirections 요청을 처리하는 메서드
func getDirectionsPublisher(request: MKDirections.Request) -> AnyPublisher<MKDirections.Response, Error> {
    Future { promise in
        let directions = MKDirections(request: request)
        directions.calculate { response, error in
            if let response = response {
                promise(.success(response))
            } else if let error = error {
                promise(.failure(error))
            }
        }
    }
    .eraseToAnyPublisher()
}

 

Publisher.Empty 테스트 목적으로 사용

Publisher.Empty는 아무 값도 발생시키지 않는 Publisher로 주로 테스트 목적으로 사용될 수 있습니다.

let publisher = Empty<String, Never>()
publisher.subscribe(TestSubscriber())

 

CurrentValueSubject 현재 값 유지하면서 새로운 값 publish

🔵 현재 값을 유지하면서 이후 값을 Publish하는 Publisher입니다.

 

내부적으로 현재 값을 유지하면서 새로운 값을 받으면 그 값을 Publish 합니다. CurrentValueSubject는 가장 마지막으로 publish한 elemoent의 버퍼를 유지합니다

let publisher = CurrentValueSubject<String, Never>("CurrentValueSubject에서 Publish한 값")
publisher.subscribe(TestSubscriber())

 

아래 결과에서 나온 결과를 확인한 후 CurrentValueSubject Publisher을 Subscribe를 했을 때에는 receive(completion:)이 실행되지 않는 것을 확인할 수 있습니다.

 

즉 CurrentValueSubject Publisher가 Publish 끝나는 시점을 정의해야 하며 끝나는 시점을 만들지 않는다면 Subscriber는 Publisher에 연결되어 계속 값의 변화를 감시하게 됩니다.

func receive(completion: Subscribers.Completion<Never>) {
    print("값 받기 완료")
}

정리

• 초기값이 있음
• 구독자가 생기면 즉시 현재 값을 전달
• 상태를 계속 저장하므로, 더 많은 메모리를 사용할 수 있음

Future, Empty, Just와 같은 Publisher는 구독자들이 구독하는 시점으로 한 번 값을 Publish한 후 관계를 끊지만
CurrentValueSubject Publisher는 구독자들이 구독하는 시점으로부터 계속 Publisher와 Subscribe의 관계가 연결됩니다.

 

import Combine

class ExampleViewModel {
    var textSubject = CurrentValueSubject<String, Never>("초기값")
    
    var cancellable: AnyCancellable?

    init() {
        // textSubject가 변경될 때마다 구독자가 해당 값을 받는다.
        cancellable = textSubject
            .sink { value in
                print("수신된 값: \(value)")
            }
        
        // 값이 변경됨
        textSubject.send("새로운 값")
    }
}

// ViewModel 생성
let viewModel = ExampleViewModel()
// 초기값인 "초기값"이 출력되고, 이후에 "새로운 값"이 출력됨

viewModel.textSubject.send("hi")
print(viewModel.textSubject.value)
viewModel.textSubject.send("hi2")
print(viewModel.textSubject.value)

//수신된 값: 초기값
//수신된 값: 새로운 값
//수신된 값: hi
//hi
//수신된 값: hi2
//hi2

PassthroughSubject 값 유지 없이 publish

🔵PassthroughtSubject 란 내부 상태를 유지하지 않고, Subscriber들에게 값을 전달하는 Publisher입니다.

 

즉 값이 발생할 때마다 즉시 전달됩니다. 즉CurrentValueSubject Publisher 와 달리 PassthroghtSubject는 초기값이 없어 Subscriber들이 Subscribe할 때 전송하는 메시지가 없습니다. 또한 초기값 또는 버퍼가 없기 때문에 메모리 효율이 조금 더 좋습니다. 추가로 subscriber가 없거나 현재 demand가 0이면 value를 삭제합니다.

 

정리

• 초기값이 없음
• 값이 Publish 될 때에만 Subscriber 들에게 값을 전달
• Subscribe 시점 이전에 Publish 된 값은 Subscriber 가 받을 수 없음 (내부 상태를 유지하지 않기 때문)
• 상태를 계속 저장하지 않기 때문에 메모리 사용량이 적음

let publisher = PassthroughSubject<String, Never>()
publisher.subscribe(TestSubscriber())

import Combine

class ExampleViewModel {
    var textSubject = PassthroughSubject<String, Never>()
    
    var cancellable: AnyCancellable?

    init() {
        // textSubject가 변경될 때마다 구독자가 해당 값을 받습니다
        cancellable = textSubject
            .sink { value in
                print("수신된 값: \(value)")
            }
        
        // 구독 이후 값을 publish (이 시점에 구독자가 있으므로 값이 전달됨)
        textSubject.send("새로운 값")
    }
}

// ViewModel 생성
let viewModel = ExampleViewModel()

// "hi" 값을 Puslish, 구독자가 sink 로 값을 대기하고 있어서 값이 전달됨
viewModel.textSubject.send("hi")
// PassthroughSubject는 상태를 저장하지 않으므로, 과거 값을 추적하거나 출력할 방법이 없음
// print(viewModel.textSubject.value) 불가능

 

Subject 

• Subscriber들에게 send를 사용하여 데이터 스트림에 값을 주입할 수 있는 Publisher입니다.
• send는 3가지 메서드가 제공됩니다
   - send(completion:): Subscrber에게 completion signal을 보내는 메서드입니다.
   - send(input:) Subscriber에게 값을 보내는 메서드입니다.
   - send(subscription: ): Subscriber에게 구독을 시작했을 때 보내는 메서드로 Subscriber에 구현되어 있습니다.

let publisher = PassthroughSubject<String, Error>()
var cancellables = Set<AnyCancellable>()
publisher.sink { completion in
    switch completion {
    case .finished:
        print("완료")
        break
    case .failure(let error):
        print("발생한 에러:\(error.localizedDescription)")
    }
} receiveValue: { value in
    print("Publisher로부터 받은 값:\(value)")
}.store(in: &cancellables)
publisher.send(subscription: Subscriptions.empty)
publisher.send("테스트1")
publisher.send(completion: .finished)

 

Publishers.Merge

여러개의 Publisher를 병합하여 하나의 Publisher를 만드는 Publisher입니다. 모든 입력 Publisher들로부터 Publish된 값들을 순서대로 전달합니다.

let publisher1 = Just("저스트")
let publisher2 = CurrentValueSubject<String, Never>("CurrentValueSubject에서 Publish한 값")
let mergedPublishers = Publishers.Merge(publisher1, publisher2)
mergedPublishers.subscribe(TestSubscriber())

Publishers.Zip

Publishers.Zip은 여러개의 Publisher에서 값을 받아와서 각 Publisher에서 값을 번갈아가며 받아와 두 Publisher의 값들을 순서대로 결합하여 새로운 값을 Publish하는 Publisher를 생성합니다.

let publisher1 = CurrentValueSubject<String, Never>("CurrentValueSubject1에서 보내는 1번째 값")
let publisher2 = CurrentValueSubject<String, Never>("CurrentValueSubject2에서 보내는 1번째 값")

let zippedPublisher = Publishers.Zip(publisher1, publisher2)
    .map { value1, value2 in
        return "\(value1) - \(value2)"
    }

zippedPublisher.subscribe(TestSubscriber())
publisher1.send("CurrentValueSubject1에서 보내는 2번째 값")
publisher2.send("CurrentValueSubject2에서 보내는 2번째 값")
publisher1.send("CurrentValueSubject1에서 보내는 3번째 값")
//publisher2.send("CurrentValueSubject2에서 보내는 3번째 값")

 

여기서 CombineLates와 중요한 차이점은 Publisher의 값이 무조건 모두 변경돼야 Zip Publisher에서 값을 받습니다.

예를 들어 다음과 같이 1개의 Publisher에서는 2개번 다른 Publisher에서는 3번하면 2번째까지만 값을 받습니다.

let publisher1 = CurrentValueSubject<String, Never>("CurrentValueSubject1에서 보내는 1번째 값")
let publisher2 = CurrentValueSubject<String, Never>("CurrentValueSubject2에서 보내는 1번째 값")

let zippedPublisher = Publishers.Zip(publisher1, publisher2)
    .map { value1, value2 in
        return "\(value1) - \(value2)"
    }

zippedPublisher.subscribe(TestSubscriber())
publisher1.send("CurrentValueSubject1에서 보내는 2번째 값")
publisher2.send("CurrentValueSubject2에서 보내는 2번째 값")
publisher1.send("CurrentValueSubject1에서 보내는 3번째 값")
//publisher2.send("CurrentValueSubject2에서 보내는 3번째 값")

 

Publishers.CombineLastes

여러 개의 Publisher로부터 값을 받아와 모든 Publisher들로부터 새로운 값이 Publish될 때마다 값을 결합하여 새로운 값을 Publish하는 Publisher입니다

let publisher1 = CurrentValueSubject<String, Never>("CurrentValueSubject1에서 보내는 1번째 값")
let publisher2 = CurrentValueSubject<String, Never>("CurrentValueSubject2에서 보내는 1번째 값")

let combinedPublisher = Publishers.CombineLatest(publisher1, publisher2)
    .map { value1, value2 in
        return "\(value1) - \(value2)"
    }

combinedPublisher.subscribe(TestSubscriber())
publisher1.send("CurrentValueSubject1에서 보내는 2번째 값")
publisher2.send("CurrentValueSubject2에서 보내는 2번째 값")

 

Publisher를 만들 때 주의할 점

Publisher에서 Publish하는 Output과 Subscirber에서 Publish하는 Input 값의 타입은 무조건 일치해야합니다.

그렇지 않으면 아래와 같이 에러가 발생합니다.

 

이때 알아보다가 publisher의 메서드를 사용하여 Publisher에서 구독된 Subscriber들에게 값을 한번에 보낼 수 있는 것을 확인하였습니다.

 

 

Subscriber

Subscriber는 아래와 같이 직접 구현해도 되지만 다른 방법으로 만드는 방법도 있습니다.

class TestSubscriber: Subscriber {
    typealias Input = String
    typealias Failure = Never
    
    func receive(subscription: any Subscription) {
        print("구독 시작했을 때 실행되는 함수로 item을 요청할 수 있다")
        subscription.request(.unlimited)
    }
    
    func receive(_ input: String) -> Subscribers.Demand {
        print("받은 값:\(input)")
        return .none // publisher에 요청할 item의 개수 0
//        return .unlimited // publisher에 요청할 item의 개수 제한 없음
//        return .max(0) // publisher에 요청할 item의 최대 개수 제한
    }
    
    func receive(completion: Subscribers.Completion<Never>) {
        print("값 받기 완료")
    }
}

 

Sink

다른 방법 중 하나는 .sink를 통해서 Subscriber를 만드는 것 입니다.

sink(receiveValue:)

sink(receiveValue:)는 반환값으로 AnyCancellable 를 반환하는데 이 AnyCancellable은 여러개의 구독을 한꺼번에 묶어거나 취소하여 관리할 수 있습니다. 이 메서드는 Publisher의 Failure 가 Never인 경우에만 사용 가능합니다. 즉 실패를 확실히 안할 때 사용합니다.

let publisher = CurrentValueSubject<String, Never>("CurrentValueSubject1에서 보내는 1번째 값")
let subscriber = publisher.sink { value in
    print("Publisher로부터 받은 값:\(value)")
}
publisher.send("CurrentValueSubject1에서 보내는 2번째 값")

 

sink(receiveCompletion:receiveValue:)

sink(receiveCompletion:receiveValue:)는 closure에서 새로운 값이나 종료 이벤트에 대해 처리합니다. 이는 Failure 가 존재하므로 

let publisher = PassthroughSubject<String, Error>()
var cancellables = Set<AnyCancellable>()
publisher.sink { completion in
    switch completion {
    case .finished:
        break
    case .failure(let error):
        print("발생한 에러:\(error.localizedDescription)")
    }
} receiveValue: { value in
    print("Publisher로부터 받은 값:\(value)")
}.store(in: &cancellables)
publisher.send("테스트1")
publisher.send(completion: .finish) // finish
publisher.send(completion: .failure(let error)) // error

이때 Just Publisher는 한번 사용되고 종료되는 Publisher라서 그렇지 만일 CurrentValueSubject처럼 연결이 계속 되는 Publisher같은 경우는 다음과 같이 send(completion:)를 사용해서 구독자한테 구독을 끝낸다는 것을 보냅니다. 유튜브를 접겠다라는 의미..

let publisher = CurrentValueSubject<String, Never>("CurrentValueSubject에서 보내는 1번째 값")
let subscriver = publisher.sink { completion in
    print(completion)
} receiveValue: { value in
    print("Received value: \(value)")
}
publisher.subscribe(TestSubscriber())
publisher.send(completion: .finished)
publisher.send("hi")

 

여기서 또 하나의 포인트를 찾을 수 있었는데 Subscriber 클래스에 정의된 receive 함수보다subscriber.sink에 completion이 먼저 수행된다는 것을 확인할 수 있습니다.

 

AnyCancellable

sink의 반환값으로 받는 AnyCancellable은 현재 구독중인 Publisher의 구독을 취소할 수 있습니다, 그렇기 때문에 cancel() 메서드를 사용하여 구독을 취소하게 되면 이후 Publisher가 Publish를 아무리 해도 구독자는 그 값을 들을 수 없게 됩니다.

let publisher = CurrentValueSubject<String, Never>("CurrentValueSubject1에서 보내는 1번째 값")
let subscriber = publisher.sink { value in
    print(value)
}

publisher.send("CurrentValueSubject1에서 보내는 2번째 값")
subscriber.cancel()
publisher.send("CurrentValueSubject1에서 보내는 3번째 값")

 

AnyCancellable 저장하기

AnyCancellable는 변수에 그냥 sink를 사용하여 저장하는 방법과 아니면 Set 타입을 사용하여 저장하는 방법 2가지가 있습니다.

let publisher = PassthroughSubject<String, Never>()
let subscriber = publisher.sink { value in
    print("Publisher로부터 받은 값:\(value)")
}

var cancellables = Set<AnyCancellable>()
publisher.sink { value in
    print("Publisher로부터 받은 값:\(value)")
}.store(in: &cancellables)

Set 타입으로 저장하게 되면 여러개의 AnyCancellable를 중복없이 저장할 수 있다는 장점이 있습니다.

store

위에서 사용한 store란 cancellables 에 sink로부터 반환되는 AnyCancellable 를 저장한다는 의미로 해당 AnyCancellable를 관리할 수 있습니다. 

.store(in: &cancellables)

 

이후에 다음과 같이 publisher에서 publish하는 값의 감시를 취소하여 pubilsher에서 publish하더라도 데이터를 더이상 감지하지 않을 수 있습니다. 유튜브로 비유하면 구독자가 유튜버의 구독을 취소하는 상황이라고 볼 수 있겠습니다.

for i in cancellables {
    i.cancel()
}