diff --git a/packages/stream_chat/lib/src/api/channel.dart b/packages/stream_chat/lib/src/api/channel.dart index c05ba49a..13ea00ac 100644 --- a/packages/stream_chat/lib/src/api/channel.dart +++ b/packages/stream_chat/lib/src/api/channel.dart @@ -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 _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()); diff --git a/packages/stream_chat/lib/src/client.dart b/packages/stream_chat/lib/src/client.dart index f0823d6b..cbb0901d 100644 --- a/packages/stream_chat/lib/src/client.dart +++ b/packages/stream_chat/lib/src/client.dart @@ -602,39 +602,39 @@ class StreamChatClient { Map 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; + } } } diff --git a/packages/stream_chat/lib/src/debounce.dart b/packages/stream_chat/lib/src/debounce.dart deleted file mode 100644 index 140c5ef8..00000000 --- a/packages/stream_chat/lib/src/debounce.dart +++ /dev/null @@ -1,32 +0,0 @@ -import 'dart:async'; - -import 'package:meta/meta.dart'; - -/// Map of timeouts being debounced -Map 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 namedArguments, -}) { - if (target == null) { - return; - } - - if (timeouts.containsKey(target)) { - timeouts[target].cancel(); - } - - final timer = Timer(timeout, () { - Function.apply( - target, - positionalArguments, - namedArguments, - ); - }); - - timeouts[target] = timer; -} diff --git a/packages/stream_chat/lib/src/extensions/rate_limit.dart b/packages/stream_chat/lib/src/extensions/rate_limit.dart new file mode 100644 index 00000000..c9f5934f --- /dev/null +++ b/packages/stream_chat/lib/src/extensions/rate_limit.dart @@ -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 _lastArgs; + Map _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 args, { + Map 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 args, {Map namedArgs}) => + _debounce.call(args, namedArgs: namedArgs); +} diff --git a/packages/stream_chat/lib/stream_chat.dart b/packages/stream_chat/lib/stream_chat.dart index 05e3a4a4..0e1c9ed2 100644 --- a/packages/stream_chat/lib/stream_chat.dart +++ b/packages/stream_chat/lib/stream_chat.dart @@ -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'; diff --git a/packages/stream_chat_flutter/example/android/app/build.gradle b/packages/stream_chat_flutter/example/android/app/build.gradle index 5a4d759a..9bb36eac 100644 --- a/packages/stream_chat_flutter/example/android/app/build.gradle +++ b/packages/stream_chat_flutter/example/android/app/build.gradle @@ -34,6 +34,7 @@ android { lintOptions { disable 'InvalidPackage' + checkReleaseBuilds false } defaultConfig { diff --git a/packages/stream_chat_flutter/example/android/app/src/main/AndroidManifest.xml b/packages/stream_chat_flutter/example/android/app/src/main/AndroidManifest.xml index 3197993b..42c5fa1e 100644 --- a/packages/stream_chat_flutter/example/android/app/src/main/AndroidManifest.xml +++ b/packages/stream_chat_flutter/example/android/app/src/main/AndroidManifest.xml @@ -7,7 +7,7 @@ FlutterApplication and put your custom class here. --> sort: sortOptions, options: options, paginationParams: paginationParams, - preferOffline: onlyOffline, )) { if (clear) { _channelsController.add(channels);