Why my RxJava setup is blocking my UI thread? Working with BluetoothAdapter.startLeScan callback

Question

I am struggling to find the specific action which is blocking my UI thread, i've tried several Scheduler operators, but I'm not sure how to make it work.

I have a UI with button, which onClicked is starting bluetooth scan and updates the textView with strings like a log(it shows what is happening at the moment).

So here is my MainActivity:

 lateinit var disposable: Disposable
val textDataService = TextDataService()


override fun onCreate(savedInstanceState: Bundle?) {
    super.onCreate(savedInstanceState)
    setContentView(R.layout.activity_scan_test)


    buttonScanTestStart.setOnClickListener {
        if (isBluetoothEnabled()) {
            textViewLog.text = ""

            buttonScanTestStop.visibility = View.VISIBLE
            buttonExportScanTestRaportSummary.visibility = View.GONE
            buttonExportScanTestRaportFull.visibility = View.GONE
            buttonScanTestStart.visibility = View.GONE

            disposable=
                    Scanner()
                            .discoverSingleDevice(this, "   ", textViewLog)
                            .doOnError {
                                setText("General error: ${it.message ?: it::class.java}", textViewLog)
                                setLogText("General error: ${it.message ?: it::class.java}")
                            }
                            .repeat(1)
                            .doOnComplete {
                                buttonScanTestStop.visibility = View.GONE
                            }
                            .doOnDispose {
                                log("TEST DISPOSED")
                            }
                            .subscribe()
        } else {
            Toast.makeText(this, "Please, enable bluetooth to start test.", Toast.LENGTH_SHORT).show()
        }
    }

    buttonScanTestStop.setOnClickListener {
        disposable.dispose()
        buttonScanTestStop.visibility = View.GONE
        buttonScanTestStart.visibility = View.VISIBLE
        buttonExportScanTestRaportSummary.visibility = View.VISIBLE
        buttonExportScanTestRaportFull.visibility = View.VISIBLE

        textDataService.generateScanGeneralStatisticsLogText(textViewLog)
    }

Here is the Scanner class:

class Scanner {


private fun scan(context: Context, textView: TextView) = Observable.create<ScannedItem> { emitter ->
    val bluetoothAdapter = context.getBluetoothAdapter()
    if (bluetoothAdapter.isEnabled) {

        val scanCallback = BluetoothAdapter.LeScanCallback { bluetoothDevice, rssi, _ ->

            if(bluetoothDevice.name == null){
                    bluetoothRawLog("Scanned Item -> ${bluetoothDevice.logText()} | rssi = $rssi | time: = ${getCurrentTime()}")
                    scannedOtherDevices.add(bluetoothDevice.address)
            }
            else{
                if(!scannedDevices.contains(bluetoothDevice.name)){
                        setText("Scanned Item -> ${bluetoothDevice.logText()} | rssi = $rssi | time: = ${getCurrentTime()}\n", textView)

                    setLogText("Scanned Item -> ${bluetoothDevice.logText()} | rssi = $rssi | time: = ${getCurrentTime()}\n")
                }
                scannedDevices.add(bluetoothDevice.name)


            }
            bluetoothRawLog("Scanned Item -> ${bluetoothDevice.logText()} | rssi = $rssi")


            bluetoothDevice.name?.let {
                emitter.onNext(ScannedItem(it, bluetoothDevice.address))
            }
        }

        emitter.setCancellable { bluetoothAdapter.stopLeScan(scanCallback) }
        bluetoothAdapter.startLeScan(scanCallback)

    } else
        emitter.onError(IllegalStateException("Bluetooth turned off"))
}
        .doOnNext { log("Scanned -> $it") }
        .timeout(12, TimeUnit.SECONDS)
        .observeOn(AndroidSchedulers.mainThread())
        .doOnError { if (it is TimeoutException) setLogText("Scanner -> no named devices for 12 seconds: resetting...") }
        .retry { t -> t is TimeoutException }


  fun discoverSingleDevice(context: Context, searchName: String, textView: TextView): Observable<ScannedItem> = scan(context, textView)
        .filter { it.name.contains(searchName) }
        .take(1)

And here is my Kotlin extension functions, which i am using to set Text:

fun setLogText(text: String) {
    logBuffer.append(text)
}

fun setText(text: String, textView: TextView) {
    textView.append(text)
}

Here is also my layout:

<?xml version="1.0" encoding="utf-8"?>

<LinearLayout
    android:layout_margin="16dp"
    android:layout_width="wrap_content"
    android:layout_height="wrap_content"
    android:layout_gravity="center"
    android:orientation="horizontal">

    <Button
        android:layout_weight="1"
        android:id="@+id/buttonScanTestStart"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_gravity="center"
        android:text="Start Test" />
    <Button
        android:layout_weight="1"
        android:visibility="gone"
        android:id="@+id/buttonScanTestStop"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_gravity="center"
        android:text="Stop Test" />
    <Button
        android:layout_weight="1"
        android:id="@+id/buttonExportScanTestRaportFull"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_gravity="center"
        android:text="Export raport"
        android:visibility="gone" />

    <Button
        android:layout_weight="1"
        android:id="@+id/buttonExportScanTestRaportSummary"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_gravity="center"
        android:text="Export summary"
        android:visibility="gone" />
</LinearLayout>

<LinearLayout
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    android:orientation="vertical">

    <LinearLayout
        android:visibility="gone"
        android:id="@+id/scanTestStartedLinearLayout"
        android:gravity="center"
        android:orientation="horizontal"
        android:layout_width="match_parent"
        android:layout_height="wrap_content">
        <TextView
            android:text="Test running..."
            android:layout_width="wrap_content"
            android:layout_height="wrap_content" />
        <ProgressBar
            android:id="@+id/scanTestProgressBar"
            android:layout_width="16dp"
            android:layout_height="16dp" />
    </LinearLayout>

    <TextView
        android:id="@+id/textViewScanTestLog"
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:layout_margin="16dp"
        android:gravity="center" />

    <ScrollView
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:layout_margin="16dp"
        android:gravity="center">

        <LinearLayout
            android:layout_width="match_parent"
            android:layout_height="wrap_content"
            android:orientation="vertical">

            <TextView
                android:id="@+id/textViewLog"
                android:layout_width="match_parent"
                android:layout_height="wrap_content" />

        </LinearLayout>

    </ScrollView>

</LinearLayout>

So the text is appearing in my textView pretty smooth, everything is OK. However, when there is more amount of available devices to scan i can't stop the test by clicking the button. It is just freezed and responding like way too late when i try to click it.

The discoverSingleDevice function is will be run forever, because i filter to have the search name from my editText(which i hardcoded with " ") to make it run forever and i want to only stop this by clicking the stop button in my layout. As a second parameter i passed to it textViewLog from my layout(Kotlin is smart enough to find it by id)

So how to prevent blocking the UI with this setup?

UPDATE:

As WenChao suggested, I've done something like this:

       val observable = Scanner()
                    .discoverSingleDevice(this, "   ", textViewLog)
                    .doOnError {
                        nexoSetText("General error: ${it.message ?: it::class.java}", textViewLog)
                        nexoSetLogText("General error: ${it.message ?: it::class.java}")
                    }
                    .repeat(1)
                    .doOnComplete {
                        buttonScanTestStop.visibility = View.GONE
                    }
                    .doOnDispose {
                        nexoLog("TEST DISPOSED")
                    }

         disposable =  observable
                    .subscribeOn(Schedulers.newThread()).subscribe()

However, it not helped, it is not that simple. Again, I've tried several things with other operators. Please, be specific to my case - I provided code.

It is like after 2 minutes of scanning the app is rather freezing. In the logs of course appearing I/Choreographer: Skipped 54 frames! The application may be doing too much work on its main thread.

UPDATE 2:

As PhoenixWang suggested, I've log to check on which thread the BluetoothAdapter runs. So I've done it like this in my scan method in the Scanner class:

  private fun scan(context: Context, textView: TextView) = Observable.create<ScannedItem> { emitter ->
    val bluetoothAdapter = context.getBluetoothAdapter()
    if (bluetoothAdapter.isEnabled) {

        val scanCallback = BluetoothAdapter.LeScanCallback { bluetoothDevice, rssi, _ ->
            Log.v("BluetoothThread", "Running on: " + Thread.currentThread().name)
///the rest of the code

So how to already fix this? I've thought that it will actually run on the separate thread(not main) because i've done the naive fix, which is visible in my first Update. Can you help?

UPDATE 3:

So the problem is with the BluetoothAdapter.leScan - it is running on main thread, even if i do something like this:

Observable.fromCallable{ bluetoothAdapter.startLeScan(scanCallback)}.subscribeOn(Schedulers.newThread()).subscribe()

As PhoenixWang suggesting, BluetoothAdapter.LeScanCallback run in mainthread because of his implementation. Observable/RxJava can't change it. So how to resolve my problem?

Answer 1

To be more clear. I make a answer but can not solve your problem . Your code:

 Observable.fromCallable{ bluetoothAdapter.startLeScan(scanCallback)}.subscribeOn(Sche‌​dulers.newThread()).‌​subscribe().

means that your bluetoothAdapter.startLeScan(scanCallback) is run in the newThread. But not how bluetoothAdapter call you back.

So in the implementation of your startLeScan. It can start a new thread or run in other thread depends on it's implementation.

Same idea with your Observable.create(//your stuff)

They are guaranteed running on your specified scheduler. But it can also start a new thread inside of it. That's why I mentioned in comment to check the implementation of your BluetoothAdapter.

Update

To be more clear, an example implementation of BluetoothAdapter that can block your UI Thread.

class BluetoothAdapter {

    fun startLeScan(callback: LeScanCallback) {
        val handler = Handler(Looper.getMainLooper())
        handler.post {
            // here you back to your UI Thread. 
            callback.onSuccess(1)
        }
    }

    interface LeScanCallback {
        fun onSuccess(result: Int)
    }
}

So if your implementation of BluetoothAdapter will change the thread back to UI thread. You eventually will block your UI thread no matter what thread of your Observable. That what I mean Check the implementation of your BluetoothAdapter

Answer 2

 .subscribeOn(Schedulers.io())
 .observeOn(AndroidSchedulers.mainThread())

Add .subscribeOn(Schedulers.io()) along with .observeOn(AndroidSchedulers.mainThread()) like above. Try this. Should work.

Answer 3

RxJava is synchronous by default. In your case, it runs on the caller thread which is the main thread of Android

You have to specify what thread you want the subscription to happen.

Observable.fromCallable(whatever())
        .subscribeOn(Schedulers.newThread()).subscribe()

read more here.