Added retry mechanism parameter for LongWriteOperation

mockrxandroidble/src/main/java/com/polidea/rxandroidble/mockrxandroidble/RxBleConnectionMock.java

@@ -10,6 +10,7 @@
 import com.polidea.rxandroidble.RxBleDeviceServices;
 import com.polidea.rxandroidble.exceptions.BleConflictingNotificationAlreadySetException;
 import com.polidea.rxandroidble.internal.connection.ImmediateSerializedBatchAckStrategy;
+import com.polidea.rxandroidble.internal.connection.NoRetryStrategy;
 import com.polidea.rxandroidble.internal.util.ObservableUtil;
 
 import java.util.HashMap;
@@ -283,6 +284,8 @@ public LongWriteOperationBuilder createNewLongWriteBuilder() {
 
  private WriteOperationAckStrategy writeOperationAckStrategy = // default
  new ImmediateSerializedBatchAckStrategy();
+ private WriteOperationRetryStrategy writeOperationRetryStrategy = // default
+ new NoRetryStrategy();
 
  @Override
  public LongWriteOperationBuilder setBytes(@NonNull byte[] bytes) {
@@ -316,6 +319,13 @@ public LongWriteOperationBuilder setMaxBatchSize(int maxBatchSize) {
  return this;
  }
 
+ @Override
+ public LongWriteOperationBuilder setWriteOperationRetryStrategy(
+ @NonNull WriteOperationRetryStrategy writeOperationRetryStrategy) {
+ this.writeOperationRetryStrategy = writeOperationRetryStrategy;
+ return this;
+ }
+
  @Override
  public LongWriteOperationBuilder setWriteOperationAckStrategy(@NonNull WriteOperationAckStrategy writeOperationAckStrategy) {
  this.writeOperationAckStrategy = writeOperationAckStrategy;

rxandroidble/src/main/java/com/polidea/rxandroidble/RxBleConnection.java

@@ -27,6 +27,7 @@
 import rx.Completable;
 import rx.Observable;
 import rx.Scheduler;
+import rx.functions.Func2;
 
 /**
  * The BLE connection handle, supporting GATT operations. Operations are enqueued and the library makes sure that they are not
@@ -131,6 +132,13 @@ interface LongWriteOperationBuilder {
  */
  LongWriteOperationBuilder setMaxBatchSize(@IntRange(from = 1, to = GATT_MTU_MAXIMUM - GATT_WRITE_MTU_OVERHEAD) int maxBatchSize);
 
+ /**
+ * Setter for a retry strategy in case something goes wrong when writing data.
+ * @param writeOperationRetryStrategy the retry strategy
+ * @return the LongWriteOperationBuilder
+ */
+ LongWriteOperationBuilder setWriteOperationRetryStrategy(@NonNull WriteOperationRetryStrategy writeOperationRetryStrategy);
+
  /**
  * Setter for a strategy used to mark batch write completed. Only after previous batch has finished, the next (if any left) can be
  * written.
@@ -442,6 +450,9 @@ Observable<byte[]> writeDescriptor(@NonNull UUID serviceUuid, @NonNull UUID char
  */
  Observable<byte[]> writeDescriptor(@NonNull BluetoothGattDescriptor descriptor, @NonNull byte[] data);
 
+ interface WriteOperationRetryStrategy extends Func2<Integer, Throwable, Boolean> {
+
+ }
 
  /**
  * Performs a GATT request connection priority operation, which requests a connection parameter

rxandroidble/src/main/java/com/polidea/rxandroidble/internal/connection/LongWriteOperationBuilderImpl.java

@@ -23,6 +23,7 @@ public final class LongWriteOperationBuilderImpl implements RxBleConnection.Long
  private Observable<BluetoothGattCharacteristic> writtenCharacteristicObservable;
  private PayloadSizeLimitProvider maxBatchSizeProvider;
  private RxBleConnection.WriteOperationAckStrategy writeOperationAckStrategy = new ImmediateSerializedBatchAckStrategy();
+ private RxBleConnection.WriteOperationRetryStrategy writeOperationRetryStrategy = new NoRetryStrategy();
 
  private byte[] bytes;
 
@@ -63,6 +64,13 @@ public RxBleConnection.LongWriteOperationBuilder setMaxBatchSize(final int maxBa
  return this;
  }
 
+ @Override
+ public RxBleConnection.LongWriteOperationBuilder setWriteOperationRetryStrategy(
+ @NonNull RxBleConnection.WriteOperationRetryStrategy writeOperationRetryStrategy) {
+ this.writeOperationRetryStrategy = writeOperationRetryStrategy;
+ return this;
+ }
+
  @Override
  public RxBleConnection.LongWriteOperationBuilder setWriteOperationAckStrategy(
  @NonNull RxBleConnection.WriteOperationAckStrategy writeOperationAckStrategy) {
@@ -87,7 +95,7 @@ public Observable<byte[]> build() {
  public Observable<byte[]> call(BluetoothGattCharacteristic bluetoothGattCharacteristic) {
  return operationQueue.queue(
  operationsProvider.provideLongWriteOperation(bluetoothGattCharacteristic,
- writeOperationAckStrategy, maxBatchSizeProvider, bytes)
+ writeOperationAckStrategy, writeOperationRetryStrategy, maxBatchSizeProvider, bytes)
  );
  }
  });

rxandroidble/src/main/java/com/polidea/rxandroidble/internal/connection/NoRetryStrategy.java

@@ -0,0 +1,11 @@
+package com.polidea.rxandroidble.internal.connection;
+
+import com.polidea.rxandroidble.RxBleConnection;
+
+public class NoRetryStrategy implements RxBleConnection.WriteOperationRetryStrategy {
+
+ @Override
+ public Boolean call(Integer integer, Throwable throwable) {
+ return false;
+ }
+}

rxandroidble/src/main/java/com/polidea/rxandroidble/internal/operations/CharacteristicLongWriteOperation.java

@@ -8,6 +8,7 @@
 
 import com.polidea.rxandroidble.ClientComponent;
 import com.polidea.rxandroidble.RxBleConnection.WriteOperationAckStrategy;
+import com.polidea.rxandroidble.RxBleConnection.WriteOperationRetryStrategy;
 import com.polidea.rxandroidble.exceptions.BleDisconnectedException;
 import com.polidea.rxandroidble.exceptions.BleException;
 import com.polidea.rxandroidble.exceptions.BleGattCallbackTimeoutException;
@@ -33,6 +34,7 @@
 import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Func1;
+import rx.functions.Func2;
 
 public class CharacteristicLongWriteOperation extends QueueOperation<byte[]> {
 
@@ -43,8 +45,10 @@ public class CharacteristicLongWriteOperation extends QueueOperation<byte[]> {
  private final BluetoothGattCharacteristic bluetoothGattCharacteristic;
  private final PayloadSizeLimitProvider batchSizeProvider;
  private final WriteOperationAckStrategy writeOperationAckStrategy;
+ private final WriteOperationRetryStrategy writeOperationRetryStrategy;
  private final byte[] bytesToWrite;
  private byte[] tempBatchArray;
+ private int retryCounter;
 
  CharacteristicLongWriteOperation(
  BluetoothGatt bluetoothGatt,
@@ -54,6 +58,7 @@ public class CharacteristicLongWriteOperation extends QueueOperation<byte[]> {
  BluetoothGattCharacteristic bluetoothGattCharacteristic,
  PayloadSizeLimitProvider batchSizeProvider,
  WriteOperationAckStrategy writeOperationAckStrategy,
+ WriteOperationRetryStrategy writeOperationRetryStrategy,
  byte[] bytesToWrite) {
  this.bluetoothGatt = bluetoothGatt;
  this.rxBleGattCallback = rxBleGattCallback;
@@ -62,12 +67,14 @@ public class CharacteristicLongWriteOperation extends QueueOperation<byte[]> {
  this.bluetoothGattCharacteristic = bluetoothGattCharacteristic;
  this.batchSizeProvider = batchSizeProvider;
  this.writeOperationAckStrategy = writeOperationAckStrategy;
+ this.writeOperationRetryStrategy = writeOperationRetryStrategy;
  this.bytesToWrite = bytesToWrite;
+ this.retryCounter = 0;
  }
 
  @Override
  protected void protectedRun(final Emitter<byte[]> emitter, final QueueReleaseInterface queueReleaseInterface) throws Throwable {
- int batchSize = batchSizeProvider.getPayloadSizeLimit();
+ final int batchSize = batchSizeProvider.getPayloadSizeLimit();
 
  if (batchSize <= 0) {
  throw new IllegalArgumentException("batchSizeProvider value must be greater than zero (now: " + batchSize + ")");
@@ -87,9 +94,11 @@ protected void protectedRun(final Emitter<byte[]> emitter, final QueueReleaseInt
  timeoutObservable,
  timeoutConfiguration.timeoutScheduler
  )
+ .doOnNext(resetRetryCounter())
  .repeatWhen(bufferIsNotEmptyAndOperationHasBeenAcknowledgedAndNotUnsubscribed(
  writeOperationAckStrategy, byteBuffer, emitterWrapper
  ))
+ .retry(retryStrategy(byteBuffer, batchSize))
  .toCompletable()
  .subscribe(
  new Action0() {
@@ -209,4 +218,30 @@ public Boolean call(Object emission) {
  }
  };
  }
+
+ private Action1<ByteAssociation<UUID>> resetRetryCounter() {
+ return new Action1<ByteAssociation<UUID>>() {
+ @Override
+ public void call(ByteAssociation<UUID> uuidByteAssociation) {
+ retryCounter = 0;
+ }
+ };
+ }
+
+ private Func2<Integer, Throwable, Boolean> retryStrategy(final ByteBuffer byteBuffer, final int batchSize) {
+ return new Func2<Integer, Throwable, Boolean>() {
+ @Override
+ public Boolean call(Integer integer, Throwable throwable) {
+ // Individual counter for each batch payload.
+ retryCounter++;
+ final Boolean retry = writeOperationRetryStrategy.call(retryCounter, throwable);
+ if (!retry) {
+ return false;
+ }
+ // Reset buffer to last position
+ byteBuffer.position(byteBuffer.position() - batchSize);
+ return true;
+ }
+ };
+ }
 }

rxandroidble/src/main/java/com/polidea/rxandroidble/internal/operations/OperationsProvider.java

@@ -15,6 +15,7 @@ public interface OperationsProvider {
  CharacteristicLongWriteOperation provideLongWriteOperation(
  BluetoothGattCharacteristic bluetoothGattCharacteristic,
  RxBleConnection.WriteOperationAckStrategy writeOperationAckStrategy,
+ RxBleConnection.WriteOperationRetryStrategy writeOperationRetryStrategy,
  PayloadSizeLimitProvider maxBatchSizeProvider,
  byte[] bytes);
 

rxandroidble/src/main/java/com/polidea/rxandroidble/internal/operations/OperationsProviderImpl.java

@@ -49,6 +49,7 @@ public class OperationsProviderImpl implements OperationsProvider {
  public CharacteristicLongWriteOperation provideLongWriteOperation(
  BluetoothGattCharacteristic bluetoothGattCharacteristic,
  RxBleConnection.WriteOperationAckStrategy writeOperationAckStrategy,
+ RxBleConnection.WriteOperationRetryStrategy writeOperationRetryStrategy,
  PayloadSizeLimitProvider maxBatchSizeProvider,
  byte[] bytes) {
 
@@ -59,6 +60,7 @@ public CharacteristicLongWriteOperation provideLongWriteOperation(
  bluetoothGattCharacteristic,
  maxBatchSizeProvider,
  writeOperationAckStrategy,
+ writeOperationRetryStrategy,
  bytes);
  }