listen method

Future<StreamSubscription<Uint8List>?> listen(dynamic onData(BagelEvent))

Trigger a callback when an item gets updated created or deleted on BagelDB on a full collection or on a single item

Implementation

Future<StreamSubscription<Uint8List>?> listen(
    Function(BagelEvent) onData) async {
  String requestID = "", nestedID = "";
  if (nestedCollectionsIDs.isNotEmpty) {
    nestedID = nestedCollectionsIDs.join(".");
  }
  Response<ResponseBody> rs;
  List<dynamic> data = [];
  BagelResponse res = await get();
  if (_item != null && nestedCollectionsIDs.length % 2 != 0) {
    data.add(res.data);
  } else {
    data = res.data;
  }
  StreamSubscription<Uint8List>? subscription;

  _listen() async {
    String token =
        BagelUsersRequest(bagelDB).getAccessToken() ?? bagelDB.token;
    String uri =
        '$liveEndpoint/collection/$collectionID/live?authorization=$token&requestID=$requestID&nestedID=$nestedID&itemID=${_item ?? ""}&everything=$everything';
    Dio dio = await _dio();
    rs = await dio.get<ResponseBody>(uri,
        options: Options(
          responseType: ResponseType.stream,
          headers: {'Access-Control-Allow-Origin': true},
        ));
    subscription = rs.data?.stream.listen((e) async {
      Event event = Event.fromUint8List(e);
      if (event.type == "start") {
        requestID = event.data;
        onData(BagelEvent(trigger: event.type, data: data));
      } else if (event.type == "stop") {
        _listen();
      } else if (event.type == "message") {
        dynamic response;

        response = {};
        RegExp r = RegExp(r'"itemID":"(\w*)');
        response["itemID"] = r.firstMatch(event.data)?.group(1);
        _item = response["itemID"];
        response["item"] = {};
        RegExp trgr = RegExp(r'"trigger":"(\w*)');
        response["trigger"] = trgr.firstMatch(event.data)?.group(1);
        switch (response["trigger"]) {
          case "update":
            var i = data.indexWhere(
                (element) => element["_id"] == response["itemID"]);
            if (i != -1) {
              BagelResponse res = await get();
              response["item"] = res.data;
              data[i] = response["item"];
              onData(BagelEvent.fromPayload(response, data));
            }
            break;
          case "delete":
            data = data
                .where((element) => element["_id"] != response["itemID"])
                .toList();
            onData(BagelEvent.fromPayload(response, data));
            break;
          case "create":
            BagelResponse res = await get();
            response["item"] = res.data;
            data.add(response["item"]);
            onData(BagelEvent.fromPayload(response, data));
            break;
        }
      }
    });
  }

  _listen();
  return subscription;
}