fix(llc): Fix attachment upload state uneven progress. (#315)

* feat(llc): Add rate limiter functions. fix(llc): Fix attachment upload state uneven progress. Signed-off-by: Sahil Kumar <[email protected]> * feat(llc): Export `async.dart` and `rate_limit.dart` Signed-off-by: Sahil Kumar <[email protected]> * refactor(llc): Replace our `rate_limit` implementation with the dart port of js `lodash` to better handlw `cancel` and `flush` functions. Signed-off-by: Sahil Kumar <[email protected]> * fix(UI-Kit): Make android example run on real devices Signed-off-by: Sahil Kumar <[email protected]> * fix offline behaviour Co-authored-by: Salvatore Giordano <[email protected]>
2021-03-08 12:35:43 +01:00
parent 5933929399
commit e60294dc92
11 changed files with 413 additions and 81 deletions
@@ -6,12 +6,12 @@ import 'package:dio/dio.dart';
 import 'package:logging/logging.dart';
 import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/src/api/retry_queue.dart';
-import 'package:stream_chat/src/debounce.dart';
 import 'package:stream_chat/src/event_type.dart';
 import 'package:stream_chat/src/models/attachment_file.dart';
 import 'package:stream_chat/src/models/channel_state.dart';
 import 'package:stream_chat/src/models/user.dart';
 import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat/src/extensions/rate_limit.dart';

 /// This a the class that manages a specific channel.
 class Channel {
@@ -218,28 +218,29 @@ class Channel {
    }

    client.logger.info('Found ${attachments.length} attachments');
+
+    void updateAttachment(Attachment attachment) {
+      final index =
+          message.attachments.indexWhere((it) => it.id == attachment.id);
+      if (index != -1) {
+        message.attachments[index] = attachment;
+        state?.addMessage(message);
+      }
+    }
+
    return Future.wait(attachments.map((it) {
      client.logger.info('Uploading ${it.id} attachment...');

-      void updateAttachment(Attachment attachment) {
-        final index =
-            message.attachments.indexWhere((it) => it.id == attachment.id);
-        if (index != -1) {
-          message.attachments[index] = attachment;
-          state?.addMessage(message);
-        }
-      }
+      final throttledUpdateAttachment = updateAttachment.throttled(
+        const Duration(milliseconds: 500),
+      );

      void onSendProgress(int sent, int total) {
-        debounce(
-          timeout: const Duration(seconds: 1),
-          target: updateAttachment,
-          positionalArguments: [
-            it.copyWith(
-              uploadState: UploadState.inProgress(uploaded: sent, total: total),
-            ),
-          ],
-        );
+        throttledUpdateAttachment([
+          it.copyWith(
+            uploadState: UploadState.inProgress(uploaded: sent, total: total),
+          ),
+        ]);
      }

      final isImage = it.type == 'image';
@@ -282,6 +283,7 @@ class Channel {
          it.copyWith(uploadState: UploadState.failed(error: e.toString())),
        );
      }).whenComplete(() {
+        throttledUpdateAttachment?.cancel();
        _cancelableAttachmentUploadRequest.remove(it.id);
      });
    })).whenComplete(() {
@@ -1202,7 +1204,12 @@ class Channel {
 /// The class that handles the state of the channel listening to the events
 class ChannelClientState {
  /// Creates a new instance listening to events and updating the state
-  ChannelClientState(this._channel, ChannelState channelState) {
+  ChannelClientState(
+    this._channel,
+    ChannelState channelState,
+  ) : _debouncedUpdatePersistenceChannelState = _channel
+            ?._client?.chatPersistenceClient?.updateChannelState
+            ?.debounced(const Duration(seconds: 1)) {
    retryQueue = RetryQueue(
      channel: _channel,
      logger: Logger('RETRY QUEUE ${_channel.cid}'),
@@ -1697,15 +1704,11 @@ class ChannelClientState {
  ChannelState get channelState => _channelStateController.value;
  BehaviorSubject<ChannelState> _channelStateController;

+  final Debounce _debouncedUpdatePersistenceChannelState;
+
  set _channelState(ChannelState v) {
    _channelStateController.add(v);
-    if (_channel._client.persistenceEnabled) {
-      debounce(
-        timeout: const Duration(milliseconds: 500),
-        target: _channel._client.chatPersistenceClient?.updateChannelState,
-        positionalArguments: [v],
-      );
-    }
+    _debouncedUpdatePersistenceChannelState?.call([v]);
  }

  /// The channel threads related to this channel
@@ -1850,6 +1853,7 @@ class ChannelClientState {

  /// Call this method to dispose this object
  void dispose() {
+    _debouncedUpdatePersistenceChannelState?.cancel();
    _unreadCountController.close();
    retryQueue.dispose();
    _subscriptions.forEach((s) => s.cancel());
@@ -602,39 +602,39 @@ class StreamChatClient {
    Map<String, dynamic> options,
    PaginationParams paginationParams = const PaginationParams(),
    int messageLimit,
-    bool preferOffline = false,
    bool waitForConnect = true,
  }) async* {
    final hash = base64.encode(utf8.encode(
      '$filter${_asMap(sort)}$options${paginationParams?.toJson()}'
-      '$messageLimit$preferOffline',
+      '$messageLimit',
    ));

    if (_queryChannelsStreams.containsKey(hash)) {
      yield await _queryChannelsStreams[hash];
    } else {
-      if (preferOffline) {
-        final channels = await queryChannelsOffline(
-          filter: filter,
-          sort: sort,
-          paginationParams: paginationParams,
-        );
-        if (channels.isNotEmpty) yield channels;
-      }
-
-      final newQueryChannelsFuture = queryChannelsOnline(
+      final channels = await queryChannelsOffline(
        filter: filter,
        sort: sort,
-        options: options,
        paginationParams: paginationParams,
-        messageLimit: messageLimit,
-      ).whenComplete(() {
-        _queryChannelsStreams.remove(hash);
-      });
+      );
+      if (channels.isNotEmpty) yield channels;

-      _queryChannelsStreams[hash] = newQueryChannelsFuture;
+      if (wsConnectionStatus == ConnectionStatus.connected) {
+        final newQueryChannelsFuture = queryChannelsOnline(
+          filter: filter,
+          sort: sort,
+          options: options,
+          paginationParams: paginationParams,
+          messageLimit: messageLimit,
+          waitForConnect: waitForConnect,
+        ).whenComplete(() {
+          _queryChannelsStreams.remove(hash);
+        });

-      yield await newQueryChannelsFuture;
+        _queryChannelsStreams[hash] = newQueryChannelsFuture;
+
+        yield await newQueryChannelsFuture;
+      }
    }
  }

@@ -1,32 +0,0 @@
-import 'dart:async';
-
-import 'package:meta/meta.dart';
-
-/// Map of timeouts being debounced
-Map<Function, Timer> timeouts = {};
-
-/// Runs a function avoiding calling it too many times in a [timeoutMS] window
-void debounce({
-  @required Duration timeout,
-  @required Function target,
-  List positionalArguments,
-  Map<Symbol, dynamic> namedArguments,
-}) {
-  if (target == null) {
-    return;
-  }
-
-  if (timeouts.containsKey(target)) {
-    timeouts[target].cancel();
-  }
-
-  final timer = Timer(timeout, () {
-    Function.apply(
-      target,
-      positionalArguments,
-      namedArguments,
-    );
-  });
-
-  timeouts[target] = timer;
-}
@@ -0,0 +1,335 @@
+// ignore_for_file: lines_longer_than_80_chars
+
+import 'dart:async' show Timer;
+import 'dart:math' as math;
+
+/// Useful rate limiter extensions for [Function] class.
+extension RateLimit on Function {
+  /// Converts this into a [Debounce] function.
+  Debounce debounced(
+    Duration wait, {
+    bool leading = false,
+    bool trailing = true,
+    Duration maxWait,
+  }) =>
+      Debounce(
+        this,
+        wait,
+        leading: leading,
+        trailing: trailing,
+        maxWait: maxWait,
+      );
+
+  /// Converts this into a [Throttle] function.
+  Throttle throttled(
+    Duration wait, {
+    bool leading = true,
+    bool trailing = true,
+  }) =>
+      Throttle(
+        this,
+        wait,
+        leading: leading,
+        trailing: trailing,
+      );
+}
+
+/// TopLevel lambda to create [Debounce] functions.
+Debounce debounce(
+  Function func,
+  Duration wait, {
+  bool leading = false,
+  bool trailing = true,
+  Duration maxWait,
+}) =>
+    Debounce(
+      func,
+      wait,
+      leading: leading,
+      trailing: trailing,
+      maxWait: maxWait,
+    );
+
+/// TopLevel lambda to create [Throttle] functions.
+Throttle throttle(
+  Function func,
+  Duration wait, {
+  bool leading = true,
+  bool trailing = true,
+}) =>
+    Throttle(
+      func,
+      wait,
+      leading: leading,
+      trailing: trailing,
+    );
+
+/// Creates a debounced function that delays invoking `func` until after `wait`
+/// milliseconds have elapsed since the last time the debounced function was
+/// invoked. The debounced function comes with a [Debounce.cancel] method to cancel
+/// delayed `func` invocations and a [Debounce.flush] method to immediately invoke them.
+/// Provide `leading` and/or `trailing` to indicate whether `func` should be
+/// invoked on the `leading` and/or `trailing` edge of the `wait` interval.
+/// The `func` is invoked with the last arguments provided to the [call]
+/// function. Subsequent calls to the debounced function return the result of
+/// the last `func` invocation.
+///
+/// **Note:** If `leading` and `trailing` options are `true`, `func` is
+/// invoked on the trailing edge of the timeout only if the debounced function
+/// is invoked more than once during the `wait` timeout.
+///
+/// If `wait` is [Duration.zero] and `leading` is `false`,
+/// `func` invocation is deferred until the next tick.
+///
+/// See [David Corbacho's article](https://css-tricks.com/debouncing-throttling-explained-examples/)
+/// for details over the differences between [Debounce] and [Throttle].
+///
+/// Some examples:
+///
+/// Avoid calling costly network calls when user is typing something.
+/// ```dart
+///   void fetchData(String query) async {
+///     final data = api.getData(query);
+///     doSomethingWithTheData(data);
+///   }
+///
+///   final debouncedFetchData = Debounce(
+///     fetchData,
+///     const Duration(milliseconds: 350),
+///   );
+///
+///   void onSearchQueryChanged(query) {
+///      debouncedFetchData(query);
+///   }
+/// ```
+///
+/// Cancel the trailing debounced invocation.
+/// ```dart
+///   void dispose() {
+///     debounced.cancel();
+///   }
+/// ```
+///
+/// Check for pending invocations.
+/// ```dart
+///   final status = debounced.isPending ? "Pending..." : "Ready";
+/// ```
+class Debounce {
+  /// Creates a new instance of [Debounce].
+  Debounce(
+    this._func,
+    Duration wait, {
+    bool leading = false,
+    bool trailing = true,
+    Duration maxWait,
+  })  : _leading = leading,
+        _trailing = trailing,
+        _wait = wait?.inMilliseconds ?? 0,
+        _maxing = maxWait != null {
+    if (_maxing) {
+      _maxWait = math.max(maxWait.inMilliseconds, _wait);
+    }
+  }
+
+  final Function _func;
+  final bool _leading;
+  final bool _trailing;
+  final int _wait;
+  final bool _maxing;
+
+  int _maxWait;
+  List<Object> _lastArgs;
+  Map<Symbol, Object> _lastNamedArgs;
+  Timer _timer;
+  int _lastCallTime;
+  Object _result;
+  int _lastInvokeTime = 0;
+
+  Object _invokeFunc(int time) {
+    final args = _lastArgs;
+    final namedArgs = _lastNamedArgs;
+    _lastArgs = _lastNamedArgs = null;
+    _lastInvokeTime = time;
+    return _result = Function.apply(_func, args, namedArgs);
+  }
+
+  Timer _startTimer(Function pendingFunc, int wait) =>
+      Timer(Duration(milliseconds: wait), pendingFunc);
+
+  bool _shouldInvoke(int time) {
+    final timeSinceLastCall = time - (_lastCallTime ?? double.nan);
+    final timeSinceLastInvoke = time - _lastInvokeTime;
+
+    // Either this is the first call, activity has stopped and we're at the
+    // trailing edge, the system time has gone backwards and we're treating
+    // it as the trailing edge, or we've hit the `maxWait` limit.
+    return _lastCallTime == null ||
+        (timeSinceLastCall >= _wait) ||
+        (timeSinceLastCall < 0) ||
+        (_maxing && timeSinceLastInvoke >= _maxWait);
+  }
+
+  Object _trailingEdge(int time) {
+    _timer = null;
+
+    // Only invoke if we have `lastArgs` which means `func` has been
+    // debounced at least once.
+    if (_trailing && _lastArgs != null) {
+      return _invokeFunc(time);
+    }
+    _lastArgs = _lastNamedArgs = null;
+    return _result;
+  }
+
+  int _remainingWait(int time) {
+    final timeSinceLastCall = time - _lastCallTime;
+    final timeSinceLastInvoke = time - _lastInvokeTime;
+    final timeWaiting = _wait - timeSinceLastCall;
+
+    return _maxing
+        ? math.min(timeWaiting, _maxWait - timeSinceLastInvoke)
+        : timeWaiting;
+  }
+
+  void _timerExpired() {
+    final time = DateTime.now().millisecondsSinceEpoch;
+    if (_shouldInvoke(time)) {
+      _trailingEdge(time);
+    } else {
+      // Restart the timer.
+      _timer = _startTimer(_timerExpired, _remainingWait(time));
+    }
+  }
+
+  Object _leadingEdge(int time) {
+    // Reset any `maxWait` timer.
+    _lastInvokeTime = time;
+    // Start the timer for the trailing edge.
+    _timer = _startTimer(_timerExpired, _wait);
+    // Invoke the leading edge.
+    return _leading ? _invokeFunc(time) : _result;
+  }
+
+  /// Cancels all the remaining delayed functions.
+  void cancel() {
+    _timer?.cancel();
+    _lastInvokeTime = 0;
+    _lastArgs = _lastNamedArgs = _lastCallTime = _timer = null;
+  }
+
+  /// Immediately invokes all the remaining delayed functions.
+  Object flush() {
+    final now = DateTime.now().millisecondsSinceEpoch;
+    return _timer == null ? _result : _trailingEdge(now);
+  }
+
+  /// True if there are functions remaining to get invoked.
+  bool get isPending => _timer != null;
+
+  /// Calls/invokes this class like a function.
+  /// Pass [args] and [namedArgs] to be used while invoking [_func].
+  Object call(
+    List<dynamic> args, {
+    Map<Symbol, dynamic> namedArgs,
+  }) {
+    final time = DateTime.now().millisecondsSinceEpoch;
+    final isInvoking = _shouldInvoke(time);
+
+    _lastArgs = args;
+    _lastNamedArgs = namedArgs;
+    _lastCallTime = time;
+
+    if (isInvoking) {
+      if (_timer == null) {
+        return _leadingEdge(_lastCallTime);
+      }
+      if (_maxing) {
+        // Handle invocations in a tight loop.
+        _timer = _startTimer(_timerExpired, _wait);
+        return _invokeFunc(_lastCallTime);
+      }
+    }
+    _timer ??= _startTimer(_timerExpired, _wait);
+    return _result;
+  }
+}
+
+/// Creates a throttled function that only invokes `func` at most once per
+/// every `wait` milliseconds. The throttled function comes with a [Throttle.cancel]
+/// method to cancel delayed `func` invocations and a [Throttle.flush] method to
+/// immediately invoke them. Provide `leading` and/or `trailing` to indicate
+/// whether `func` should be invoked on the `leading` and/or `trailing` edge of the `wait` timeout.
+/// The `func` is invoked with the last arguments provided to the
+/// throttled function. Subsequent calls to the throttled function return the
+/// result of the last `func` invocation.
+///
+/// **Note:** If `leading` and `trailing` options are `true`, `func` is
+/// invoked on the trailing edge of the timeout only if the throttled function
+/// is invoked more than once during the `wait` timeout.
+///
+/// If `wait` is [Duration.zero] and `leading` is `false`, `func` invocation is deferred
+/// until the next tick.
+///
+/// See [David Corbacho's article](https://css-tricks.com/debouncing-throttling-explained-examples/)
+/// for details over the differences between [Throttle] and [Debounce].
+///
+/// Some examples:
+///
+/// Avoid excessively rebuilding UI progress while uploading data to server.
+/// ```dart
+///   void updateUI(Data data) {
+///     updateProgress(data);
+///   }
+///
+///   final throttledUpdateUI = Throttle(
+///     updateUI,
+///     const Duration(milliseconds: 350),
+///   );
+///
+///   void onUploadProgressChanged(progress) {
+///      throttledUpdateUI(progress);
+///   }
+/// ```
+///
+/// Cancel the trailing throttled invocation.
+/// ```dart
+///   void dispose() {
+///     throttled.cancel();
+///   }
+/// ```
+///
+/// Check for pending invocations.
+/// ```dart
+///   final status = throttled.isPending ? "Pending..." : "Ready";
+/// ```
+class Throttle {
+  /// Creates a new instance of [Throttle]
+  Throttle(
+    Function func,
+    Duration wait, {
+    bool leading = true,
+    bool trailing = true,
+  }) : _debounce = Debounce(
+          func,
+          wait,
+          leading: leading,
+          trailing: trailing,
+          maxWait: wait,
+        );
+
+  final Debounce _debounce;
+
+  /// Cancels all the remaining delayed functions.
+  void cancel() => _debounce.cancel();
+
+  /// Immediately invokes all the remaining delayed functions.
+  Object flush() => _debounce.flush();
+
+  /// True if there are functions remaining to get invoked.
+  bool get isPending => _debounce.isPending;
+
+  /// Calls/invokes this class like a function.
+  /// Pass [args] and [namedArgs] to be used while invoking `func`.
+  Object call(List<dynamic> args, {Map<Symbol, dynamic> namedArgs}) =>
+      _debounce.call(args, namedArgs: namedArgs);
+}
@@ -1,5 +1,6 @@
 library stream_chat;

+export 'package:async/async.dart';
 export 'package:dio/src/dio_error.dart';
 export 'package:dio/src/multipart_file.dart';
 export 'package:dio/src/options.dart' show ProgressCallback;
@@ -14,6 +15,7 @@ export './src/attachment_file_uploader.dart' show AttachmentFileUploader;
 export './src/client.dart';
 export './src/db/chat_persistence_client.dart';
 export './src/event_type.dart';
+export './src/extensions/rate_limit.dart';
 export './src/extensions/string_extension.dart';
 export './src/models/action.dart';
 export './src/models/attachment.dart';
@@ -34,6 +34,7 @@ android {

    lintOptions {
        disable 'InvalidPackage'
+        checkReleaseBuilds false
    }

    defaultConfig {
@@ -7,7 +7,7 @@
         FlutterApplication and put your custom class here. -->
    <uses-permission android:name="android.permission.INTERNET"/>
    <application
-        android:name="io.flutter.app.FlutterApplication"
+        android:name=".Application"
        android:label="example"
        android:icon="@mipmap/ic_launcher">
        <activity
@@ -0,0 +1,17 @@
+package com.example.example
+
+import io.flutter.app.FlutterApplication
+import io.flutter.plugin.common.PluginRegistry
+import io.flutter.plugin.common.PluginRegistry.PluginRegistrantCallback
+import io.flutter.plugins.pathprovider.PathProviderPlugin
+
+class Application : FlutterApplication(), PluginRegistrantCallback {
+    override fun onCreate() {
+        super.onCreate()
+    }
+
+    override fun registerWith(registry: PluginRegistry?) {
+        PathProviderPlugin.registerWith(registry?.registrarFor(
+                "io.flutter.plugins.pathprovider.PathProviderPlugin"))
+    }
+}
@@ -1,6 +1,12 @@
 package com.example.example

+import androidx.annotation.NonNull;
 import io.flutter.embedding.android.FlutterActivity
+import io.flutter.embedding.engine.FlutterEngine
+import io.flutter.plugins.GeneratedPluginRegistrant

 class MainActivity: FlutterActivity() {
-}
+    override fun configureFlutterEngine(@NonNull flutterEngine: FlutterEngine) {
+        GeneratedPluginRegistrant.registerWith(flutterEngine);
+    }
+}
@@ -6,7 +6,7 @@ buildscript {
    }

    dependencies {
-        classpath 'com.android.tools.build:gradle:4.0.1'
+        classpath 'com.android.tools.build:gradle:3.6.2'
        classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlin_version"
    }
 }
@@ -107,7 +107,6 @@ class ChannelsBlocState extends State<ChannelsBloc>
        sort: sortOptions,
        options: options,
        paginationParams: paginationParams,
-        preferOffline: onlyOffline,
      )) {
        if (clear) {
          _channelsController.add(channels);