What is AsyncSequence?

• 예제 코드

@main
struct QuakesTool {
	static func main() async throws {
		let endpointURL = URL(string: "<https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.csv>")!
		
		// lines가 AsyncSequence
		// 맨 위 헤더는 드롭
		for try await event in endpointURL.lines.dropFirst() {
			let value = event.split(separator: ",")
			let time = values[0]
			let latitude = values[1]
			let longtitude = values[2]
			let magnitude = values[4]
			
			print("Magnitude \\(magnitude) on \\(time) at \\(latitude) \\(longtitude)")
		}
	}
}

• 기본 sequence와의 차이는 Async라는 것뿐이다.
  • 다만 가져오다가 오류가 날 수 도 있기 때문에 throw가 일어날 수 있다.
  • for-in문은 Iterator에서 값이 나올 때마다 실행된다.
  • iterator가 nil을 반환하거나(정상종료), throw가 나면(비정상 종료) 종료된다.
• 기존 for-in에 대한 이해

// 이러한 for문이
for quake in quakes {
	if quake.magnitude > 3 {
		displaySignificantEarthquake(quake)
	}
}

// 이렇게 바뀐다.
var iterator = quakes.makeIterator()
while let quake = iterator.next() {
	if quake.magnitude > 3 {
		displaySignificantEarthquake(quake)
	}
}

• asyncSequence에서도 똑같다. async를 위한 새로운 for-in 패턴이 생겼을 뿐

// 이게
for await quake in quakes {			
	if quake.magnitude > 3 {
		displaySignificantEarthquake(quake)
	}
}

// 이렇게 바뀐다
var iterator = quakes.makeAsyncIterator()
while let quake = await iterator.next() {
	if quake.magnitude > 3 {
		displaySignificantEarthquake(quake)
	}
}

• break, continue도 동일하게 동작한다.

for await quake in quakes {
	if quake.location == nil {
		break
	}

	if quake.magnitude > 3 {
		displaySignificantEarthquake(quake)
	}
}

for await quake in quakes {
	if quake.depth > 5 {
		continue
	}

	if quake.magnitude > 3 {
		displaySignificantEarthquake(quake)
	}
}

• 다만 에러 가능성이 있으면 try를 써줘야 한다.

do {
	for try await quake in quakeDownload {
		...
	}
} catch {
	...
}

• 루프를 개별적으로 돌리고 싶으면, async로 감싸라.

async {
	for await quake in quakes {
		...
	}
}

async {
	do {
		for try await quake in quakeDownload {
			...
		}
	} catch {
		...
	}
}

• 취소도 간편하다.

let iteration1 = async {
	for await quake in quakes {
		...
	}
}

let iteration2 = async {
	do {
		for try await quake in quakeDownload {
			...
		}
	} catch {
		...
	}
}

iteration1.cancel()
iteration2.cancel()

Usage and APIs

• FileHandle에서 파일 읽기

 // FileHandle
public var bytes: AsyncBytes

for try await line in FileHandle.standardInput.bytes.lines {

}

• URL에서 읽어오기

// URL
public var resourcesBytes: AsyncBytes
public var lines: AsyncLineSequence<AsyncBytes>

let url = URL(fileURLWithPath: "/tmp/somefile.txt")

for try await line in url.lines {

}

• URLSession

func bytes(from: URL) async throws -> (AsyncBytes, URLResponse)
func bytes(from: URLReqeust) async throws -> (AsyncBytes, URLResponse)

let (bytes, response) = try await URLSesssion.shared.bytes(from: url)
guard let httpResponse = response as? HTTPURLResponse,
			httpResponse.statusCode == 200 else {
			throw MyNetworkingError.invalidServiceResponse
}

for try await byte in bytes {

}

• Notification

public func notifications(named: Notification.Name, object: AnyObject) -> Notifications

let center = NotificationCenter.default

let notification = await center.notifications(named: .NSPersistentStoreRemoteChange).first {
	$0.userInfo[NSStoreUUIDKey] == storeUUID
}

• 시퀀스 자체를 변경하는 로직들은 Sequence와 공유한다.

Adopting AsyncSequence

• 값을 돌려받지 않거나, 새로운 값이 발생했음을 알려주는 정도의 callback이나 delegate는 우선적으로 AsyncSequence 적용 대상이다.

// 기존 패턴
class QuakeMonitor {
	var quakeHandler: (Quake) -> Void
	func startMonitoring()
	func stopMonitoring()
}

let monitor = QuakeMonitor()
monitor.quakeHandler = { quake in
	// ...
}

monitor.startMonitoring()

monitor.stopMonitoring()

// async로 변경
let quakes = AsyncStream(Quake.self) { continuation in
	let monitor = QuakeMonitor()
	monitor.quakeHandler = { quake in
		continuation.yield(quake)
	}
	
	continuation.onTermination = { _ in
		monitor.stopMonitoring()
	}

	monitor.startMonitoring()
}

// ASyncStream 생성자

public struct AsyncStream<Element>: AsyncSequence {
		public init(
					_ elementType: Element.Type = Element.self,
					maxBufferedElements limit: Int = .max,
					_ build: (Continuation) -> Void
		)
}

// 에러를 던질 수 있는 AsyncStream
public struct AsyncThrowingStream<Element>: AsyncSequence {
			public init(
					_ elementType: Element.Type = Element.self,
					maxBufferedElements limit: Int = .max,
					_ build: (Continuation) -> Void
		)
}