dispatch.dart 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137
  1. import 'dart:ffi';
  2. import 'package:dartz/dartz.dart';
  3. import 'package:appflowy_backend/log.dart';
  4. // ignore: unnecessary_import
  5. import 'package:appflowy_backend/protobuf/dart-ffi/ffi_response.pb.dart';
  6. import 'package:appflowy_backend/protobuf/flowy-error/errors.pb.dart';
  7. import 'package:appflowy_backend/protobuf/flowy-net/network_state.pb.dart';
  8. import 'package:isolates/isolates.dart';
  9. import 'package:isolates/ports.dart';
  10. import 'package:ffi/ffi.dart';
  11. // ignore: unused_import
  12. import 'package:flutter/services.dart';
  13. import 'dart:async';
  14. import 'dart:typed_data';
  15. import 'package:appflowy_backend/ffi.dart' as ffi;
  16. import 'package:appflowy_backend/protobuf/flowy-user/protobuf.dart';
  17. import 'package:appflowy_backend/protobuf/dart-ffi/protobuf.dart';
  18. import 'package:appflowy_backend/protobuf/flowy-folder2/protobuf.dart';
  19. import 'package:appflowy_backend/protobuf/flowy-database2/protobuf.dart';
  20. import 'package:appflowy_backend/protobuf/flowy-document2/protobuf.dart';
  21. // ignore: unused_import
  22. import 'package:protobuf/protobuf.dart';
  23. import 'dart:convert' show utf8;
  24. import '../protobuf/flowy-config/entities.pb.dart';
  25. import '../protobuf/flowy-config/event_map.pb.dart';
  26. import '../protobuf/flowy-net/event_map.pb.dart';
  27. import 'error.dart';
  28. part 'dart_event/flowy-folder2/dart_event.dart';
  29. part 'dart_event/flowy-net/dart_event.dart';
  30. part 'dart_event/flowy-user/dart_event.dart';
  31. part 'dart_event/flowy-database2/dart_event.dart';
  32. part 'dart_event/flowy-document2/dart_event.dart';
  33. part 'dart_event/flowy-config/dart_event.dart';
  34. enum FFIException {
  35. RequestIsEmpty,
  36. }
  37. class DispatchException implements Exception {
  38. FFIException type;
  39. DispatchException(this.type);
  40. }
  41. class Dispatch {
  42. static Future<Either<Uint8List, Uint8List>> asyncRequest(FFIRequest request) {
  43. // FFIRequest => Rust SDK
  44. final bytesFuture = _sendToRust(request);
  45. // Rust SDK => FFIResponse
  46. final responseFuture = _extractResponse(bytesFuture);
  47. // FFIResponse's payload is the bytes of the Response object
  48. final payloadFuture = _extractPayload(responseFuture);
  49. return payloadFuture;
  50. }
  51. }
  52. Future<Either<Uint8List, Uint8List>> _extractPayload(
  53. Future<Either<FFIResponse, FlowyInternalError>> responseFuture) {
  54. return responseFuture.then((result) {
  55. return result.fold(
  56. (response) {
  57. switch (response.code) {
  58. case FFIStatusCode.Ok:
  59. return left(Uint8List.fromList(response.payload));
  60. case FFIStatusCode.Err:
  61. return right(Uint8List.fromList(response.payload));
  62. case FFIStatusCode.Internal:
  63. final error = utf8.decode(response.payload);
  64. Log.error("Dispatch internal error: $error");
  65. return right(emptyBytes());
  66. default:
  67. Log.error("Impossible to here");
  68. return right(emptyBytes());
  69. }
  70. },
  71. (error) {
  72. Log.error("Response should not be empty $error");
  73. return right(emptyBytes());
  74. },
  75. );
  76. });
  77. }
  78. Future<Either<FFIResponse, FlowyInternalError>> _extractResponse(
  79. Completer<Uint8List> bytesFuture) {
  80. return bytesFuture.future.then((bytes) {
  81. try {
  82. final response = FFIResponse.fromBuffer(bytes);
  83. return left(response);
  84. } catch (e, s) {
  85. final error = StackTraceError(e, s);
  86. Log.error('Deserialize response failed. ${error.toString()}');
  87. return right(error.asFlowyError());
  88. }
  89. });
  90. }
  91. Completer<Uint8List> _sendToRust(FFIRequest request) {
  92. Uint8List bytes = request.writeToBuffer();
  93. assert(bytes.isEmpty == false);
  94. if (bytes.isEmpty) {
  95. throw DispatchException(FFIException.RequestIsEmpty);
  96. }
  97. final Pointer<Uint8> input = calloc.allocate<Uint8>(bytes.length);
  98. final list = input.asTypedList(bytes.length);
  99. list.setAll(0, bytes);
  100. final completer = Completer<Uint8List>();
  101. final port = singleCompletePort(completer);
  102. ffi.async_event(port.nativePort, input, bytes.length);
  103. calloc.free(input);
  104. return completer;
  105. }
  106. Uint8List requestToBytes<T extends GeneratedMessage>(T? message) {
  107. try {
  108. if (message != null) {
  109. return message.writeToBuffer();
  110. } else {
  111. return emptyBytes();
  112. }
  113. } catch (e, s) {
  114. final error = StackTraceError(e, s);
  115. Log.error('Serial request failed. ${error.toString()}');
  116. return emptyBytes();
  117. }
  118. }
  119. Uint8List emptyBytes() {
  120. return Uint8List.fromList([]);
  121. }