dispatch.dart 4.0 KB

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