How to throttle Firestore's stream on client side?

user_735332 0
Viewed 100

So today my code went crazy and wrote to Firestore database 1000 times and read back about 50k docs within a span of a few seconds. It was due to a bug in calling write operations which then cause the Firebase snapshots() stream to fire 1000 times with about 50 reads each time. So, then it lead me to chasing some details on how to prevent this from happening in the future.

I looked into RxDart package here around throttling streams, but it is said that using this feature just simply discards the data, which means that the data is still being read. So if a looping write is not caught, it will potentially be disastrous from a cost perspective.

Stream<T> throttle( Stream window( T event ), {bool trailing = false, bool leading = true} )

Emits a value from the source Stream, then ignores subsequent source values while the window Stream is open, then repeats this process.

If leading is true, then the first item in each window is emitted. If trailing is true, then the last item in each window is emitted.

You can use the value of the last throttled event to determine the length of the next window.

I know that I can prevent this using Firestore rules, but honestly, this is not a server problem, it is a client problem, so I'd like to address it client side first. The server solution would be more to prevent misuse rather than catching errors.

rxdart google-cloud-firestore firebase flutter stream
1 Answers

You can use class like this everywhere that you want throttle :

/// Throttling
/// Have method [throttle]
class Throttling extends Stream<bool> implements Sink<Function> {
  Duration _duration;

  /// Get current duration
  Duration get duration => _duration;

  /// Set new duration
  set duration(Duration value) {
    assert(duration is Duration && !duration.isNegative);
    _duration = value;
  }

  bool _isReady = true;

  /// is ready
  bool get isReady => _isReady;
  Future<void> get _waiter => Future.delayed(_duration);
  // ignore: close_sinks
  final StreamController<bool> _stateSC = StreamController<bool>.broadcast();

  /// Throttling
  /// Have method [throttle]
  /// Must be closed with [close] method
  Throttling({Duration duration = const Duration(seconds: 1)})
      : assert(duration is Duration && !duration.isNegative),
        _duration = duration {
    _stateSC.sink.add(true);
  }

  /// limits the maximum number of times a given
  /// event handler can be called over time
  dynamic throttle(Function func) {
    if (!_isReady) return null;
    _stateSC.sink.add(false);
    _isReady = false;
    _waiter.then((_) {
      _isReady = true;
      if (!_stateSC.isClosed) {
        _stateSC.sink.add(true);
      }
    });
    return Function.apply(func, []);
  }

  @override
  StreamSubscription<bool> listen(
      void onData(bool event), {
        Function onError,
        void onDone(),
        bool cancelOnError,
      }) =>
      _stateSC.stream.listen(
        onData,
        onError: onError,
        onDone: onDone,
        cancelOnError: cancelOnError,
      );

  /// Closing instances of Sink prevents
  /// memory leaks and unexpected behavior.
  @Deprecated('Use [close] instead')
  void dispose() => close();

  /// Shortcut for [throttle] method
  @override
  dynamic add(Function data) => throttle(data);

  @override
  Future<void> close() => _stateSC.close();
}

and use class like this:

  final throttle = Throttling(duration: const Duration(milliseconds: 500));
        throttle.throttle(() {
//code you want
        });
user_60844210
Related