snapshots method

Implementation

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

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

  //uselongPolling = false;

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

    final pollingInterval = _safeLongPollingInterval(_oracledb.settings);

    Future<void> longPolling() async {
      DocumentSnapshot newSnap = await this.get();
      var newVer = null;
      if (newSnap._otherMetadata["ASOF"] != null) {
        newVer = BigInt.from(newSnap._otherMetadata["ASOF"]);
      } else {
        newVer = newSnap._otherMetadata["VERSION"];
      }
      if (lastDocUpdate == null ||
          (newVer != null && lastDocUpdate < newVer)) {
        controller.add(newSnap as DocumentSnapshot<T>);
      }

      lastDocUpdate = newVer;
    }

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

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

    return controller.stream;
  } else {
    bool isCancelled = false;
    bool hasRegisteredRealtimeListener = false;
    FusabaseAuth auth = FusabaseAuth.instanceFor(app: _oracledb.app);
    var email = auth.currentUser?.email ?? "";
    var userLvlPayload = {
      "path": this._path,
      "conditions": [],
      "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": [],
      "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<DocumentSnapshot<T>> firstPoll() async {
      dynamic oldData = await oracledb._snapStore.getData(queryId, oracledb);

      DocumentSnapshot newSnap = await get();

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

    dynamic payload = {
      "queryId": queryIdSocketInt,
      "status": 1,
      "payload": socketPayload,
    };

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

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

    return controller.stream;
  }
}