RxBluetoothKit - реализовать протокол чтения и записи и автоматически отключаться

Question

Я реализую протокол BLE между центральным (iPhone) и периферийным (пользовательское устройство). Протокол работает следующим образом:

1. центральный соединяется с периферийным устройством и устанавливает уведомление
2. периферийное устройство отправляет данные по характеристике уведомления
3. центральная обработка данных и отправка ответа по отдельной характеристике
4. периферийное устройство отправляет дополнительные данные по характеристике уведомления
5. центральный процесс данных и отключений.

Я пытаюсь реализовать это чисто, используя RxBluetoothKit. В настоящее время это работает, но я бы хотел решить следующие задачи:

1. Каков наилучший способ чистого отключения на шаге 5. Я надеюсь, что мне не придется избавляться от общего наблюдаемого, а скорее просто "завершить". В настоящее время я использую takeUntil, но не уверен, что это лучший способ.

2. Разрешить корректную очистку уведомления перед отключением. С моим текущим кодом я получаю сообщение «API MISUSE может принимать команды только в подключенном состоянии», потому что я считаю, что уведомление очищается во время отключения.

Спасибо.

enum TestPeripheralService: String, ServiceIdentifier {

    case main = "CED916FA-6692-4A12-87D5-6F2764762B23"

    var uuid: CBUUID { return CBUUID(string: self.rawValue) }
}

enum TestPeripheralCharacteristic: String, CharacteristicIdentifier {

    case writer = "CED927B4-6692-4A12-87D5-6F2764762B2A"
    case reader = "CED9D5D8-6692-4A12-87D5-6F2764762B2A"

    var uuid: CBUUID { return CBUUID(string: self.rawValue) }

    var service: ServiceIdentifier { return TestPeripheralService.main }
}

fileprivate lazy var centralManager: CentralManager = {
    RxBluetoothKitLog.setLogLevel(.verbose)
    return CentralManager(queue: .main)
}()

func executeConnectionAndHandshake() {

    let disconnectSubject = PublishSubject<Bool>.init()
    var peripheral: Peripheral?
    var packetNum = 0
    _ = centralManager
        .observeState()
        .startWith(centralManager.state)
        .filter { $0 == .poweredOn }
        .flatMap { _ in self.centralManager.scanForPeripherals(withServices: [TestPeripheralService.main.uuid]) }
        .flatMap { $0.peripheral.establishConnection().takeUntil(disconnectSubject) }
        .do(onNext: { peripheral = $0 })
        .flatMap { $0.discoverServices([TestPeripheralService.main.uuid])}
        .flatMap { $0[0].discoverCharacteristics(nil)}
        .flatMap { _ in

            Observable<Bool>.create { event in

                let disposables = CompositeDisposable()
                let readSubject = PublishSubject<Data>.init()
                _ = disposables.insert(peripheral!.observeValueUpdateAndSetNotification(for: TestPeripheralCharacteristic.reader)
                    .subscribe(onNext: {
                        packetNum += 1
                        let packet = $0.value!
                        if (packetNum <= 1) {
                            readSubject.onNext(packet)
                        } else {
                            event.onNext(true)
                            event.onCompleted()
                        }

                    }, onError: { event.onError($0) })
                )

                _ = disposables.insert(readSubject
                    .flatMapLatest { data -> Single<Characteristic> in
                        var writeData = Data(capacity: 300)
                        for _ in 0..<300 {
                            writeData.append(0xFF)
                        }
                        return peripheral!.writeValue(writeData, for: TestPeripheralCharacteristic.writer, type: .withResponse)
                    }
                    .subscribe(onError: { event.onError($0) })
                )

                return Disposables.create {
                    disposables.dispose()
                }
            }
            .do(onCompleted: { disconnectSubject.onNext(true) })
        }
        .subscribe(onError: { print($0) },
                   onCompleted: { print("Connection and handshake completed") })
}