kotlin - 使用 kotlin-flow 对内部流进行双重订阅

Question

下面的代码片段存在以下问题；

1. 当当前连接的条形码扫描仪的集合发生变化时，流set将发出当前连接的扫描仪scanners: Flow<Set<BarcodeScanner>>。

2. 如果我连接了一个名为 A 的扫描仪，并且我扫描了一个条形码，那么它工作正常，结果将显示在文本视图中。

3. 如果我连接一个额外的设备，然后再次使用设备 A 扫描，我将看到 2 次扫描结果。

4. 看起来每次更改 Set<> 时都会在 中创建一个额外的流订阅，collect { scanner.scan() }并且之前的流仍然存在。

我该如何解决这个问题？我已经和不同的运营商一起玩了一整天……我正在使用 Flow 1.2.2

 uiScope.launch { //CoroutineScope(Job() + Dispatchers.Main)
             scanners //Flow<Set<BarcodeScanner>> will emit the set of scanner, currently connected scanners, when one connects/disconnects
                .onEach {
                    txtView.text = "Barcode scanners: ${it.size}\n"
                }
                .flatMapMerge { // Flatten to Flow<BarcodeScanner>
                    it.asFlow()
                }
                .onEach { scanner ->
                    txtView.append("${scanner.name()} [${scanner.hashCode()}]\n")
                }
                .collect { scanner ->
                    //if a new Set<BarcodeScanner> is emitted, this Flow<Barcode> returned by scanner.scan() will be subscribed again, without removing the old stream
                    //resulting in multiple results on a single scan.
                    scanner.scan() //returns an Flow<Barcode>
                        .onEach { barcode ->
                            findViewById<TextView>(R.id.txtViewBarcodes).append("[${scanner.name()}]: ${barcode.value} (${barcode.type.description})\n")
                        }
                        .onCompletion {
                            Timber.i("[${scanner.name()}]: DISCONNECTED") //Never called
                        }
                        .launchIn(scanScope)
                }
        }

Answer 1

我认为而不是：

.collect { scanner ->
    scanner.scan()
        .onEach { barcode ->
    ...

你应该使用：

.flatMapLatest { scanner ->
    scanner.scan()
}
.onEach { barcode ->
...

在同一 UI 范围内启动流的位置。您遇到的问题launchIn(scanScope)是创建一个完全独立的协程，当发出一组新的条形码时，外部协程无法取消。