Add support for long-writes

[nister]

Summary

Ex: http://stackoverflow.com/questions/24135682/android-sending-data-20-bytes-by-ble

Need to send data > 20 bytes via BLE. It works normally if I using common Android BLE API. But I get only first part (20 bytes) with RxAndroidBle. Behavior is the same in case with .setupNotification, readCharacteristic, writeCharacteristic.

How can I handle it with RxAndroidBle and\or what I'm doing wrong?

Preconditions

Steps to reproduce actual result

1. Send any data from BLE device > 20 bytes

Actual result

Only fiirst part receiving

Expected result

Get all data by parts

Thanks!

[zhuozhibin]

now I face the same problem, any solution?

[dariuszseweryn]

Sorry for such a long wait. As for now you have to chunk the data yourself. I am currently chasing bugs mostly and unfortunately don't have enough time to research the problem and possible use cases.
Feel free to describe it to me in greater detail. :)

[maoueh]

If it can help others, here the code I use to do the split. I release this to public domain without restrictions. Use (and adapt) at your own risk (note this is a slightly modified version of the class I use in my own project):

Usage:

bleClient.establishConnection().flatMap(connection -> {
  return new PacketWriter(connection).write(characteristic, bytes);
});

// PacketWriter.java
package co.acme.ble;

import com.google.common.annotations.VisibleForTesting;
import com.polidea.rxandroidble.RxBleConnection;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;

import rx.Observable;

public class PacketWriter {

  private static final int MAX_BLE_PACKET_BYTE_COUNT = 20;

  private RxBleConnection mBleConnection;

  public PacketWriter(RxBleConnection bleConnection) {
    mBleConnection = bleConnection;
  }

  /**
   * FIXME: Make this more efficient by keeping a 20 bytes buffer that is re-used for full packet send instead of
   * copying to a new array over and over. We could even think about a specialized {@link Iterable} that would do the
   * split and caching "magically" and we would simply need to iterate over it.
   */
  public Observable<Void> write(UUID characteristic, byte[] bytes) {
    if (bytes == null || bytes.length <= 0) {
      return Observable.just(null);
    }

    return Observable.concat(computeWritePacketObservables(characteristic, bytes));
  }

  @VisibleForTesting
  int computePacketCount(byte[] bytes) {
    return (int) Math.ceil(bytes.length / (double) MAX_BLE_PACKET_BYTE_COUNT);
  }

  @VisibleForTesting
  List<Observable<Void>> computeWritePacketObservables(UUID characteristic, byte[] bytes) {
    int packetCount = computePacketCount(bytes);
    int lastPacketIndex = packetCount - 1;

    List<Observable<Void>> observables = new ArrayList<>();
    for (int i = 0; i < packetCount; ++i) {
      boolean isLastPacket = i == lastPacketIndex;

      int dataStartIndexInclusive = i * 20;
      int dataEndIndexInclusive = isLastPacket ? bytes.length : dataStartIndexInclusive + MAX_BLE_PACKET_BYTE_COUNT;

      observables.add(writeToCharacteristic(characteristic, bytes, dataStartIndexInclusive, dataEndIndexInclusive));
    }

    return observables;
  }

  private Observable<Void> writeToCharacteristic(UUID characteristic,
                                                 byte[] bytes,
                                                 int startIndexInclusive,
                                                 int endIndexExclusive) {
    return Observable.fromCallable(() -> Arrays.copyOfRange(bytes, startIndexInclusive, endIndexExclusive))
                     .flatMap(data -> {
                       return mBleConnection.writeCharacteristic(characteristic, data);
                     })
                     .map(ignore -> null);
  }
}

And the test class I used:

// PacketWriterTest.java
package co.acme.ble;

import com.polidea.rxandroidble.RxBleConnection;

import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import rx.Observable;
import rx.Subscriber;

import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
import static org.mockito.BDDMockito.given;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.verify;
import static org.mockito.internal.verification.VerificationModeFactory.times;

public class PacketWriterTest {

  @Mock
  RxBleConnection mBleConnection;

  private PacketWriter mPacketWriter;
  private RecordingSubscriber<Void> mRecordingSubscriber = new RecordingSubscriber<>();

  @Before
  public void setUp() {
    MockitoAnnotations.initMocks(this);

    given(mBleConnection.writeCharacteristic(any(UUID.class),
                                             any(byte[].class))).willReturn(Observable.just(new byte[]{}));

    mPacketWriter = new PacketWriter(mBleConnection);
  }

  @Test
  public void testWriteSinglePacket() throws Exception {
    writePacket("0102");

    ArgumentCaptor<byte[]> captor = ArgumentCaptor.forClass(byte[].class);
    verify(mBleConnection, times(1)).writeCharacteristic(any(UUID.class), captor.capture());

    assertThat(captor.getValue(), is(payload("0102")));
  }

  @Test
  public void testWriteSinglePacketFlush() throws Exception {
    writePacket("0102030405060708091011121314151617181920");

    ArgumentCaptor<byte[]> captor = ArgumentCaptor.forClass(byte[].class);
    verify(mBleConnection, times(1)).writeCharacteristic(any(UUID.class), captor.capture());

    assertThat(captor.getValue(), is(payload("0102030405060708091011121314151617181920")));
  }

  @Test
  public void testWriteTwoPacket() throws Exception {
    writePacket("010203040506070809101112131415161718192021222324252627282930");

    ArgumentCaptor<byte[]> captor = ArgumentCaptor.forClass(byte[].class);
    verify(mBleConnection, times(2)).writeCharacteristic(any(UUID.class), captor.capture());

    assertThat(captor.getAllValues().get(0), is(payload("0102030405060708091011121314151617181920")));
    assertThat(captor.getAllValues().get(1), is(payload("21222324252627282930")));
  }

  @Test
  public void testWriteTwoPacketFlush() throws Exception {
    writePacket("01020304050607080910111213141516171819200102030405060708091011121314151617181920");

    ArgumentCaptor<byte[]> captor = ArgumentCaptor.forClass(byte[].class);
    verify(mBleConnection, times(2)).writeCharacteristic(any(UUID.class), captor.capture());

    assertThat(captor.getAllValues().get(0), is(payload("0102030405060708091011121314151617181920")));
    assertThat(captor.getAllValues().get(1), is(payload("0102030405060708091011121314151617181920")));
  }

  @Test
  public void testComputePacketCount() {
    assertThat(mPacketWriter.computePacketCount(payload("01")), is(1));
    assertThat(mPacketWriter.computePacketCount(payload("01020304050607080910111213141516171819")), is(1));
    assertThat(mPacketWriter.computePacketCount(payload("0102030405060708091011121314151617181920")), is(1));

    assertThat(mPacketWriter.computePacketCount(payload("010203040506070809101112131415161718192021")), is(2));
    assertThat(mPacketWriter.computePacketCount(payload("0102030405060708091011121314151617181920" +
                                                        "0102030405060708091011121314151617181920")), is(2));

    assertThat(mPacketWriter.computePacketCount(payload("0102030405060708091011121314151617181920" +
                                                        "010203040506070809101112131415161718192021")), is(3));
  }

  private void writePacket(String hexadecimalValue) throws Exception {
    mPacketWriter.write(UUID.randomUUID(), payload(hexadecimalValue)).toBlocking().subscribe(mRecordingSubscriber);

    if (mRecordingSubscriber.getOnErrorException() != null) {
      throw (Exception) mRecordingSubscriber.getOnErrorException();
    }
  }

  private byte[] payload(String input) {
    int length = input.length();
    byte[] data = new byte[length / 2];
    for (int i = 0; i < length; i += 2) {
      data[i / 2] = (byte) ((Character.digit(input.charAt(i), 16) << 4) + Character.digit(input.charAt(i + 1), 16));
    }

    return data;
  }

  private static class RecordingSubscriber<T> extends Subscriber<T> {

    private List<T> mEmittedElements = new ArrayList<>();
    private Throwable mOnErrorException;
    private boolean mIsCompletedCalled;

    @Override
    public void onCompleted() {
      mIsCompletedCalled = true;
    }

    @Override
    public void onError(Throwable exception) {
      mOnErrorException = exception;
    }

    @Override
    public void onNext(T element) {
      mEmittedElements.add(element);
    }

    public List<T> getEmittedElements() {
      return mEmittedElements;
    }

    public T getFirstElement() {
      return mEmittedElements.get(0);
    }

    public Throwable getOnErrorException() {
      return mOnErrorException;
    }

    public boolean isCompletedCalled() {
      return mIsCompletedCalled;
    }
  }
}

Good luck.

Regards,
Matt

[pregno]

this could be managed changing the MTU

[dariuszseweryn]

I have put somewhere what I do use for byte splitting but cannot find it for reference so I will put it here again:

public class ByteSplitter extends Observable<byte[]> {
    
    protected ByteSplitter(@NonNull byte[] bytes) {
        super(new SyncOnSubscribe<ByteBuffer, byte[]>() {

            private static final int MAX_PACKET_LENGTH = 20;

            @Override
            protected ByteBuffer generateState() {
                return ByteBuffer.wrap(bytes);
            }

            @Override
            protected ByteBuffer next(ByteBuffer state, Observer<? super byte[]> observer) {
                final int remainingBytes = state.remaining();
                final int bytePackageLength = Math.min(remainingBytes, MAX_PACKET_LENGTH);
                if (bytePackageLength > 0) {
                    final byte[] bytesPacket = new byte[bytePackageLength];
                    state.get(bytesPacket);
                    observer.onNext(bytesPacket);
                }
                if (state.remaining() == 0) {
                    observer.onCompleted();
                }
                return state;
            }
        });
    }
}

[Zkffkah]

@dariuszseweryn #99 Here comes the reference.

[dariuszseweryn]

@nister Could you elaborate more on how did you managed to send more than 20 bytes of data using the Android BLE API? The topic on stackoverflow is describing a custom BLE protocol in order to introduce some flow control in the communication. It describes nothing that the library is not supporting.

[tahir9111]

Are we getting this support soon? Byte>20 transmit? How can we achieve it with current library, with bytesplitter

[pregno]

If you are using Android 5.x or later you can set the MTU > 20 before sending the data

[dariuszseweryn]

And if you use 4.3 / 4.4 or your peripheral does not support larger MTUs - you can just make several writes one after another with the next batches of what you want to transmit. The bytesplitter just splits your large byte array into batches that can be written at once - one after another.

I have finished a helper but it is still under a code review.

[dariuszseweryn]

Functionality added with:
a2e85fc
and
b85fbf2

[nister]

@dariuszseweryn Thanks!