Race condition between 'onCharacteristicWrite()' and 'onCharacteristicChanged()' operations on the single characteristic

[joe-skb7]

Describe the bug
When peripheral device has a single characteristic for both TX and RX operations, next race condition happens: when writing to the characteristic, onNotificationReceived() event can happen during 'write' operation, replacing content of internal Android buffer, which leads to sending the received buffer (instead of one we want to send). Observing everything on the same thread doesn't help.

The root cause can be that BluetoothGattCharacteristic class shares mValue between read and write operations, though I'm not confident about this. Such race condition also mentioned here.

One possible workaround would be to send data only on new data received. Such a workaround is implemented in class below (see setWriteOnRead()), and it's working for me. But it's of course just a hack. Another way to solve the problem could be to keep two different instances of BluetoothGattCharacteristic class, one for read and one for write. This way we can keep buffers separate, avoiding race condition. Not sure if it's possible to implement something like this, though.

Of course, it's 100% Android bug, not an RxAndroidBle library bug. Frankly, I've stumbled upon this bug first when using Android BLE framework directly, and kinda hoped that this library would help me to avoid that (and other issues).

Here is the class I'm using for connection and data exchange (basically it's modified sample/example4_characteristic):

package com.xxx.xxx;

import android.bluetooth.BluetoothGattCharacteristic;
import android.content.Context;
import android.util.Log;

import com.jakewharton.rx.ReplayingShare;
import com.polidea.rxandroidble2.RxBleClient;
import com.polidea.rxandroidble2.RxBleConnection;
import com.polidea.rxandroidble2.RxBleDevice;

import java.util.LinkedList;
import java.util.Queue;
import java.util.UUID;

import io.reactivex.Observable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.subjects.PublishSubject;

public class BleManager {
    public interface Callbacks {
        void onConnectionEstablished();
        void onConnectionFailure(Throwable throwable);
        void onNotificationHasBeenSetUp();
        void onNotificationReceived(byte[] bytes);
    }

    private static final String TAG = DeviceActivity.class.getSimpleName();
    private static final String HM10_RX_TX = "0000ffe1-0000-1000-8000-00805f9b34fb";
    private static final UUID UUID_HM10_RX_TX = UUID.fromString(HM10_RX_TX);

    private RxBleDevice mBleDevice;
    private Observable<RxBleConnection> mConnectionObservable;
    private PublishSubject<Boolean> mDisconnectTriggerSubject = PublishSubject.create();
    private final CompositeDisposable mCompositeDisposable = new CompositeDisposable();
    private BleManager.Callbacks mCallbacks;
    private boolean mWriteOnRead = false;
    private Queue<String> mWriteQueue = new LinkedList<>();

    // ---------------------------------------------------------------------------------------------

    @SuppressWarnings("unused")
    private  BleManager() {
    }

    BleManager(Context context, BleManager.Callbacks callbacks, String macAddress) {
        mCallbacks = callbacks;
        RxBleClient mRxBleClient = App.getRxBleClient(context);
        mBleDevice = mRxBleClient.getBleDevice(macAddress);
        Log.d(TAG, "Using device address: " + macAddress);
    }

    public void start() {
        Log.d(TAG, "Starting GATT...");
        mConnectionObservable = prepareConnectionObservable();
        establishConnection();
        Log.d(TAG, "GATT started; waiting for connection...");
    }

    public void stop() {
        mCompositeDisposable.clear();
    }

    public boolean isConnected() {
        return mBleDevice.getConnectionState() == RxBleConnection.RxBleConnectionState.CONNECTED;
    }

    public void disconnect() {
        mDisconnectTriggerSubject.onNext(true);
    }

    public void write(String buf) {
        if (!isConnected()) {
            Log.w(TAG, "Attempt to write on not established connection");
            return;
        }

        if (mWriteOnRead) {
            mWriteQueue.add(buf);
        } else {
            writeInternal(buf);
        }
    }

    // Workaround for Android BLE race condition bug (shared mValue in BluetoothGattCharacteristic)
    public void setWriteOnRead(boolean enable) {
        mWriteOnRead = enable;
    }

    // ---------------------------------------------------------------------------------------------

    private void writeInternal(String buf) {
        final Disposable disposable = mConnectionObservable
                .firstOrError()
                .flatMap(rxBleConnection -> {
                    final byte[] tx = buf.getBytes();
                    return rxBleConnection.writeCharacteristic(UUID_HM10_RX_TX, tx);
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                        bytes -> onWriteSuccess(),
                        this::onWriteFailure
                );

        mCompositeDisposable.add(disposable);
    }

    private void onConnectionFailure(Throwable throwable) {
        mCallbacks.onConnectionFailure(throwable);
    }

    private void onWriteSuccess() {
        // Log.d(TAG, "### Write success");
    }

    private void onWriteFailure(Throwable throwable) {
        Log.e(TAG, "### Write error: " + throwable);
    }

    private void onNotificationReceived(byte[] bytes) {
        if (mWriteOnRead && !mWriteQueue.isEmpty()) {
            String data = mWriteQueue.remove();
            writeInternal(data);
        }
        mCallbacks.onNotificationReceived(bytes);
    }

    private void onNotificationSetupFailure(Throwable throwable) {
        Log.e(TAG, "### Notification error: " + throwable);
    }

    private void onConnectionFinished() {
        Log.d(TAG, "### Connection finished");
    }

    private void onConnectionEstablished(BluetoothGattCharacteristic characteristic) {
        Log.d(TAG, "### Connection has been established! properties");
        setupNotifications();
        mCallbacks.onConnectionEstablished();
    }

    private void notificationHasBeenSetUp(Observable<byte[]> observable) {
        mCallbacks.onNotificationHasBeenSetUp();
    }

    private Observable<RxBleConnection> prepareConnectionObservable() {
        return mBleDevice
                .establishConnection(false)
                .takeUntil(mDisconnectTriggerSubject)
                .compose(ReplayingShare.instance());
    }

    private void establishConnection() {
        if (isConnected()) {
            Log.w(TAG, "Attempt to establish already established connection");
            return;
        }

        final Disposable connectionDisposable = mConnectionObservable
                .flatMapSingle(RxBleConnection::discoverServices)
                .flatMapSingle(rxBleDeviceServices ->
                        rxBleDeviceServices.getCharacteristic(UUID_HM10_RX_TX))
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSubscribe(disposable -> Log.d(TAG, "Connecting..."))
                .subscribe(
                        this::onConnectionEstablished,
                        this::onConnectionFailure,
                        this::onConnectionFinished
                );

        mCompositeDisposable.add(connectionDisposable);
    }

    private void setupNotifications() {
        if (!isConnected()) {
            Log.w(TAG, "Attempt to setup notifications on not established connection");
            return;
        }

        // Don't add .observeOn(mainThread), to avoid receive speed degradation
        final Disposable disposable = mConnectionObservable
                .flatMap(rxBleConnection -> rxBleConnection.setupNotification(UUID_HM10_RX_TX))
                .doOnNext(this::notificationHasBeenSetUp)
                .flatMap(notificationObservable -> notificationObservable)
                .subscribe(this::onNotificationReceived, this::onNotificationSetupFailure);

        mCompositeDisposable.add(disposable);
    }
}

To Reproduce
Steps to reproduce the behavior:

1. Peripheral device should send data very fast (the one I use has a rate of 55 packets per second); always send e.g. packet "A"
2. Single characteristic should be used for TX and RX.
3. As a code reference -- class above can be used.
4. Write some uniqe packets (e.g. packet "B") to the characteristic, like 1 time per second
5. Watch received packets on peripheral device: sometimes it will receive packet "A", which is a bug

Expected behavior
For a test case above: peripheral device should never receive packet "A". In other words, described race condition should never happen on Android device, and packets we write should never be overwritten with received packets.

Smartphone (please complete the following information):

• Device: OnePlus 5
• OS: Android 9
• Library version: 1.11.1

Logs from the application when bug occurs (this will greatly help in quick understanding the problem)

2020-06-04 19:32:03.771 14587-14587/com.xxx.xxx D/RxBle#ConnectionOperationQueue: QUEUED   CharacteristicWriteOperation(93082492)
2020-06-04 19:32:03.772 14587-14674/com.xxx.xxx D/RxBle#ConnectionOperationQueue: STARTED  CharacteristicWriteOperation(93082492)
2020-06-04 19:32:03.773 14587-14674/com.xxx.xxx I/RxBle#ConnectionOperationQueue: RUNNING  CharacteristicWriteOperation{MAC='F8:36:9B:3F:32:30', characteristic=[uuid='0000ffe1-0000-1000-8000-00805f9b34fb', hexValue=[48, 0D, 0A]]}
2020-06-04 19:32:03.780 14587-14621/com.xxx.xxx I/RxBle#GattCallback: MAC='F8:36:9B:3F:32:30'  onCharacteristicChanged(), value=[uuid='0000ffe1-0000-1000-8000-00805f9b34fb', hexValue=[40, 38, 41, 6C, 6E, 2F, 50, 63, 4A, 58, 66, 7A, 76, 43, 56, 7A, 38, 0D, 0A]]
2020-06-04 19:32:03.785 14587-14621/com.xxx.xxx I/RxBle#GattCallback: MAC='F8:36:9B:3F:32:30'    onCharacteristicWrite(), status=0, value=[uuid='0000ffe1-0000-1000-8000-00805f9b34fb']
2020-06-04 19:32:03.788 14587-14674/com.xxx.xxx D/RxBle#ConnectionOperationQueue: FINISHED CharacteristicWriteOperation(93082492) in 16 ms

Additional context

From log one can see that I'm writing [48, 0D, 0A] packet. But because onCharacteristicChanged() happens during write operation, instead of [48, 0D, 0A] I'm observing this packet on peripheral device: [40, 38, 41, 6C, 6E, 2F, 50, 63, 4A, 58, 66, 7A, 76, 43, 56, 7A, 38, 0D, 0A].

[JitainSharma]

Library: 1.11.1
Android: 10

Same behaviour is also occurs over my devices as well.

Also, many times it stuck at the scan where following logs are printed:

2020-09-14 14:10:23.901 17297-17297 D/RxBle#ClientOperationQueue: QUEUED   ScanOperationApi21(162649957)

2020-09-14 14:10:23.910 17297-17526 D/RxBle#ClientOperationQueue: STARTED  ScanOperationApi21(162649957)

2020-09-14 14:10:23.912 17297-17526 I/RxBle#ClientOperationQueue: RUNNING  ScanOperationApi21{ANY_MUST_MATCH -> nativeFilters=[BluetoothLeScanFilter [mDeviceName=null, MAC=null, mUuid=000a0001-1000-8000-0080-5f9b34fb0000, mUuidMask=null, mServiceDataUuid=null, mServiceData=null, mServiceDataMask=null, mManufacturerId=-1, mManufacturerData=null, mManufacturerDataMask=null], BluetoothLeScanFilter [mDeviceName=null, MAC=null, mUuid=000a0003-0010-0080-0000-805f9b34fb00, mUuidMask=null, mServiceDataUuid=null, mServiceData=null, mServiceDataMask=null, mManufacturerId=-1, mManufacturerData=null, mManufacturerDataMask=null]]}

2020-09-14 14:10:23.917 17297-17532 I/RxBle#QueueOperation: Scan operation is requested to start.

2020-09-14 14:10:23.917 17297-17532 D/RxBle#ScanOperationApi21: No library side filtering —> debug logs of scanned devices disabled

2020-09-14 14:10:23.922 17297-17532 D/BluetoothAdapter: STATE_ON

2020-09-14 14:10:23.926 17297-17532 I/chatty: uid=10303(org.mircom) pool-23-thread- identical 1 line

2020-09-14 14:10:23.930 17297-17532 D/BluetoothAdapter: STATE_ON

2020-09-14 14:10:23.930 17297-17532 D/BluetoothLeScanner: Start Scan with callback

2020-09-14 14:10:23.936 17297-22187 D/BluetoothLeScanner: onScannerRegistered() - status=0 scannerId=12 mScannerId=0

2020-09-14 14:10:23.950 17297-17526 D/RxBle#ClientOperationQueue: FINISHED ScanOperationApi21(162649957) in 45 ms

where it shows Scan operation api 21 is finished and no scan result found/return.

Where on Android 9, it's working fine and above situation is limited or few, but recovers after giving break of 5 secs for next try. But on Android 10, it's stuck and doesn't recover after long time.