CoreBluetooth中AsyncThrowingStream的完成、错误与取消模拟

以下是针对你的需求修改后的完整实现方案，包含send函数优化、事件触发逻辑以及蓝牙回调的整合：

import CoreBluetooth

// 定义外设返回的指令类型
enum BluetoothCommandResult {
    case progress(Double)
    case success
}

// 自定义蓝牙错误类型
enum BluetoothError: Error {
    case unexpectedResponse
    case invalidResponse
    case peripheralError
    case unknownCommand
}

class TestService: NSObject, CBPeripheralDelegate {
    private weak var peripheral: CBPeripheral?
    private var targetCharacteristic: CBCharacteristic?
    private var currentContinuation: AsyncThrowingStream<BluetoothCommandResult, Error>.Continuation?
    
    // 初始化：传入已连接的外设和目标特征
    init(peripheral: CBPeripheral, targetCharacteristic: CBCharacteristic) {
        self.peripheral = peripheral
        self.targetCharacteristic = targetCharacteristic
        super.init()
        peripheral.delegate = self
    }
    
    // 核心发送函数：返回异步流接收多阶段响应
    func send(commandData: Data) -> AsyncThrowingStream<BluetoothCommandResult, Error> {
        return AsyncThrowingStream { [weak self] continuation in
            guard let self = self, let peripheral = self.peripheral, let characteristic = self.targetCharacteristic else {
                continuation.finish(throwing: BluetoothError.unexpectedResponse)
                return
            }
            
            // 保存当前流的continuation，用于后续事件回调
            self.currentContinuation = continuation
            
            // 监听流终止事件：处理取消逻辑
            continuation.onTermination = { termination in
                if case .cancelled = termination {
                    self.cancelSend()
                }
                self.currentContinuation = nil
            }
            
            // 执行蓝牙数据写入
            peripheral.writeValue(commandData, for: characteristic, type: .withResponse)
        }
    }
    
    // 触发完成事件：结束流
    private func finishSend() {
        currentContinuation?.yield(.success)
        currentContinuation?.finish()
    }
    
    // 触发取消操作：发送蓝牙取消指令并终止流
    private func cancelSend() {
        guard let peripheral = peripheral, let characteristic = targetCharacteristic else { return }
        // 替换为你的外设取消指令格式
        let cancelCommand = Data([0xFF])
        peripheral.writeValue(cancelCommand, for: characteristic, type: .withResponse)
        
        currentContinuation?.finish(throwing: CancellationError())
    }
    
    // 触发错误事件：将错误传递到流中
    private func triggerError(_ error: Error) {
        currentContinuation?.finish(throwing: error)
    }
    
    // MARK: - CBPeripheralDelegate 回调处理
    func peripheral(_ peripheral: CBPeripheral, didWriteValueFor characteristic: CBCharacteristic, error: Error?) {
        if let error = error {
            triggerError(error)
            return
        }
        // 写入成功，等待外设返回进度/完成响应
    }
    
    func peripheral(_ peripheral: CBPeripheral, didUpdateValueFor characteristic: CBCharacteristic, error: Error?) {
        guard let responseData = characteristic.value, error == nil else {
            triggerError(error ?? BluetoothError.unexpectedResponse)
            return
        }
        
        // 解析外设返回的数据（根据你的通信协议调整逻辑）
        switch parseResponse(data: responseData) {
        case .progress(let progressValue):
            currentContinuation?.yield(.progress(progressValue))
        case .success:
            finishSend()
        case .error(let error):
            triggerError(error)
        }
    }
    
    // 解析外设返回的数据示例
    private func parseResponse(data: Data) -> BluetoothCommandResult {
        guard let statusByte = data.first else {
            return .error(BluetoothError.invalidResponse)
        }
        
        switch statusByte {
        case 0x01: // 进度指令
            let progress = Double(data[1]) / 100.0 // 假设第二个字节是0-100的进度值
            return .progress(progress)
        case 0x02: // 完成指令
            return .success
        case 0x03: // 错误指令
            return .error(BluetoothError.peripheralError)
        default:
            return .error(BluetoothError.unknownCommand)
        }
    }
}

// 使用示例
// let connectedPeripheral: CBPeripheral = ... // 已连接的外设
// let targetCharacteristic: CBCharacteristic = ... // 目标特征
// 
// Task {
//     let service = TestService(peripheral: connectedPeripheral, targetCharacteristic: targetCharacteristic)
//     do {
//         let responseStream = service.send(commandData: yourCommandData)
//         for try await result in responseStream {
//             switch result {
//             case .progress(let value):
//                 print("当前进度：\(value * 100)%")
//             case .success:
//                 print("操作完成")
//             }
//         }
//     } catch is CancellationError {
//         print("操作已取消")
//     } catch {
//         print("发生错误：\(error.localizedDescription)")
//     }
// }
// 
// // 取消操作（例如用户点击取消按钮）
// // task.cancel()

关键逻辑说明 #

1. send函数：

   • 每次调用创建新的AsyncThrowingStream，绑定对应的Continuation
   • 通过onTermination监听流的取消事件，自动触发蓝牙侧的取消指令
   • 直接执行蓝牙数据写入，启动外设处理流程

2. 事件触发逻辑：

   • 完成：通过yield(.success)发送完成信号，再调用finish()终止流
   • 取消：先向外设发送取消指令，再抛出CancellationError终止流
   • 错误：直接将错误传递到流中，终止并通知消费者

3. 蓝牙回调整合：

   • didWriteValueFor处理写入错误，直接触发错误事件
   • didUpdateValueFor解析外设返回的数据，根据类型分别发送进度、完成或错误信号

内容的提问来源于stack exchange，提问作者lcj