snapshots method

Implementation

Stream<QuerySnapshot<T>> snapshots({
  bool includeMetadataChanges = false,
  ListenSource source = ListenSource.defaultSource,
}) {
  this._rt = 1;
  late StreamController<QuerySnapshot<T>> controller;
  Timer? _pollingTimer;

  var auth = FusabaseAuth.instanceFor(app: this._oracledb.app);
  var email = auth.currentUser?.email ?? "";
  var userLvlPayload = {
    "path": this._path,
    "conditions": this._conditions,
    "explicitOrder": this._explicitOrder,
    "access_token": email,
  };

  // this query id is used to store in the cache (user level)
  int queryIdInt = createHashFromJson(userLvlPayload);
  String queryId = queryIdInt.toString();

  var socketPayload = {
    "path": this._path,
    "conditions": this._conditions,
    "explicitOrder": this._explicitOrder,
  };

  // this query id is used to register listener (common for each query)
  int queryIdSocketInt = createHashFromJson(socketPayload);
  String queryIdSocket = queryIdSocketInt.toString();

  // this query id is used to store the notifiers (unique each time)
  String localQueryId = generateRandomHash();

  Future<QuerySnapshot<T>> firstPoll() async {
    dynamic oldData = await this._oracledb._snapStore.getData(
      queryId,
      oracledb,
    );

    QuerySnapshot newSnap = await this.get();

    if (oldData == null) {
      //first snapshot
      await this._oracledb._snapStore.insertData({
        "queryId": queryId,
        "snap": newSnap,
        "type": "collection",
        "path": this.path,
      });
    } else {
      await this._oracledb._snapStore.updateData({
        "queryId": queryId,
        "snap": newSnap,
        "type": "collection",
        "path": this.path,
      });
    }
    return newSnap as QuerySnapshot<T>;
  }

  bool uselongPolling =
      _oracledb.settings.webExperimentalForceLongPolling ?? true;
  if (_oracledb.settings.webExperimentalForceLongPolling ?? true) {
    uselongPolling = true;
  }

  //uselongPolling = false;

  if (uselongPolling) {
    controller = StreamController<QuerySnapshot<T>>.broadcast(
      onListen: () async {
        controller.add(await firstPoll());
      },
      onCancel: () {
        _pollingTimer?.cancel();
      },
    );

    final pollingInterval = _safeLongPollingInterval(this._oracledb.settings);

    Future<void> longPolling() async {
      dynamic oldData = await this._oracledb._snapStore.getData(
        queryId,
        oracledb,
      );
      QuerySnapshot newSnap = await this.get();
      if (oldData == null) {
        //first snapshot
        await this._oracledb._snapStore.insertData({
          "queryId": queryId,
          "snap": newSnap,
          "type": "collection",
          "path": this.path,
        });
        controller.add(newSnap as QuerySnapshot<T>);
        return;
      }

      //create version maps for old and new snapshot docs
      var oldVersionMap = {};
      var newVersionMap = {};

      //index map for old documents
      var oldIndexMap = {};

      for (var i = 0; i < newSnap._docs.length; i++) {
        newVersionMap[newSnap._docs[i].id] =
            newSnap._docs[i]._otherMetadata["ASOF"] != null
                ? BigInt.parse(newSnap._docs[i]._otherMetadata["ASOF"])
                : newSnap._docs[i]._otherMetadata["VERSION"];
      }
      for (var i = 0; i < oldData["snap"]._docs.length; i++) {
        oldVersionMap[oldData["snap"]._docs[i].id] =
            oldData["snap"]._docs[i]._otherMetadata["ASOF"] != null
                ? BigInt.parse(
                  oldData["snap"]._docs[i]._otherMetadata["ASOF"],
                )
                : oldData["snap"]._docs[i]._otherMetadata["VERSION"];
        oldIndexMap[oldData["snap"]._docs[i].id] = i;
      }

      //create doc changes
      List<DocumentChange> docsChanged = [];
      for (var i = 0; i < newSnap._docs.length; i++) {
        if (oldVersionMap[newSnap._docs[i].id] != null &&
            newVersionMap[newSnap._docs[i].id] >
                oldVersionMap[newSnap._docs[i].id]) {
          docsChanged.add(
            DocumentChange(
              doc: DocumentSnapshot(
                {
                  "DOCUMENT": newSnap._docs[i]._data,
                  "OID": newSnap._docs[i].id,
                  "VERSION": newSnap._docs[i].version,
                },
                newSnap._docs[i]._reference,
                newSnap._docs[i]._metadata,
                newSnap._docs[i].rowID,
              ),
              type: DocumentChangeType.modified,
              oldIndex: oldIndexMap[newSnap._docs[i].id],
              newIndex: i,
            ),
          );
        } else if (oldVersionMap[newSnap._docs[i].id] == null) {
          docsChanged.add(
            DocumentChange(
              doc: DocumentSnapshot(
                {
                  "DOCUMENT": newSnap._docs[i]._data,
                  "OID": newSnap._docs[i].id,
                  "VERSION": newSnap._docs[i].version,
                },
                newSnap._docs[i]._reference,
                newSnap._docs[i]._metadata,
                newSnap._docs[i].rowID,
              ),
              type: DocumentChangeType.added,
              oldIndex: -1,
              newIndex: i,
            ),
          );
        }
      }
      for (var i = 0; i < oldData["snap"]._docs.length; i++) {
        if (newVersionMap[oldData["snap"]._docs[i].id] == null) {
          docsChanged.add(
            DocumentChange(
              doc: DocumentSnapshot(
                {
                  "DOCUMENT": oldData["snap"]._docs[i]._data,
                  "OID": oldData["snap"]._docs[i].id,
                  "VERSION": oldData["snap"]._docs[i].version,
                },
                oldData["snap"]._docs[i].reference,
                oldData["snap"]._docs[i].metadata,
                oldData["snap"]._docs[i].rowID,
              ),
              type: DocumentChangeType.removed,
              oldIndex: i,
              newIndex: -1,
            ),
          );
        }
      }

      newSnap._docChanges = docsChanged;

      await this._oracledb._snapStore.updateData({
        "queryId": queryId,
        "snap": newSnap,
        "type": "collection",
        "path": this.path,
      });

      if (docsChanged.isNotEmpty) {
        controller.add(newSnap as QuerySnapshot<T>);
      }
      return;
    }

    _pollingTimer = Timer(pollingInterval, () {
      longPolling();

      _pollingTimer = Timer.periodic(pollingInterval, (Timer timer) {
        longPolling();
      });
    });

    return controller.stream;
  } else {
    bool isCancelled = false;
    bool hasRegisteredRealtimeListener = false;
    dynamic payload = {
      "queryId": queryIdSocketInt,
      "status": 1,
      "payload": socketPayload,
    };

    void addToController(dynamic snap) {
      if (!controller.isClosed) {
        controller.add(snap as QuerySnapshot<T>);
      }
    }

    controller = StreamController<QuerySnapshot<T>>.broadcast(
      onListen: () async {
        isCancelled = false;
        await _oracledb.preSnapshot();
        if (isCancelled) {
          return;
        }
        this._oracledb._queryIdMap[queryIdSocket] = queryId;
        List<String> localQueryIds = this._oracledb._localQueryIdMap
            .putIfAbsent(queryId, () => []);
        if (!localQueryIds.contains(localQueryId)) {
          localQueryIds.add(localQueryId);
        }
        this._oracledb._snapshotControllerMap[localQueryId] = addToController;
        hasRegisteredRealtimeListener = true;
        this._oracledb.sendMessageSocket(payload);
        dynamic fsnap = await firstPoll();
        if (!isCancelled && !controller.isClosed) {
          controller.add(fsnap);
        }
      },
      onCancel: () {
        isCancelled = true;
        if (!hasRegisteredRealtimeListener) {
          return;
        }
        hasRegisteredRealtimeListener = false;
        if (this._oracledb._snapshotControllerMap.containsKey(localQueryId)) {
          this._oracledb._snapshotControllerMap.remove(localQueryId);
        }
        if (this._oracledb._localQueryIdMap.containsKey(queryId) &&
            this._oracledb._localQueryIdMap[queryId]!.contains(
              localQueryId,
            )) {
          this._oracledb._localQueryIdMap[queryId]?.remove(localQueryId);
        }
        if (!this._oracledb._localQueryIdMap.containsKey(queryId) ||
            this._oracledb._localQueryIdMap[queryId]!.isEmpty) {
          this._oracledb._localQueryIdMap.remove(queryId);
          this._oracledb._queryIdMap.remove(queryIdSocket);
          dynamic unsubPayload = {
            "queryId": queryIdSocketInt,
            "status": 0,
            "payload": socketPayload,
          };
          this._oracledb.sendMessageSocket(unsubPayload);
        }
      },
    );

    return controller.stream;
  }
}