Swift) AsyncStream 정리하기 (feat: Sequence, IteratorProtocol)

일본 어딘가의 녹차마을

안녕하세요. 후르륵짭짭입니다.

한달만에 글을 작성하네요 ㅋㅋㅋㅋㅋ.

최근에 여행도 다녀오고 여러가지로 바쁜 일이 많아서 공부할 시간이나 글을 작성할 시간이 별로 없었네요.

지금 늦은 시간인데,,, 오늘 안 적으면 또 다음주로 미뤄질 것 같아서 적어보려고 합니다.

2023.06.12 - [Xcode/Swift - PlayGround] - Swift) Async - Await 정리하기 #2 (Async let, withTaskGroup, Task)

이전에 Async Await에 대해 정리한 적이 있습니다.

근데 이것도 하나의 연장선으로 비동기 Sequence인 Async Stream에 알아보려고 합니다.

 

** Sequence **

일단 Async Stream을 알기 전에 Sequence 부터 알아야하는데, 

Sequence는 말 그대로 순차적으로 작업하기 위한 프로토콜 입니다.

평소에 For ... Loop을 많이 사용하는데, 이것을 사용할 수 있게 해주는 프로토콜이죠.

Sequence 프로토콜은 IteratorProtocol이랑 함께 사용되는데, 아래 코드를 보면 이해가 될 겁니다.

struct SequenceItem : Sequence {

    let startPoint : Int
    let endPoint : Int

	//makeIterator 부분에 IteratorProtocol을 반환해준다. (필수 구현)
    func makeIterator() -> some IteratorProtocol {
        return SequenceIteratorProtocol(startPoint: startPoint, endPoint: endPoint, emoji: "🟡")
    }
}

struct SequenceIteratorProtocol : IteratorProtocol {

    var startPoint : Int
    let endPoint : Int
    let emoji : String?
	
    //next에서 반환하고자 하는 Item을 준다. (필수 구현)
    mutating func next() -> String? {
        if startPoint == endPoint {
            return nil
        }
        startPoint += 1
        return "\(startPoint)" + (emoji ?? "")
    }

}

for item in SequenceItem(startPoint: 0, endPoint: 10) {
    print(item)
}

//결과 
1🟡
2🟡
3🟡
4🟡
5🟡
6🟡
7🟡
8🟡
9🟡
10🟡

 

startPoint 부터 endPoint 까지 숫자를 String으로 출력하는 Sequence 입니다.

SequenceItem이 Sequence 프로토콜을 준수 했기 때문에 For ... Loop를 사용할 수 있는 겁니다.

그럼 Sequence 프로토콜을 준수하면 어떤 점이 좋을까요?

바로 map, drop, filter, flatmap 등의 함수형 프로그래밍을 사용할 수 있게 됩니다. 

 

** AsyncSequence ** 

AsyncSequence도 마찮가지로 Sequence 처럼 순차적으로 값을 리턴하는 것이지만

비동기로 값을 리턴한다는 차이가 있습니다. 

struct AsyncSequeceItem : AsyncSequence {

    typealias Element = String

    func makeAsyncIterator() -> AsyncSequeceIterator {
        return AsyncIterator(startPoint: 0, endPoint: 10)
    }

}

struct AsyncSequeceIterator : AsyncIteratorProtocol {

    var startPoint : Int
    let endPoint : Int

    func returnRandomNum() async -> String{

        return await withCheckedContinuation({ continuation in
            DispatchQueue.main.asyncAfter(deadline: .now() + 1, execute: {
                continuation.resume(returning: "\(startPoint) \(Date())")
            })
        })

    }

    mutating func next() async -> String? {

        if startPoint >= endPoint {
            print("finish")
            return nil
        }

        startPoint += 1

        return await returnRandomNum()
    }

}

for await item in AsyncSequeceItem() {
    print(item)
}

//결과
1 2023-07-23 15:21:50 +0000
2 2023-07-23 15:21:51 +0000
3 2023-07-23 15:21:52 +0000
4 2023-07-23 15:21:53 +0000
5 2023-07-23 15:21:54 +0000
6 2023-07-23 15:21:55 +0000
7 2023-07-23 15:21:57 +0000
8 2023-07-23 15:21:58 +0000
9 2023-07-23 15:21:59 +0000
10 2023-07-23 15:22:00 +0000

Sequence와 이름만 살짝 다를 뿐 사용하기 위해서는

AsyncSequence와 AsyncIteratorProtocol를 준수한는 객체가 있어야한다는 것은 똑같죠.

결과 부분을 보면 출력되는 시간이 1초 차이가 발생합니다. 

하지만 매번 이렇게 구현하기 힘드니 AsyncStream이라는 것이 등장 했습니다 .

 

** AsyncStream ** 

AsyncStream은 Gernic Type으로 되어 있는데, 이 Element는 Sendable로 되어 있습니다.

어떻게 사용하는지 확인하기 위해서 숫자를 랜덤하게 1초마다 생성하는 객체를 먼저 만들겠습니다.

class NumberGenerator {

    var timer : Timer?
    var priceHandler: (Int) -> Void = { _ in }

    init(){
        self.timer = Timer.scheduledTimer(withTimeInterval: 1, repeats: true, block: { [weak self] _ in
            guard let self = self else {return}
            self.getPrice()

        })
    }

    private func getPrice() {
        priceHandler(Int.random(in: 0...100))
    }

    func stopUpdating() {
        timer?.invalidate()
    }

}

요렇게 Timer로 1초간 값을 반환해주는 녀석을 만들어 줍니다 .

let stream = AsyncStream<String>{ continuation in
    var count = 0
    let obj = NumberGenerator()

	//상태에 따라 결과가 여기로 온다.
    continuation.onTermination = { result in
        switch result {
        case .finished:
            print("Stream is Finished")
        case .cancelled:
            print("Number is even")
        }
    }

    obj.priceHandler = { number in

        if number % 2 == 0 {
        	//짝수인 경우 반환 안 함
            continuation.onTermination?(.cancelled)
        }
        else {
        	//홀수인 경우 값 반환
            continuation.yield("\(count) : \(number)")
        }

        count += 1
        if count == 10 {
            obj.stopUpdating()
            //Stream 종료
            continuation.finish()
        }
    }
}

for await item in stream.map({"Hello \($0)"}) {
    print(item)
}

//결과
Hello 0 : 11
Number is even
Number is even
Hello 3 : 41
Hello 4 : 27
Number is even
Hello 6 : 37
Number is even
Number is even
Number is even
Stream is Finished

그리고 AsyncStream을 만들어 주는데 Element를 String으로 해주고 

NumberGenerator 객체의 priceHandler로 나온 값을 반환하는 겁니다.

AsyncStream에는 여러가지 함수가 존재하는데 

Sendable 값을 반환하는 yield :

Stream을 종료하는 finish :

상태를 직접 설정할 수 있는 onTermination : 

finished 와 cancelled 이 존재하는데, cancelled은 Stream을 종료하지 않고 무시하게 되고 finished는 Stream을 종료하게 됩니다.

이렇게 있습니다.

 

이렇게 AsyncStream을 사용하면 Sequence를 구현하지 않더라도 비동기적인 Sequence를 만들 수 있습니다.

그래서 아래 처럼 고차함수를 사용할 수 있게 됩니다.

for await item in stream.map({"Hello \($0)"}) {
    print(item)
}

 

** 참고 사이트 ** 

https://ios-daniel-yang.tistory.com/44

[iOS/Swift] 프로토콜 AsyncSequence (비동기 시퀀스)

 

https://ios-daniel-yang.tistory.com/45

[iOS/Swift] AsyncStream / AsyncThrowingStream

 

https://zeddios.tistory.com/1339

AsyncSequence

 

https://zeddios.tistory.com/1340

Swift ) Sequence