Implement queue (#11)

* Implement queue

* Enable queue by default, remove queue from web and update readme

* Rename useQueue to queuesRequests

* Rename queuesRequests and improve readme

* Rename _executeMethod and improve readme

* Rename to _isQueueEnabled

---------

Co-authored-by: Foti Dim <fdimanidis@gmail.com>

lib/src/ble_command_queue.dart

@@ -0,0 +1,80 @@
+import 'dart:async';
+
+/// Original Author: Ryan Knell (https://github.com/rknell/dart_queue)
+
+/// Queue to execute Futures in order.
+/// It awaits each future before executing the next one.
+class BleCommandQueue {
+  final Set<int> _activeItems = {};
+  int _lastProcessId = 0;
+  bool _isCancelled = false;
+  final List<_QueuedFuture> _nextCycle = [];
+
+  Future<T> add<T>(Future<T> Function() closure, {Duration? timeout}) {
+    if (_isCancelled) throw Exception('Queue Cancelled');
+    final completer = Completer<T>();
+    _nextCycle.add(_QueuedFuture<T>(closure, completer, timeout));
+    _updateRemainingItems();
+    if (_activeItems.isEmpty) _queueUpNext();
+    return completer.future;
+  }
+
+  void dispose() {
+    for (final item in _nextCycle) {
+      item.completer.completeError(Exception('Queue Cancelled'));
+    }
+    _nextCycle.removeWhere((item) => item.completer.isCompleted);
+    _isCancelled = true;
+  }
+
+  void _queueUpNext() {
+    if (_nextCycle.isNotEmpty && !_isCancelled && _activeItems.length <= 1) {
+      final processId = _lastProcessId;
+      _activeItems.add(processId);
+      final item = _nextCycle.first;
+      _lastProcessId++;
+      _nextCycle.remove(item);
+      item.onComplete = () async {
+        _activeItems.remove(processId);
+        _updateRemainingItems();
+        _queueUpNext();
+      };
+      unawaited(item.execute());
+    }
+  }
+
+  void _updateRemainingItems() {
+    // int remainingQueueItems = _nextCycle.length + _activeItems.length;
+    // onRemainingItemsUpdate?.call(_);
+  }
+}
+
+class _QueuedFuture<T> {
+  final Completer completer;
+  final Future<T> Function() closure;
+  Function? onComplete;
+  final Duration? timeout;
+
+  _QueuedFuture(this.closure, this.completer, this.timeout, {this.onComplete});
+
+  Future<void> execute() async {
+    try {
+      T result;
+      if (timeout != null) {
+        result = await closure().timeout(timeout!);
+      } else {
+        result = await closure();
+      }
+      if (result != null) {
+        completer.complete(result);
+      } else {
+        completer.complete(null);
+      }
+      await Future.microtask(() {});
+    } catch (e, stack) {
+      completer.completeError(e, stack);
+    } finally {
+      if (onComplete != null) onComplete?.call();
+    }
+  }
+}

lib/src/universal_ble.dart

@@ -1,4 +1,4 @@
-import 'dart:io';
+import 'dart:async';
 
 import 'package:flutter/foundation.dart';
 import 'package:universal_ble/src/universal_ble_linux/universal_ble_linux.dart';
@@ -14,10 +14,30 @@ class UniversalBle {
   static void setInstance(UniversalBlePlatform instance) =>
       _platform = instance;
 
+  /// Set global timeout for all commands
+  static Duration? timeout = const Duration(seconds: 10);
+
+  static BleCommandQueue? _queue = BleCommandQueue();
+
+  /// Setup global queue for all commands, by default queue is enabled
+  static set queuesCommands(bool value) {
+    if (value) {
+      _queue ??= BleCommandQueue();
+      UniversalBlePlatform.logInfo('Queue enabled');
+    } else {
+      _queue?.dispose();
+      _queue = null;
+      UniversalBlePlatform.logInfo('Queue disabled');
+    }
+  }
+
   /// To get Bluetooth state availability
   /// To get updates, set [onAvailabilityChange] listener
   static Future<AvailabilityState> getBluetoothAvailabilityState() async {
-    return await _platform.getBluetoothAvailabilityState();
+    return await _executeCommand(
+      () => _platform.getBluetoothAvailabilityState(),
+      timeout: timeout,
+    );
   }
 
   /// To Start scan, get scan results in [onScanResult] listener
@@ -26,13 +46,19 @@ class UniversalBle {
   static Future<void> startScan({
     WebRequestOptionsBuilder? webRequestOptions,
   }) async {
-    await _platform.startScan(webRequestOptions: webRequestOptions);
+    return await _executeCommand(
+      () => _platform.startScan(webRequestOptions: webRequestOptions),
+      timeout: null,
+    );
   }
 
   /// To Stop scan, set [onScanResult] listener to `null` if you don't need it anymore
   /// might throw errors if Bluetooth is not available
   static Future<void> stopScan() async {
-    await _platform.stopScan();
+    return await _executeCommand(
+      () => _platform.stopScan(),
+      timeout: null,
+    );
   }
 
   /// To connect to a device, get connection state in [onConnectionChanged] listener
@@ -43,17 +69,26 @@ class UniversalBle {
     String deviceId, {
     Duration? connectionTimeout,
   }) async {
-    await _platform.connect(deviceId, connectionTimeout: connectionTimeout);
+    return await _executeCommand(
+      () => _platform.connect(deviceId, connectionTimeout: connectionTimeout),
+      timeout: timeout,
+    );
   }
 
   /// To disconnect from a device, get connection state in [onConnectionChanged] listener
   static Future<void> disconnect(String deviceId) async {
-    await _platform.disconnect(deviceId);
+    return await _executeCommand(
+      () => _platform.disconnect(deviceId),
+      timeout: timeout,
+    );
   }
 
   /// To discover services of a device
   static Future<List<BleService>> discoverServices(String deviceId) async {
-    return await _platform.discoverServices(deviceId);
+    return await _executeCommand(
+      () => _platform.discoverServices(deviceId),
+      timeout: timeout,
+    );
   }
 
   /// To set a characteristic notifiable, set `bleInputProperty` to [BleInputProperty.notification] or [BleInputProperty.indication], get updates in [onValueChanged] listener
@@ -64,22 +99,28 @@ class UniversalBle {
     String characteristic,
     BleInputProperty bleInputProperty,
   ) async {
-    await _platform.setNotifiable(
-      deviceId,
-      service,
-      characteristic,
-      bleInputProperty,
+    return await _executeCommand(
+      () => _platform.setNotifiable(
+        deviceId,
+        service,
+        characteristic,
+        bleInputProperty,
+      ),
+      timeout: timeout,
     );
   }
 
   /// To read a characteristic value
-  /// on iOS and MacOS, this method will also trigger [onValueChanged] listener
+  /// on iOS and MacOS, this command will also trigger [onValueChanged] listener
   static Future<Uint8List> readValue(
     String deviceId,
     String service,
     String characteristic,
   ) async {
-    return await _platform.readValue(deviceId, service, characteristic);
+    return await _executeCommand(
+      () => _platform.readValue(deviceId, service, characteristic),
+      timeout: timeout,
+    );
   }
 
   /// To write a characteristic value
@@ -91,33 +132,50 @@ class UniversalBle {
     Uint8List value,
     BleOutputProperty bleOutputProperty,
   ) async {
-    await _platform.writeValue(
-      deviceId,
-      service,
-      characteristic,
-      value,
-      bleOutputProperty,
+    await _executeCommand(
+      () => _platform.writeValue(
+        deviceId,
+        service,
+        characteristic,
+        value,
+        bleOutputProperty,
+      ),
+      timeout: timeout,
     );
   }
 
   /// `requestMtu` not supported on `Linux` and `Web
   static Future<int> requestMtu(String deviceId, int expectedMtu) async {
-    return await _platform.requestMtu(deviceId, expectedMtu);
+    return await _executeCommand(
+      () => _platform.requestMtu(deviceId, expectedMtu),
+      timeout: timeout,
+    );
   }
 
-  /// Pair methods are not supported on `iOS`, `MacOS` and `Web`
+  /// Pair commands are not supported on `iOS`, `MacOS` and `Web`
   static Future<bool> isPaired(String deviceId) async {
-    return await _platform.isPaired(deviceId);
+    return await _executeCommand(
+      () => _platform.isPaired(deviceId),
+      timeout: timeout,
+    );
   }
 
-  /// To trigger pair request, might throw errors if device is already paired
+  /// To trigger pair request
+  /// might throw errors if device is already paired
   static Future<void> pair(String deviceId) async {
-    await _platform.pair(deviceId);
+    return await _executeCommand(
+      () => _platform.pair(deviceId),
+      timeout: timeout,
+    );
   }
 
-  /// To trigger unPair request, might throw errors if device is not paired
+  /// To trigger unPair request
+  /// might throw errors if device is not paired
   static Future<void> unPair(String deviceId) async {
-    await _platform.unPair(deviceId);
+    return await _executeCommand(
+      () => _platform.unPair(deviceId),
+      timeout: timeout,
+    );
   }
 
   /// To get connected devices to the system ( connected by any app )
@@ -128,13 +186,19 @@ class UniversalBle {
   static Future<List<BleScanResult>> getConnectedDevices({
     List<String>? withServices,
   }) async {
-    return await _platform.getConnectedDevices(withServices);
+    return await _executeCommand(
+      () => _platform.getConnectedDevices(withServices),
+      timeout: timeout,
+    );
   }
 
   /// Enabling Bluetooth, might throw errors if Bluetooth is not available
   /// Not supported on `Web` and `Apple`
   static Future<bool> enableBluetooth() async {
-    return await _platform.enableBluetooth();
+    return await _executeCommand(
+      () => _platform.enableBluetooth(),
+      timeout: timeout,
+    );
   }
 
   /// To get Bluetooth state availability
@@ -165,21 +229,17 @@ class UniversalBle {
 
   static UniversalBlePlatform _defaultPlatform() {
     if (kIsWeb) return UniversalBleWeb.instance;
-    if (Platform.isLinux) return UniversalBleLinux.instance;
+    if (defaultTargetPlatform == TargetPlatform.linux) {
+      return UniversalBleLinux.instance;
+    }
     return UniversalBlePigeonChannel.instance;
   }
+
+  static Future<T> _executeCommand<T>(
+    Future<T> Function() command, {
+    required Duration? timeout,
+  }) {
+    return _queue?.add(command, timeout: timeout) ??
+        (timeout != null ? command().timeout(timeout) : command());
+  }
 }
-
-// Callback types
-typedef OnConnectionChanged = void Function(
-    String deviceId, BleConnectionState state);
-
-typedef OnValueChanged = void Function(
-    String deviceId, String characteristicId, Uint8List value);
-
-typedef OnScanResult = void Function(BleScanResult scanResult);
-
-typedef OnAvailabilityChange = void Function(AvailabilityState state);
-
-typedef OnPairingStateChange = void Function(
-    String deviceId, bool isPaired, String? error);

lib/src/universal_ble_platform_interface.dart

@@ -56,3 +56,17 @@ abstract class UniversalBlePlatform {
     log(message, name: 'UniversalBle');
   }
 }
+
+// Callback types
+typedef OnConnectionChanged = void Function(
+    String deviceId, BleConnectionState state);
+
+typedef OnValueChanged = void Function(
+    String deviceId, String characteristicId, Uint8List value);
+
+typedef OnScanResult = void Function(BleScanResult scanResult);
+
+typedef OnAvailabilityChange = void Function(AvailabilityState state);
+
+typedef OnPairingStateChange = void Function(
+    String deviceId, bool isPaired, String? error);

lib/src/universal_ble_web/queue.dart

@@ -1,139 +0,0 @@
-import 'dart:async';
-
-/// Original Author: Ryan Knell (https://github.com/rknell/dart_queue/commits/master/lib/src/dart_queue_base.dart)
-
-/// Queue to execute Futures in order.
-/// It awaits each future before executing the next one.
-class Queue {
-  final List<_QueuedFuture> _nextCycle = [];
-  final Duration? delay;
-  final Duration? timeout;
-  int parallel;
-  int _lastProcessId = 0;
-  bool _isCancelled = false;
-  bool get isCancelled => _isCancelled;
-  StreamController<int>? _remainingItemsController;
-
-  Stream<int>? get remainingItems {
-    _remainingItemsController ??= StreamController<int>();
-    return _remainingItemsController?.stream.asBroadcastStream();
-  }
-
-  final List<Completer<void>> _completeListeners = [];
-  Future get onComplete {
-    final completer = Completer();
-    _completeListeners.add(completer);
-    return completer.future;
-  }
-
-  Set<int> activeItems = {};
-
-  void cancel() {
-    for (final item in _nextCycle) {
-      item.completer.completeError(QueueCancelledException());
-    }
-    _nextCycle.removeWhere((item) => item.completer.isCompleted);
-    _isCancelled = true;
-  }
-
-  void dispose() {
-    _remainingItemsController?.close();
-    cancel();
-  }
-
-  Queue({this.delay, this.parallel = 1, this.timeout});
-
-  Future<T> add<T>(Future<T> Function() closure) {
-    if (isCancelled) throw QueueCancelledException();
-    final completer = Completer<T>();
-    _nextCycle.add(_QueuedFuture<T>(closure, completer, timeout));
-    _updateRemainingItems();
-    unawaited(_process());
-    return completer.future;
-  }
-
-  Future<void> _process() async {
-    if (activeItems.length < parallel) {
-      _queueUpNext();
-    }
-  }
-
-  void _updateRemainingItems() {
-    final remainingItemsController = _remainingItemsController;
-    if (remainingItemsController != null &&
-        remainingItemsController.isClosed == false) {
-      remainingItemsController.sink.add(_nextCycle.length + activeItems.length);
-    }
-  }
-
-  void _queueUpNext() {
-    if (_nextCycle.isNotEmpty &&
-        !isCancelled &&
-        activeItems.length <= parallel) {
-      final processId = _lastProcessId;
-      activeItems.add(processId);
-      final item = _nextCycle.first;
-      _lastProcessId++;
-      _nextCycle.remove(item);
-      item.onComplete = () async {
-        activeItems.remove(processId);
-        var completionDelay = delay;
-        if (completionDelay != null) {
-          await Future.delayed(completionDelay);
-        }
-        _updateRemainingItems();
-        _queueUpNext();
-      };
-      unawaited(item.execute());
-    } else if (activeItems.isEmpty && _nextCycle.isEmpty) {
-      for (final completer in _completeListeners) {
-        if (completer.isCompleted != true) {
-          completer.complete();
-        }
-      }
-      _completeListeners.clear();
-    }
-  }
-}
-
-class QueueCancelledException implements Exception {}
-
-class _QueuedFuture<T> {
-  final Completer completer;
-  final Future<T> Function() closure;
-  final Duration? timeout;
-  Function? onComplete;
-
-  _QueuedFuture(this.closure, this.completer, this.timeout, {this.onComplete});
-
-  bool _timedOut = false;
-
-  Future<void> execute() async {
-    try {
-      T result;
-      Timer? timeoutTimer;
-
-      var executionTimeout = timeout;
-      if (executionTimeout != null) {
-        timeoutTimer = Timer(executionTimeout, () {
-          _timedOut = true;
-          if (onComplete != null) {
-            onComplete?.call();
-          }
-        });
-      }
-      result = await closure();
-      if (result != null) {
-        completer.complete(result);
-      } else {
-        completer.complete(null);
-      }
-      timeoutTimer?.cancel();
-      await Future.microtask(() {});
-    } catch (e) {
-      completer.completeError(e);
-    } finally {
-      if (onComplete != null && !_timedOut) onComplete?.call();
-    }
-  }
-}

lib/src/universal_ble_web/universal_ble_web.dart

@@ -5,7 +5,6 @@ import 'dart:typed_data';
 import 'package:flutter_web_bluetooth/flutter_web_bluetooth.dart';
 import 'package:universal_ble/src/models/model_exports.dart';
 import 'package:universal_ble/src/universal_ble_platform_interface.dart';
-import 'package:universal_ble/src/universal_ble_web/queue.dart';
 
 class UniversalBleWeb extends UniversalBlePlatform {
   static UniversalBleWeb? _instance;
@@ -15,7 +14,6 @@ class UniversalBleWeb extends UniversalBlePlatform {
     _setupListeners();
   }
 
-  late final _queue = Queue();
   final Map<String, BluetoothDevice> _bluetoothDeviceList = {};
   final Map<String, StreamSubscription> _deviceAdvertisementStreamList = {};
   final Map<String, StreamSubscription> _connectedDeviceStreamList = {};
@@ -222,27 +220,24 @@ class UniversalBleWeb extends UniversalBlePlatform {
     Uint8List value,
     BleOutputProperty bleOutputProperty,
   ) async {
-    await _queue.add(() async {
-      final bleCharacteristic = await _getBleCharacteristic(
-        deviceId: deviceId,
-        serviceId: service,
-        characteristicId: characteristic,
+    final bleCharacteristic = await _getBleCharacteristic(
+      deviceId: deviceId,
+      serviceId: service,
+      characteristicId: characteristic,
+    );
+
+    if (bleCharacteristic == null) {
+      throw Exception(
+        'Characteristic $characteristic for service $service not found',
       );
+    }
 
-      if (bleCharacteristic == null) {
-        throw Exception(
-          'Characteristic $characteristic for service $service not found',
-        );
-      }
-
-      if (bleOutputProperty == BleOutputProperty.withResponse) {
-        await bleCharacteristic
-            .writeValueWithResponse(Uint8List.fromList(value));
-      } else {
-        await bleCharacteristic
-            .writeValueWithoutResponse(Uint8List.fromList(value));
-      }
-    });
+    if (bleOutputProperty == BleOutputProperty.withResponse) {
+      await bleCharacteristic.writeValueWithResponse(Uint8List.fromList(value));
+    } else {
+      await bleCharacteristic
+          .writeValueWithoutResponse(Uint8List.fromList(value));
+    }
   }
 
   @override
@@ -251,19 +246,17 @@ class UniversalBleWeb extends UniversalBlePlatform {
     String service,
     String characteristic,
   ) async {
-    return _queue.add<Uint8List>(() async {
-      var bleCharacteristic = await _getBleCharacteristic(
-        deviceId: deviceId,
-        serviceId: service,
-        characteristicId: characteristic,
-      );
-      if (bleCharacteristic == null) {
-        throw Exception(
-            'Characteristic $characteristic for service $service not found');
-      }
-      var data = await bleCharacteristic.readValue();
-      return data.buffer.asUint8List();
-    });
+    var bleCharacteristic = await _getBleCharacteristic(
+      deviceId: deviceId,
+      serviceId: service,
+      characteristicId: characteristic,
+    );
+    if (bleCharacteristic == null) {
+      throw Exception(
+          'Characteristic $characteristic for service $service not found');
+    }
+    var data = await bleCharacteristic.readValue();
+    return data.buffer.asUint8List();
   }
 
   /// `Unimplemented`

lib/universal_ble.dart

@@ -1,7 +1,6 @@
 library universal_ble;
 
+export 'package:universal_ble/src/ble_command_queue.dart';
 export 'package:universal_ble/src/universal_ble_platform_interface.dart';
 export 'package:universal_ble/src/universal_ble.dart';
 export 'package:universal_ble/src/models/model_exports.dart';
-
-