Отправка списка команд устройству с помощью RxAndroidBle (rxJava)

Question

Я пытаюсь отправить список команд на устройство через rxJava. Вот мой код:

public void startWriteCommucation(final ArrayList<byte[]> b) { 
    if (isConnected()){ 
      connectionObservable 
        .flatMap(new Func1<RxBleConnection, Observable<Observable<byte[]>>>() { 
         @Override 
         public Observable<Observable<byte[]>> call(final RxBleConnection rxBleConnection) { 
          final List<Observable<byte[]>> list = new ArrayList<>(); 
          for (byte[] bytes: b){ 
           Log.e("Observer", Arrays.toString(bytes)); 
           list.add(rxBleConnection 
             .writeCharacteristic(BleDevice.characteristicWrite, bytes)); 
          } 
          return Observable.from(list); 
         } 
        }) 
        .concatMap(new Func1<Observable<byte[]>, Observable<byte[]>>() { 
         @Override 
         public Observable<byte[]> call(Observable<byte[]> observable) { 
          return observable; 
         } 
        }) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(new Action1<byte[]>() { 
         @Override 
         public void call(byte[] bytes) { 
          view.setTextStatus("Write success"); 
          Log.e("Subscriber", Arrays.toString(bytes)); 
         } 
        }); 
     } 
} 

Это работает, затем я нажимаю кнопку один раз. Например, мой метод clikc:

public void onClick(){ 
     ArrayList<byte[]> listCmd = new ArrayList<>(); 
     listCmd.add(new byte[]{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}); 
     listCmd.add(new byte[]{0, 0, 0, 0, 0, 0, 0, 0, 0, 0}); 
     startWriteCommucation(listCmd); 
} 

И myLogs в LogCat:

E/Observer: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
E/Observer: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 

E/Subscriber: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
E/Subscriber: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 

Но проблема возникает, когда я использую быстрый двойной щелчок на кнопке. Затем первый клик с наблюдаемым все еще работает, я снова нажимаю, чтобы снова вызвать метод startWriteCommunication. И после этого мои журналы выглядят так:

E/Observer: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
E/Observer: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 
E/Observer: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
E/Observer: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 

E/Subscriber: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
E/Subscriber: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 
E/Subscriber: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 
E/Subscriber: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 

Основная проблема, что они не в порядке, и мое устройство работает некорректно. Можете ли вы помочь найти плоблем?

Answer 1

Проблема заключается в ошибке библиотеки RxAndroidBle (которая делает ответы не соответствующими запросам) и обмена соединением между двумя потоками связи, которые являются состояниями (требуется сделать две записи в порядке без какой-либо связи между ними).

Ошибка: значение (byte []), которое должно быть записано в BluetoothGattCharacteristic, устанавливается слишком рано. Если для одного и того же признака есть два параллельных автора, один из них может перезаписать байт [], который был установлен другим из-за состояния гонки. Я сделал исправление для библиотеки, которая сейчас находится в процессе обзора кода и должна быть применена в ближайшем будущем к выпуску SNAPSHOT.

С изменениями на выходе будет выглядеть следующим образом:

D/Observer: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
D/Observer: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 
D/Observer: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
D/Observer: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 

D/Subscriber: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
D/Subscriber: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
D/Subscriber: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 
D/Subscriber: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0] 

Возможное решение

Если вы не заинтересованы в стрельбе Поток в два раза, если пользователь будет нажать на кнопку два раза быстрее - вы можете создать общий поток:

Observable<byte[]> theSharedFlow = rxBleConnection 
    .writeCharacteristic(uuid, data1) 
    .flatMap(writtenBytes -> rxBleConnection.writeCharacteristic(uuid, data2)) 
    .share() 

Который при подписке несколько раз будет выполнен только один раз, пока не закончит. В приведенном выше фрагменте второй writeCharacteristic() будет подписан (и поставлен в очередь на связь) после того, как первый из них испустит записанные байты.

Если приложение предназначено для отправки произвольных наборов команд в произвольном порядке при совместном использовании соединения, то приложение должно убедиться, что предыдущий набор завершен.

Я надеюсь, что я ответил на ваш вопрос. Если вы предоставите больше информации о прецеденте, я попытаюсь улучшить свой ответ.

С наилучшими пожеланиями

Edit:

Альтернативное решение:

Чтобы иметь порядок сохраняется все Наблюдаемые должны быть подписаны, чтобы они должны прибыть. Договор Observable является наблюдаемым (если он холодный) не выполняется до подписки. И при использовании flatMap() второй Observable подписывается после того, как первый из них испустит.

Чтобы иметь как пишет передаются в порядке, они должны быть подписаны в том же порядке, так что поток может выглядеть следующим образом:

connectionObservable 
      .flatMap(rxBleConnection -> { 
       Observable<byte[]> mergedObservable = null; 
       for (byte[] bytes : b) { 
        Log.d("Observer", Arrays.toString(bytes)); 
        final Observable<byte[]> writeObservable = rxBleConnection 
          .writeCharacteristic(uuid, bytes); 

        if (mergedObservable == null) { 
         mergedObservable = writeObservable; 
        } else { 
         // merging two Observables to be subscribed at the same time when subscribed 
         mergedObservable = mergedObservable.mergeWith(writeObservable); 
        } 
       } 
       return mergedObservable; 
      }) 
      // removed .concatMap() 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(
        bytes -> Log.d("Subscriber", Arrays.toString(bytes)), 
        throwable -> Log.e("Subscriber", "error", throwable) 
      ); 

RxJava имеет явно больше способов для достижения такого же поведения, но это не часть этого вопроса.