dispatch.dart 3.2 KB

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