SwiftでAsyncSequenceを使った非同期データストリーム処理方法を徹底解説

import Foundation

// 非同期に整数を生成するAsyncSequence
struct AsyncIntSequence: AsyncSequence {
    typealias Element = Int

    // AsyncIteratorを定義
    struct AsyncIterator: AsyncIteratorProtocol {
        var current = 0

        // 非同期で次の要素を返す
        mutating func next() async -> Int? {
            // デモ用に少し待機
            await Task.sleep(1_000_000_000) // 1秒待機
            current += 1
            return current <= 5 ? current : nil
        }
    }

    // AsyncIteratorを返す
    func makeAsyncIterator() -> AsyncIterator {
        return AsyncIterator()
    }
}

// AsyncSequenceを使って非同期処理を実行
func processAsyncSequence() async {
    let sequence = AsyncIntSequence()

    // for awaitループを使って非同期にデータを処理
    for await number in sequence {
        print("Received number: \(number)")
    }
    print("All numbers processed!")
}

// 実行
Task {
    await processAsyncSequence()
}

import Foundation

// 非同期ストリームを作成
func createAsyncStream() -> AsyncStream<Int> {
    return AsyncStream { continuation in
        Task {
            for i in 1...5 {
                // 1秒ごとに値をストリームに追加
                await Task.sleep(1_000_000_000)
                continuation.yield(i)
            }
            // ストリームの終了を通知
            continuation.finish()
        }
    }
}

// AsyncStreamのデータを処理
func processAsyncStream() async {
    let stream = createAsyncStream()

    // for awaitを使ってストリームからデータを受け取る
    for await value in stream {
        print("Received value: \(value)")
    }
    print("Stream processing complete!")
}

// 実行
Task {
    await processAsyncStream()
}

import Foundation

// 非同期でAPIからデータを取得するAsyncSequence
struct APIFetcher: AsyncSequence {
    typealias Element = Data

    // 非同期イテレータを定義
    struct AsyncIterator: AsyncIteratorProtocol {
        var urls: [URL]
        var currentIndex = 0

        mutating func next() async -> Data? {
            guard currentIndex < urls.count else { return nil }
            let url = urls[currentIndex]
            currentIndex += 1

            do {
                // URLから非同期でデータを取得
                let (data, _) = try await URLSession.shared.data(from: url)
                return data
            } catch {
                print("Error fetching data: \(error)")
                return nil
            }
        }
    }

    // URLリストを渡してAsyncIteratorを生成
    let urls: [URL]
    func makeAsyncIterator() -> AsyncIterator {
        return AsyncIterator(urls: urls)
    }
}

// 非同期データの取得処理
func fetchAPIData() async {
    let urls = [
        URL(string: "https://jsonplaceholder.typicode.com/todos/1")!,
        URL(string: "https://jsonplaceholder.typicode.com/todos/2")!
    ]

    let fetcher = APIFetcher(urls: urls)

    for await data in fetcher {
        if let json = try? JSONSerialization.jsonObject(with: data, options: []) {
            print("Received JSON: \(json)")
        }
    }
    print("All data processed!")
}

// 実行
Task {
    await fetchAPIData()
}

import Foundation

// ファイルから非同期に行を読み込むAsyncSequence
struct AsyncFileReader: AsyncSequence {
    typealias Element = String

    // AsyncIteratorを定義
    struct AsyncIterator: AsyncIteratorProtocol {
        let fileHandle: FileHandle
        let delimiter = "\n".data(using: .utf8)!

        mutating func next() async -> String? {
            while let chunk = try? fileHandle.read(upToCount: 1024), !chunk.isEmpty {
                if let line = String(data: chunk, encoding: .utf8) {
                    return line
                }
            }
            return nil
        }
    }

    // ファイルハンドルを渡してAsyncIteratorを生成
    let fileHandle: FileHandle
    func makeAsyncIterator() -> AsyncIterator {
        return AsyncIterator(fileHandle: fileHandle)
    }
}

// 非同期でファイルの内容を処理
func processFile() async {
    if let fileHandle = FileHandle(forReadingAtPath: "path/to/your/file.txt") {
        let reader = AsyncFileReader(fileHandle: fileHandle)

        for await line in reader {
            print("Read line: \(line)")
        }
        print("All lines processed!")
    } else {
        print("Failed to open file.")
    }
}

// 実行
Task {
    await processFile()
}

import Foundation

// 非同期でデータを取得するAsyncSequence
struct APIFetcherWithErrorHandling: AsyncSequence {
    typealias Element = Data

    struct AsyncIterator: AsyncIteratorProtocol {
        var urls: [URL]
        var currentIndex = 0

        mutating func next() async throws -> Data? {
            guard currentIndex < urls.count else { return nil }
            let url = urls[currentIndex]
            currentIndex += 1

            do {
                // 非同期でデータを取得し、エラー処理を実装
                let (data, _) = try await URLSession.shared.data(from: url)
                return data
            } catch {
                // エラー発生時にスロー
                throw error
            }
        }
    }

    let urls: [URL]
    func makeAsyncIterator() -> AsyncIterator {
        return AsyncIterator(urls: urls)
    }
}

// 非同期データ処理とエラーハンドリング
func fetchAPIDataWithHandling() async {
    let urls = [
        URL(string: "https://jsonplaceholder.typicode.com/todos/1")!,
        URL(string: "https://invalid-url")!  // エラーを発生させるURL
    ]

    let fetcher = APIFetcherWithErrorHandling(urls: urls)

    do {
        // エラーをキャッチするdo-catchブロックを使用
        for try await data in fetcher {
            if let json = try? JSONSerialization.jsonObject(with: data, options: []) {
                print("Received JSON: \(json)")
            }
        }
    } catch {
        // エラー発生時の処理
        print("Failed to fetch data: \(error)")
    }
}

// 実行
Task {
    await fetchAPIDataWithHandling()
}

func fetchAPIDataWithResult() async -> Result<Data, Error> {
    let url = URL(string: "https://jsonplaceholder.typicode.com/todos/1")!

    do {
        let (data, _) = try await URLSession.shared.data(from: url)
        return .success(data)
    } catch {
        return .failure(error)
    }
}

// 実行と結果の処理
Task {
    let result = await fetchAPIDataWithResult()
    switch result {
    case .success(let data):
        print("Data received: \(data)")
    case .failure(let error):
        print("Failed to fetch data: \(error)")
    }
}

import Foundation

// カスタムAsyncSequenceの定義
struct CountdownSequence: AsyncSequence {
    typealias Element = Int

    // 非同期イテレータを定義
    struct AsyncIterator: AsyncIteratorProtocol {
        var currentValue: Int

        // 非同期に次の要素を返す
        mutating func next() async -> Int? {
            guard currentValue > 0 else { return nil }

            // 1秒待機してカウントダウンを進める
            await Task.sleep(1_000_000_000) // 1秒待機
            let value = currentValue
            currentValue -= 1
            return value
        }
    }

    // 初期値を指定してイテレータを生成
    let start: Int
    func makeAsyncIterator() -> AsyncIterator {
        return AsyncIterator(currentValue: start)
    }
}

// 非同期カウントダウン処理
func runCountdown() async {
    let countdown = CountdownSequence(start: 5)

    for await value in countdown {
        print("Countdown: \(value)")
    }
    print("Countdown complete!")
}

// 実行
Task {
    await runCountdown()
}

import Foundation

let stream = AsyncStream(Int.self) { continuation in
    for i in 1...5 {
        continuation.yield(i)
    }
    continuation.finish()
}

Task {
    for await value in stream {
        print("Received value: \(value)")
    }
}

import Combine

let publisher = (1...5).publisher

let cancellable = publisher
    .sink(receiveCompletion: { completion in
        print("Completed with: \(completion)")
    }, receiveValue: { value in
        print("Received value: \(value)")
    })

import Foundation

// 非同期でWebSocketからメッセージを受信するAsyncSequence
struct WebSocketMessageSequence: AsyncSequence {
    typealias Element = String

    let webSocket: URLSessionWebSocketTask

    // イテレータを定義
    struct AsyncIterator: AsyncIteratorProtocol {
        let webSocket: URLSessionWebSocketTask

        mutating func next() async -> String? {
            do {
                let message = try await webSocket.receive().get()
                switch message {
                case .string(let text):
                    return text
                default:
                    return nil
                }
            } catch {
                print("Error receiving WebSocket message: \(error)")
                return nil
            }
        }
    }

    func makeAsyncIterator() -> AsyncIterator {
        return AsyncIterator(webSocket: webSocket)
    }
}

// WebSocketからのメッセージを非同期で処理
func handleWebSocketMessages() async {
    let url = URL(string: "wss://your-chat-server.com")!
    let webSocketTask = URLSession.shared.webSocketTask(with: url)
    webSocketTask.resume()

    let messageSequence = WebSocketMessageSequence(webSocket: webSocketTask)

    for await message in messageSequence {
        print("New message: \(message)")
        // UIを更新したり、通知を送信する処理を追加
    }

    print("WebSocket connection closed")
}

// 実行
Task {
    await handleWebSocketMessages()
}

import CoreMotion
import Foundation

// 非同期でモーションセンサーのデータを取得するAsyncSequence
struct MotionDataSequence: AsyncSequence {
    typealias Element = CMAccelerometerData

    let motionManager: CMMotionManager

    // イテレータを定義
    struct AsyncIterator: AsyncIteratorProtocol {
        let motionManager: CMMotionManager

        mutating func next() async -> CMAccelerometerData? {
            guard motionManager.isAccelerometerAvailable else { return nil }

            return await withCheckedContinuation { continuation in
                motionManager.startAccelerometerUpdates(to: .main) { data, error in
                    if let data = data {
                        continuation.resume(returning: data)
                    } else {
                        continuation.resume(returning: nil)
                    }
                }
            }
        }
    }

    func makeAsyncIterator() -> AsyncIterator {
        return AsyncIterator(motionManager: motionManager)
    }
}

// モーションセンサーデータを非同期で処理
func handleMotionData() async {
    let motionManager = CMMotionManager()
    motionManager.accelerometerUpdateInterval = 0.1

    let motionDataSequence = MotionDataSequence(motionManager: motionManager)

    for await data in motionDataSequence {
        print("Accelerometer Data: \(data.acceleration)")
        // リアルタイムでのデータ処理（例：UIの更新、データの記録など）
    }

    print("Motion data processing completed")
}

// 実行
Task {
    await handleMotionData()
}

import Foundation

// 非同期でオーディオストリームデータを取得するAsyncSequence
struct AudioStreamSequence: AsyncSequence {
    typealias Element = Data

    let url: URL

    // イテレータを定義
    struct AsyncIterator: AsyncIteratorProtocol {
        let url: URL

        mutating func next() async -> Data? {
            do {
                let (data, _) = try await URLSession.shared.data(from: url)
                return data
            } catch {
                print("Error fetching audio stream: \(error)")
                return nil
            }
        }
    }

    func makeAsyncIterator() -> AsyncIterator {
        return AsyncIterator(url: url)
    }
}

// オーディオストリームデータを非同期で処理
func handleAudioStream() async {
    let audioStreamURL = URL(string: "https://your-audio-stream-url.com")!
    let audioStreamSequence = AudioStreamSequence(url: audioStreamURL)

    for await audioData in audioStreamSequence {
        print("Received audio data: \(audioData.count) bytes")
        // オーディオデータの再生処理やバッファリング処理を追加
    }

    print("Audio stream completed")
}

// 実行
Task {
    await handleAudioStream()
}

import Foundation

// 非同期でデータを取得するタスク
func fetchDataWithCancellation() async throws -> String {
    for i in 1...5 {
        // キャンセルが要求されたらタスクを終了
        try Task.checkCancellation()

        // デモ用に待機
        await Task.sleep(1_000_000_000) // 1秒待機
        print("Fetched part \(i)")
    }
    return "Data fetched successfully"
}

// キャンセル可能なタスクの処理
func handleFetchData() async {
    let task = Task {
        do {
            let result = try await fetchDataWithCancellation()
            print(result)
        } catch {
            print("Task was cancelled: \(error)")
        }
    }

    // 3秒後にタスクをキャンセル
    await Task.sleep(3_000_000_000)
    task.cancel()
}

// 実行
Task {
    await handleFetchData()
}

import Foundation

// 非同期でネットワークリクエストを再試行する関数
func fetchDataWithRetry(attempts: Int) async throws -> String {
    var currentAttempt = 0

    while currentAttempt < attempts {
        do {
            // 実際のデータ取得をシミュレーション
            if currentAttempt < 2 {
                throw URLError(.badServerResponse) // 初回は失敗するシミュレーション
            }
            return "Data fetched on attempt \(currentAttempt + 1)"
        } catch {
            currentAttempt += 1
            print("Attempt \(currentAttempt) failed, retrying...")
            if currentAttempt == attempts {
                throw error
            }
            // 再試行前に少し待機
            await Task.sleep(2_000_000_000) // 2秒待機
        }
    }

    throw URLError(.cannotFindHost)
}

// 再試行可能なデータ取得処理
func handleRetry() async {
    do {
        let result = try await fetchDataWithRetry(attempts: 3)
        print(result)
    } catch {
        print("Failed after multiple attempts: \(error)")
    }
}

// 実行
Task {
    await handleRetry()
}

import Foundation

// 再試行とキャンセルを組み合わせた非同期処理
func fetchDataWithRetryAndCancellation(attempts: Int) async throws -> String {
    var currentAttempt = 0

    while currentAttempt < attempts {
        // キャンセルチェック
        try Task.checkCancellation()

        do {
            // 実際のデータ取得をシミュレーション
            if currentAttempt < 2 {
                throw URLError(.badServerResponse) // 初回は失敗するシミュレーション
            }
            return "Data fetched on attempt \(currentAttempt + 1)"
        } catch {
            currentAttempt += 1
            print("Attempt \(currentAttempt) failed, retrying...")
            if currentAttempt == attempts {
                throw error
            }
            // 再試行前に少し待機
            await Task.sleep(2_000_000_000)
        }
    }

    throw URLError(.cannotFindHost)
}

// タスクをキャンセル可能な再試行処理
func handleRetryWithCancellation() async {
    let task = Task {
        do {
            let result = try await fetchDataWithRetryAndCancellation(attempts: 3)
            print(result)
        } catch {
            print("Task was cancelled or failed after multiple attempts: \(error)")
        }
    }

    // 3秒後にタスクをキャンセル
    await Task.sleep(3_000_000_000)
    task.cancel()
}

// 実行
Task {
    await handleRetryWithCancellation()
}