Performance about rxandroidble HOT 5 CLOSED

Observable.concat(
        perform(device, this::unoptimizedSendUuid).map(aFloat -> "Unoptimized UUID: " + String.valueOf(aFloat) + " Bps"),
        Observable.<String>empty().delay(1, TimeUnit.SECONDS),
        perform(device, this::unoptimizedSendCharacteristic).map(aFloat -> "Unoptimized Char: " + String.valueOf(aFloat) + " Bps"),
        Observable.<String>empty().delay(1, TimeUnit.SECONDS),
        perform(device, this::longWriteSend).map(aFloat -> "Long Write: " + String.valueOf(aFloat) + " Bps"),
        Observable.<String>empty().delay(1, TimeUnit.SECONDS),
        perform(device, this::optimizedSend).map(aFloat -> "Optimized: " + String.valueOf(aFloat) + " Bps")
)
        .observeOn(AndroidSchedulers.mainThread())
        .doOnUnsubscribe(this::clearSubscription)
        .subscribe(
                result -> Log.e("RESULT", result),
                e -> Log.e("ERROR", "Whoops!", e)
        );

private Observable<Float> perform(RxBleDevice device, TestSetup testSetup) {
    return device.establishConnection(false)
            .flatMap(RxBleConnection::discoverServices,
                    (connection, services) -> services.getCharacteristic(genericCommunicationCharacteristicUuid)
                            .flatMap(
                                    connection::setupNotification,
                                    (characteristic, responseObs) -> {
                                        characteristic.setWriteType(BluetoothGattCharacteristic.WRITE_TYPE_NO_RESPONSE);
                                        final Completable readyForTransmission =
                                                responseObs.filter(bytes -> bytes[0] == 0x01).first().toCompletable();
                                        final Observable<byte[]> speedTest = testSetup
                                                .create(new TestConnection(connection, responseObs, characteristic, 19));
                                        return readyForTransmission.andThen(speedTest);
                                    }
                            )
                            .flatMap(observable -> observable)
            )
            .flatMap(observable -> observable)
            .timestamp()
            .buffer(1, TimeUnit.MINUTES)
            .take(1)
            .map(timestampeds -> {
                final int bytesSent = (timestampeds.size() - 1) * 20 * 19;
                final float timeSeconds = (timestampeds.get(timestampeds.size() - 1).getTimestampMillis() - timestampeds.get(0).getTimestampMillis()) * 0.001f;
                return bytesSent / timeSeconds;
            });
}

interface TestSetup {

    Observable<byte[]> create(TestConnection testConnection);
}

static class TestConnection {

    final RxBleConnection connection;

    final Observable<byte[]> responseObs;

    final BluetoothGattCharacteristic characteristic;

    final int batchCount;

    public TestConnection(RxBleConnection connection, Observable<byte[]> responseObs,
                          BluetoothGattCharacteristic characteristic, int batchCount) {
        this.connection = connection;
        this.responseObs = responseObs;
        this.characteristic = characteristic;
        this.batchCount = batchCount;
    }
}

private Observable<byte[]> unoptimizedSendUuid(TestConnection testConnection) {
    final RxBleConnection connection = testConnection.connection;
    final BluetoothGattCharacteristic characteristic = testConnection.characteristic;
    final UUID uuid = characteristic.getUuid();
    return Observable.range(0, testConnection.batchCount)
            .concatMap(frameIndex -> connection.writeCharacteristic(
                    uuid,
                    new byte[]{
                            frameIndex.byteValue(), 0, 0, 0, 0, 0, 0, 0, 0,
                            0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
                    }
            ))
            .ignoreElements()
            .repeatWhen(observable -> Observable.zip(observable, testConnection.responseObs, (o, bytes) -> o))
            .mergeWith(testConnection.responseObs);
}

private Observable<byte[]> unoptimizedSendCharacteristic(TestConnection testConnection) {
    final RxBleConnection connection = testConnection.connection;
    final BluetoothGattCharacteristic characteristic = testConnection.characteristic;
    return Observable.range(0, testConnection.batchCount)
            .concatMap(frameIndex -> connection.writeCharacteristic(
                    characteristic,
                    new byte[]{
                            frameIndex.byteValue(), 0, 0, 0, 0, 0, 0, 0, 0,
                            0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
                    }
            ))
            .ignoreElements()
            .repeatWhen(observable -> Observable.zip(observable, testConnection.responseObs, (o, bytes) -> o))
            .mergeWith(testConnection.responseObs);
}

private Observable<byte[]> longWriteSend(TestConnection testConnection) {
    final byte[] bytesToSend = new byte[20 * 19];
    for (int i = 0; i < bytesToSend.length; i = i + 20) {
        bytesToSend[i] = (byte) (i / 20);
    }
    return testConnection.connection
            .createNewLongWriteBuilder()
            .setBytes(bytesToSend)
            .setCharacteristic(testConnection.characteristic)
            .build()
            .ignoreElements()
            .repeatWhen(observable -> Observable.zip(observable, testConnection.responseObs, (o, bytes) -> o))
            .mergeWith(testConnection.responseObs);
}

private Observable<byte[]> optimizedSend(TestConnection testConnection) {
    return testConnection.connection.queue((bluetoothGatt, rxBleGattCallback, scheduler) -> Observable.create(
            emitter -> {
                Log.i("START", String.valueOf(testConnection.batchCount));
                final byte[] data = new byte[20];
                testConnection.characteristic.setValue(data);
                final AtomicBoolean writeCompleted = new AtomicBoolean(false);
                final AtomicBoolean ackCompleted = new AtomicBoolean(false);
                final AtomicInteger batchesSent = new AtomicInteger(0);
                final Runnable writeNextBatch = () -> {
                    data[0]++;
                    if (!bluetoothGatt.writeCharacteristic(testConnection.characteristic)) {
                        emitter.onError(new BleGattCannotStartException(bluetoothGatt, BleGattOperationType.CHARACTERISTIC_WRITE));
                    } else {
                        Log.i("SEND", String.valueOf(data[0]));
                        batchesSent.incrementAndGet();
                    }
                };
                final BluetoothGattCallback bluetoothGattCallback = new BluetoothGattCallback() {
                    @Override
                    public void onCharacteristicWrite(BluetoothGatt gatt, BluetoothGattCharacteristic characteristic, int status) {
                        if (status != BluetoothGatt.GATT_SUCCESS) {
                            emitter.onError(new BleGattException(gatt, status, BleGattOperationType.CHARACTERISTIC_WRITE));
                        } else if (batchesSent.get() == testConnection.batchCount) {
                            if (ackCompleted.get()) {
                                batchesSent.set(0);
                                ackCompleted.set(false);
                                emitter.onNext(null);
                                writeNextBatch.run();
                            } else {
                                writeCompleted.set(true);
                            }
                        } else {
                            characteristic.setValue(data);
                            writeNextBatch.run();
                        }
                    }

                    @Override
                    public void onCharacteristicChanged(BluetoothGatt gatt, BluetoothGattCharacteristic characteristic) {
                        final byte[] bytes = characteristic.getValue();
                        Log.i("ACK", Arrays.toString(bytes) + "/" + bytes.length + "/" + System.identityHashCode(bytes));
                        characteristic.setValue(data);
                        if (writeCompleted.get()) {
                            batchesSent.set(0);
                            writeCompleted.set(false);
                            emitter.onNext(null);
                            writeNextBatch.run();
                        } else {
                            ackCompleted.set(true);
                        }
                    }
                };

                rxBleGattCallback.setNativeCallback(bluetoothGattCallback);

                Log.i("SEND", String.valueOf(data[0]));
                if (!bluetoothGatt.writeCharacteristic(testConnection.characteristic)) {
                    emitter.onError(new BleGattCannotStartException(bluetoothGatt, BleGattOperationType.CHARACTERISTIC_WRITE));
                } else {
                    batchesSent.incrementAndGet();
                }
            },
            Emitter.BackpressureMode.NONE
    ));
}