chore: stream search result

Author: appflowy

frontend/appflowy_flutter/lib/workspace/application/command_palette/command_palette_bloc.dart

@@ -2,7 +2,6 @@ import 'dart:async';
 
 import 'package:appflowy/plugins/trash/application/trash_listener.dart';
 import 'package:appflowy/plugins/trash/application/trash_service.dart';
-import 'package:appflowy/workspace/application/command_palette/search_listener.dart';
 import 'package:appflowy/workspace/application/command_palette/search_service.dart';
 import 'package:appflowy_backend/protobuf/flowy-folder/trash.pb.dart';
 import 'package:appflowy_backend/protobuf/flowy-search/notification.pb.dart';
@@ -13,33 +12,22 @@ import 'package:freezed_annotation/freezed_annotation.dart';
 
 part 'command_palette_bloc.freezed.dart';
 
-const _searchChannel = 'CommandPalette';
-
 class CommandPaletteBloc
     extends Bloc<CommandPaletteEvent, CommandPaletteState> {
   CommandPaletteBloc() : super(CommandPaletteState.initial()) {
-    _searchListener.start(
-      onResultsChanged: _onResultsChanged,
-    );
-
     _initTrash();
     _dispatch();
   }
 
   Timer? _debounceOnChanged;
   final TrashService _trashService = TrashService();
-  final SearchListener _searchListener = SearchListener(
-    channel: _searchChannel,
-  );
   final TrashListener _trashListener = TrashListener();
   String? _oldQuery;
   String? _workspaceId;
-  int _messagesReceived = 0;
 
   @override
   Future<void> close() {
     _trashListener.close();
-    _searchListener.stop();
     _debounceOnChanged?.cancel();
     return super.close();
   }
@@ -67,10 +55,23 @@ class CommandPaletteBloc
           if (search.isNotEmpty && search != state.query) {
             _oldQuery = state.query;
             emit(state.copyWith(query: search, isLoading: true));
-            await SearchBackendService.performSearch(
-              search,
-              workspaceId: _workspaceId,
-              channel: _searchChannel,
+            unawaited(
+              SearchBackendService.performSearch(
+                search,
+                workspaceId: _workspaceId,
+              ).then(
+                (result) {
+                  result.onSuccess(
+                    (stream) async {
+                      if (!isClosed) {
+                        add(
+                          CommandPaletteEvent.newSearchStream(stream: stream),
+                        );
+                      }
+                    },
+                  );
+                },
+              ),
             );
           } else {
             emit(
@@ -83,30 +84,41 @@ class CommandPaletteBloc
             );
           }
         },
-        resultsChanged: (SearchResultNotificationPB notification) {
+        newSearchStream: (SearchResponseStream stream) {
+          if (state.searchResponseStream != null) {
+            state.searchResponseStream!.dispose();
+          }
+
+          emit(state.copyWith(searchId: stream.searchId));
+          stream.listen(
+            onData: (SearchResponsePB response) {
+              if (!isClosed) {
+                add(CommandPaletteEvent.resultsChanged(response: response));
+              }
+            },
+          );
+        },
+        resultsChanged: (SearchResponsePB response) {
           if (state.query != _oldQuery) {
             emit(
               state.copyWith(
                 resultItems: [],
                 resultSummaries: [],
-                isLoading: true,
+                isLoading: response.isLoading,
               ),
             );
             _oldQuery = state.query;
-            _messagesReceived = 0;
           }
 
-          if (state.query != notification.query) {
+          if (state.searchId != response.searchId) {
             return;
           }
 
-          _messagesReceived++;
-
           emit(
             state.copyWith(
-              resultItems: notification.result.items,
-              resultSummaries: notification.result.summaries,
-              isLoading: _messagesReceived != notification.sends.toInt(),
+              resultItems: response.result.items,
+              resultSummaries: response.result.summaries,
+              isLoading: response.isLoading,
             ),
           );
         },
@@ -128,6 +140,7 @@ class CommandPaletteBloc
               resultSummaries: [],
               query: '',
               isLoading: false,
+              searchId: null,
             ),
           );
         },
@@ -159,9 +172,6 @@ class CommandPaletteBloc
 
   void _performSearch(String value) =>
       add(CommandPaletteEvent.performSearch(search: value));
-
-  void _onResultsChanged(SearchResultNotificationPB results) =>
-      add(CommandPaletteEvent.resultsChanged(results: results));
 }
 
 @freezed
@@ -172,8 +182,12 @@ class CommandPaletteEvent with _$CommandPaletteEvent {
   const factory CommandPaletteEvent.performSearch({required String search}) =
       _PerformSearch;
 
+  const factory CommandPaletteEvent.newSearchStream({
+    required SearchResponseStream stream,
+  }) = _NewSearchStream;
+
   const factory CommandPaletteEvent.resultsChanged({
-    required SearchResultNotificationPB results,
+    required SearchResponsePB response,
   }) = _ResultsChanged;
 
   const factory CommandPaletteEvent.trashChanged({
@@ -195,8 +209,10 @@ class CommandPaletteState with _$CommandPaletteState {
     @Default(null) String? query,
     @Default([]) List<SearchResponseItemPB> resultItems,
     @Default([]) List<SearchSummaryPB> resultSummaries,
+    @Default(null) SearchResponseStream? searchResponseStream,
     required bool isLoading,
     @Default([]) List<TrashPB> trash,
+    @Default(null) String? searchId,
   }) = _CommandPaletteState;
 
   factory CommandPaletteState.initial() => CommandPaletteState(

frontend/appflowy_flutter/lib/workspace/application/command_palette/search_listener.dart

@@ -1,22 +1,67 @@
+import 'dart:async';
+import 'dart:ffi';
+import 'dart:isolate';
+import 'dart:typed_data';
+
 import 'package:appflowy_backend/dispatch/dispatch.dart';
 import 'package:appflowy_backend/protobuf/flowy-error/errors.pb.dart';
+import 'package:appflowy_backend/protobuf/flowy-search/notification.pb.dart';
 import 'package:appflowy_backend/protobuf/flowy-search/query.pb.dart';
 import 'package:appflowy_backend/protobuf/flowy-search/search_filter.pb.dart';
 import 'package:appflowy_result/appflowy_result.dart';
+import 'package:nanoid/nanoid.dart';
+import 'package:fixnum/fixnum.dart';
 
 class SearchBackendService {
-  static Future<FlowyResult<void, FlowyError>> performSearch(
+  static Future<FlowyResult<SearchResponseStream, FlowyError>> performSearch(
     String keyword, {
     String? workspaceId,
-    String? channel,
   }) async {
+    final searchId = nanoid(6);
+    final stream = SearchResponseStream(searchId: searchId);
+
     final filter = SearchFilterPB(workspaceId: workspaceId);
     final request = SearchQueryPB(
       search: keyword,
       filter: filter,
-      channel: channel,
+      searchId: searchId,
+      streamPort: Int64(stream.nativePort),
+    );
+
+    unawaited(SearchEventSearch(request).send());
+    return FlowyResult.success(stream);
+  }
+}
+
+class SearchResponseStream {
+  SearchResponseStream({required this.searchId}) {
+    _port.handler = _controller.add;
+    _subscription = _controller.stream.listen(
+      (Uint8List data) => _onResultsChanged(data),
     );
+  }
+
+  final String searchId;
+  final RawReceivePort _port = RawReceivePort();
+  final StreamController<Uint8List> _controller = StreamController.broadcast();
+  late StreamSubscription<Uint8List> _subscription;
+  void Function(SearchResponsePB response)? _onData;
+
+  int get nativePort => _port.sendPort.nativePort;
+
+  Future<void> dispose() async {
+    await _subscription.cancel();
+    _port.close();
+  }
+
+  void _onResultsChanged(Uint8List data) {
+    final response = SearchResponsePB.fromBuffer(data);
+    _onData?.call(response);
+  }
 
-    return SearchEventSearch(request).send();
+  void listen({
+    required void Function(SearchResponsePB response)? onData,
+  }) {
+    _onData = onData;
   }
 }

frontend/appflowy_flutter/lib/workspace/application/command_palette/search_service.dart

@@ -144,34 +144,34 @@ EXTERNAL SOURCES:
     :path: Flutter/ephemeral/.symlinks/plugins/window_manager/macos
 
 SPEC CHECKSUMS:
-  app_links: 9028728e32c83a0831d9db8cf91c526d16cc5468
-  appflowy_backend: 464aeb3e5c6966a41641a2111e5ead72ce2695f7
-  auto_updater_macos: 3a42f1a06be6981f1a18be37e6e7bf86aa732118
-  bitsdojo_window_macos: 7959fb0ca65a3ccda30095c181ecb856fae48ea9
-  connectivity_plus: e74b9f74717d2d99d45751750e266e55912baeb5
-  desktop_drop: e0b672a7d84c0a6cbc378595e82cdb15f2970a43
-  device_info_plus: a56e6e74dbbd2bb92f2da12c64ddd4f67a749041
-  file_selector_macos: 6280b52b459ae6c590af5d78fc35c7267a3c4b31
-  flowy_infra_ui: 8760ff42a789de40bf5007a5f176b454722a341e
+  app_links: 10e0a0ab602ffaf34d142cd4862f29d34b303b2a
+  appflowy_backend: 865496343de667fc8c600e04b9fd05234e130cf9
+  auto_updater_macos: 3e3462c418fe4e731917eacd8d28eef7af84086d
+  bitsdojo_window_macos: 44e3b8fe3dd463820e0321f6256c5b1c16bb6a00
+  connectivity_plus: 18d3c32514c886e046de60e9c13895109866c747
+  desktop_drop: 69eeff437544aa619c8db7f4481b3a65f7696898
+  device_info_plus: ce1b7762849d3ec103d0e0517299f2db7ad60720
+  file_selector_macos: cc3858c981fe6889f364731200d6232dac1d812d
+  flowy_infra_ui: 03301a39ad118771adbf051a664265c61c507f38
   FlutterMacOS: 8f6f14fa908a6fb3fba0cd85dbd81ec4b251fb24
   HotKey: 400beb7caa29054ea8d864c96f5ba7e5b4852277
-  hotkey_manager: b443f35f4d772162937aa73fd8995e579f8ac4e2
-  irondash_engine_context: 893c7d96d20ce361d7e996f39d360c4c2f9869ba
-  local_notifier: ebf072651e35ae5e47280ad52e2707375cb2ae4e
-  package_info_plus: f0052d280d17aa382b932f399edf32507174e870
-  path_provider_foundation: 080d55be775b7414fd5a5ef3ac137b97b097e564
+  hotkey_manager: c32bf0bfe8f934b7bc17ab4ad5c4c142960b023c
+  irondash_engine_context: da62996ee25616d2f01bbeb85dc115d813359478
+  local_notifier: e9506bc66fc70311e8bc7291fb70f743c081e4ff
+  package_info_plus: 12f1c5c2cfe8727ca46cbd0b26677728972d9a5b
+  path_provider_foundation: 2b6b4c569c0fb62ec74538f866245ac84301af46
   ReachabilitySwift: 32793e867593cfc1177f5d16491e3a197d2fccda
-  screen_retriever_macos: 452e51764a9e1cdb74b3c541238795849f21557f
+  screen_retriever_macos: 776e0fa5d42c6163d2bf772d22478df4b302b161
   Sentry: 1fe34e9c2cbba1e347623610d26db121dcb569f1
-  sentry_flutter: e24b397f9a61fa5bbefd8279c3b2242ca86faa90
-  share_plus: 510bf0af1a42cd602274b4629920c9649c52f4cc
-  shared_preferences_foundation: 9e1978ff2562383bd5676f64ec4e9aa8fa06a6f7
+  sentry_flutter: a39c2a2d67d5e5b9cb0b94a4985c76dd5b3fc737
+  share_plus: 1fa619de8392a4398bfaf176d441853922614e89
+  shared_preferences_foundation: fcdcbc04712aee1108ac7fda236f363274528f78
   Sparkle: 5f8960a7a119aa7d45dacc0d5837017170bc5675
-  sqflite_darwin: 20b2a3a3b70e43edae938624ce550a3cbf66a3d0
-  super_native_extensions: c2795d6d9aedf4a79fae25cb6160b71b50549189
-  url_launcher_macos: 0fba8ddabfc33ce0a9afe7c5fef5aab3d8d2d673
-  webview_flutter_wkwebview: 44d4dee7d7056d5ad185d25b38404436d56c547c
-  window_manager: 1d01fa7ac65a6e6f83b965471b1a7fdd3f06166c
+  sqflite_darwin: 5a7236e3b501866c1c9befc6771dfd73ffb8702d
+  super_native_extensions: 85efee3a7495b46b04befcfc86ed12069264ebf3
+  url_launcher_macos: c82c93949963e55b228a30115bd219499a6fe404
+  webview_flutter_wkwebview: 0982481e3d9c78fd5c6f62a002fcd24fc791f1e4
+  window_manager: 3a1844359a6295ab1e47659b1a777e36773cd6e8
 
 PODFILE CHECKSUM: 0532f3f001ca3110b8be345d6491fff690e95823