initHotList method

void initHotList(Stream<Event> changes)

Must be called in the constructor

Implementation

void initHotList(Stream<Event> changes) {
  _curReplaceOperationId = replaceOperationId;
  hotListEventSubscription = changes
      .bufferCompleter(() async => Future.wait([
            waitAllLoadsToComplete(),
            if (actualizeCompleter?.isCompleted == false)
              actualizeCompleter!.future,
          ]))
      .asyncMap((event) async {
    // At this point, no loading or actualizing operations are performed
    // because of bufferCompleter.
    _curReplaceOperationId = replaceOperationId;
    actualizeCompleter = Completer();

    try {
      return expandHotListEvents(event);
    } catch (e) {
      actualizeCompleter?.complete();
      rethrow;
    }
  }).where((event) {
    final cancelChain =
        _curReplaceOperationId != replaceOperationId || event.isEmpty;
    if (cancelChain) {
      actualizeCompleter?.complete();
    }
    return !cancelChain;
  }).asyncMap((RecordUpdates<DecisionRecord, Key> updates) async {
    final changes = filterHotListRecords(updates);
    try {
      return HotListChanges(
        recordsToInsert:
            Set.of(await convertDecisionRecords(changes.recordsToInsert)),
        recordKeysToRemove: changes.recordKeysToRemove,
      );
    } catch (e) {
      actualizeCompleter?.complete();
      rethrow;
    }
  }).where((HotListChanges<Record, Key> event) {
    final cancelChain =
        _curReplaceOperationId != replaceOperationId || event.isEmpty;
    if (cancelChain) {
      actualizeCompleter?.complete();
    }
    return !cancelChain;
  }).listen((changes) {
    actualizeCompleter?.complete();
    updateHotList(changes);
  });
}