ConvertedBroadcast.kt

/*
fun BluetoothSocket.asyncListen(): Flowable<String> =
        flowable<String>(BackpressureStrategy.BUFFER) { emitter ->
            try {
                val reader = BufferedReader(InputStreamReader(inputStream))
                while (!emitter.isCancelled) {
                    reader.readLine()?.let { emitter.onNext(it) }
                }
            } catch (e: IOException) {
                emitter.onError(e)
            }
        }
        .subscribeOnNewThread()
        .share()
*/

// for simplicity we assume just a single thread of reader is Ok
val readerThreadContext = newSingleThreadContext("reader")

fun BluetoothSocket.asyncListen(broadcast: BroadcastChannel<String>) { 
    launch(readerThreadContext) {
        val reader = BufferedReader(InputStreamReader(inputStream))
        while (isActive) {
            reader.readLine()?.let { broadcast.send(it) }
        }
    }
}

/*
private fun configIncomingData(dataStream: Flowable<String>) {
        disposables += dataStream
                .observeOnComputationThread()
                .filter(String::isNumber)
                .map(String::toDigitalValue)
                .subscribeBy(
                        onNext = {
                            val color = profile.analyteInfo.calibration.getColor(currentTime)
                            val coloredValue = ColoredValue(it, color)
                            currentTime += calibration.timeParams.increment.toFloat()
                            val measurement = Measurement(currentTime, it.toDouble())
                            if (coloredValue.color == Color.BLUE) measurements.add(measurement)
                            coloredDots.add(coloredValue)
                        },
                        onError = { postEvent(ConnectionEvent(DISCONNECTED)) }
                )
        disposables += dataStream
                .observeOnNewThread()
                .filter(String::isBaselineMessage)
                .doOnNext {
                    calibration.baseline =
                            measurements.map { (_, digitalValue) -> digitalValue }.average()
                }
                .observeOnMainThread()
                .subscribeBy(
                        onNext = { view.mainView.displaySnackbar("Baseline Analysis Completed") },
                        onError = { postEvent(ConnectionEvent(DISCONNECTED)) }
                )
        disposables += dataStream
                .filter(String::isSampleMessage)
                .observeOnMainThread()
                .subscribeBy(
                        onNext = {
                            val correctedData = calibration.correctBaseline(measurements)
                            val concentration = concentrationLevel.get().toDouble()
                            val integral: Double = correctedData.integral.withThreeDecimalPlaces
                            calibration.replicates.add(Replicate(concentration, integral))
                            view.displayData(coloredDots)
                            lastMeasurementMessage.set(
                                    "Last Measurement: L = $concentration ppm, I = $integral"
                            )
                            view.mainView.displaySnackbar("Sample Analysis Completed")
                        },
                        onError = { postEvent(ConnectionEvent(DISCONNECTED)) }
                )
    }
*/

private fun configIncomingData(): BroadcastChannel<String> {
    val broadcast = BroadcastChannel<String>(1)
    // first background consumer
    launch(CommonPool, CoroutineStart.UNDISPATCHED) { // NOTE: Start it immediately
        broadcast.consumeEach { 
            if (it.isNumber()) {
                val color = profile.analyteInfo.calibration.getColor(currentTime)
                val coloredValue = ColoredValue(it.toDigitalValue(), color)
                currentTime += calibration.timeParams.increment.toFloat()
                val measurement = Measurement(currentTime, it.toDigitalValue().toDouble())
                if (coloredValue.color == Color.BLUE) measurements.add(measurement)
                coloredDots.add(coloredValue)
            }  
        }
    }
    // second UI consumer
    launch(UI, CoroutineStart.UNDISPATCHED) { // NOTE: Start it immediately
        broadcast.consumeEach { 
         if (it.isSampleMessage()) {
            val correctedData = calibration.correctBaseline(measurements)
            val concentration = concentrationLevel.get().toDouble()
            val integral: Double = correctedData.integral.withThreeDecimalPlaces
            calibration.replicates.add(Replicate(concentration, integral))
            view.displayData(coloredDots)
            lastMeasurementMessage.set(
                    "Last Measurement: L = $concentration ppm, I = $integral"
            )
            view.mainView.displaySnackbar("Sample Analysis Completed")
        }
   // etc....         
}