dispatch.dart 3.3 KB

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