[macros] Split generation explicitly into numbered rounds. (#3861)

Author: davidmorgan

working/macros/dart_model/dart_model/lib/query.dart

@@ -2,7 +2,6 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
-import 'delta.dart';
 import 'model.dart';
 
 extension type Query.fromJson(Map<String, Object?> node) {
@@ -91,5 +90,5 @@ extension type Operation.fromJson(Map<String, Object?> node) {
 
 abstract interface class Service {
   Future<Model> query(Query query);
-  Future<Stream<Delta>> watch(Query query);
+  Future<void> changeFiles(Iterable<String> files);
 }

working/macros/dart_model/dart_model_analyzer_service/lib/dart_model_analyzer_service.dart

@@ -3,7 +3,6 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:async';
-import 'dart:convert';
 import 'dart:io';
 
 import 'package:analyzer/dart/analysis/analysis_context.dart';
@@ -12,20 +11,27 @@ import 'package:analyzer/dart/analysis/session.dart';
 import 'package:analyzer/dart/element/element.dart';
 // ignore: implementation_imports
 import 'package:analyzer/src/dart/constant/value.dart';
-import 'package:dart_model/delta.dart';
 import 'package:dart_model/model.dart';
 import 'package:dart_model/query.dart';
-import 'package:stream_transform/stream_transform.dart';
+import 'package:pool/pool.dart';
 
 // ignore_for_file: deprecated_member_use
 class DartModelAnalyzerService implements Service {
   final AnalysisContext? context;
   AnalysisSession? session;
+  final Pool pool = Pool(1);
 
   DartModelAnalyzerService({this.context, this.session});
 
   @override
   Future<Model> query(Query query) async {
+    // Lock so we don't query while reanalyzing changed files.
+    return await pool.withResource(() {
+      return _query(query);
+    });
+  }
+
+  Future<Model> _query(Query query) async {
     if (context != null) {
       session = context!.currentSession;
     }
@@ -71,29 +77,18 @@ class DartModelAnalyzerService implements Service {
   }
 
   @override
-  Future<Stream<Delta>> watch(Query query) async {
-    if (context != null) {
-      session = context!.currentSession;
+  Future<void> changeFiles(Iterable<String> files) async {
+    // Lock so we don't query while reanalyzing changed files.
+    return await pool.withResource(() {
+      return _changeFiles(files);
+    });
+  }
+
+  Future<void> _changeFiles(Iterable<String> files) async {
+    for (final file in files) {
+      context!.changeFile(file);
     }
-    // TODO(davidmorgan): watch recursivly.
-    final changes = Directory('${context!.contextRoot.root.path}/lib').watch();
-    Model previousModel = Model();
-    return Stream.fromIterable(<FileSystemEvent?>[null])
-        .followedBy(changes)
-        .asyncMap((change) async {
-          if (change != null) context!.changeFile(change.path);
-          await context!.applyPendingFileChanges();
-          final model = await this.query(query);
-          var delta = Delta.compute(previousModel, model);
-          // Round trip to check serialization works.
-          // TODO(davidmorgan): add test coverage.
-          delta = Delta.fromJson(json.decode(json.encode(delta)));
-          previousModel = model;
-          return delta.isEmpty ? null : delta;
-        })
-        .where((e) => e != null)
-        .map((e) => e!)
-        .asBroadcastStream();
+    await context!.applyPendingFileChanges();
   }
 
   Model queryLibrary(LibraryElement libraryElement, Query query) {

working/macros/dart_model/dart_model_analyzer_service/pubspec.yaml

@@ -7,6 +7,7 @@ environment:
 dependencies:
   analyzer: '>=5.2.0 <7.0.0'
   dart_model: any
+  pool: ^1.5.1
   stream_transform: any
 
 dependency_overrides:

working/macros/dart_model/dart_model_repl/lib/dart_model_repl.dart

@@ -2,6 +2,7 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
@@ -11,11 +12,12 @@ import 'package:async/async.dart';
 import 'package:dart_model/model.dart';
 import 'package:dart_model/query.dart';
 import 'package:dart_model_analyzer_service/dart_model_analyzer_service.dart';
+import 'package:macro_host/macro_host.dart';
 
 class DartModelRepl {
   final _stdinLines =
       StreamQueue(LineSplitter().bind(Utf8Decoder().bind(stdin)));
-  Service? service;
+  MacroHost? host;
 
   Future<void> run() async {
     print(''''
@@ -26,8 +28,8 @@ Welcome to ${green}package:dart_model_repl$reset!
   }
 
   Future<bool> readInput() async {
-    if (service == null) {
-      print('No service running; try "analyze <workspace>".');
+    if (host == null) {
+      print('No host running; try "analyze <workspace>".');
     }
 
     stdout.write('> ');
@@ -39,7 +41,7 @@ Welcome to ${green}package:dart_model_repl$reset!
     }
 
     if (line.startsWith('analyze ')) {
-      createAnalyzer(line.substring('analyze '.length));
+      createHost(line.substring('analyze '.length));
       return true;
     }
 
@@ -50,7 +52,7 @@ Welcome to ${green}package:dart_model_repl$reset!
           ? Query.uri(rest)
           : Query.qualifiedName(
               uri: maybeQualifiedName.uri, name: maybeQualifiedName.name);
-      final model = await service!.query(query);
+      final model = await host!.service.query(query);
       print(model.prettyPrint());
       return true;
     }
@@ -63,7 +65,8 @@ Welcome to ${green}package:dart_model_repl$reset!
           : Query.qualifiedName(
               uri: maybeQualifiedName.uri, name: maybeQualifiedName.name);
       final model = Model();
-      (await service!.watch(query)).listen((delta) {
+      host!.watch(query, 0, (round) {
+        final delta = round.delta;
         delta.update(model);
         print('=== current model for $query');
         print(model.prettyPrint());
@@ -89,12 +92,24 @@ watch <URI>[#name]
 ''');
   }
 
-  void createAnalyzer(String workspace) {
+  void createHost(String workspace) {
     final contextBuilder = ContextBuilder();
     final analysisContext = contextBuilder.createContext(
         contextRoot:
             ContextLocator().locateRoots(includedPaths: [workspace]).first);
-    service = DartModelAnalyzerService(context: analysisContext);
+    final service = DartModelAnalyzerService(context: analysisContext);
+    File? uriConverter(Uri uri) {
+      final path = analysisContext.currentSession.uriConverter.uriToPath(uri);
+      if (path == null) return null;
+      return File(path);
+    }
+
+    host = MacroHost(workspace, service, uriConverter);
+
+    // TOD0(davidmorgan): this is a hack to keep the REPL working after file
+    // watching moved from the service to the host; consider whether it's worth
+    // doing properly.
+    unawaited(host!.run());
   }
 }
 

working/macros/dart_model/dart_model_repl/pubspec.yaml

@@ -9,6 +9,7 @@ dependencies:
   async: ^2.11.0
   dart_model: any
   dart_model_analyzer_service: any
+  macro_host: any
 
 dependency_overrides:
   dart_model:
@@ -17,3 +18,9 @@ dependency_overrides:
   dart_model_analyzer_service:
     path:
       ../dart_model_analyzer_service
+  macro_host:
+    path:
+      ../macro_host
+  macro_protocol:
+    path:
+      ../macro_protocol

working/macros/dart_model/macro_client/lib/socket_host.dart

@@ -6,19 +6,18 @@ import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
-import 'package:dart_model/delta.dart';
 import 'package:dart_model/model.dart';
 import 'package:dart_model/query.dart';
 import 'package:macro_protocol/host.dart';
 import 'package:macro_protocol/message.dart';
 
-class SocketHost implements Service, Host {
+class SocketHost implements Host {
   final Socket socket;
   final StreamController<Object?> messagesController =
       StreamController.broadcast();
   Stream<Object?> get messages => messagesController.stream;
   int _id = 0;
-  final Map<int, StreamController<Delta>> _deltaStreams = {};
+  final Map<int, StreamController<Round>> _roundStreams = {};
 
   SocketHost(this.socket) {
     socket
@@ -29,38 +28,30 @@ class SocketHost implements Service, Host {
   }
 
   @override
-  Future<Model> query(Query query) async {
-    socket.writeln(json.encode(QueryRequest(query)));
-    return Model.fromJson((await messages.first) as Map<String, Object?>);
-  }
-
-  @override
-  Future<Stream<Delta>> watch(Query query) async {
+  Future<Stream<Round>> watch(Query query) async {
     ++_id;
     socket.writeln(json.encode(WatchRequest(query: query, id: _id)));
-    final result = StreamController<Delta>();
-    _deltaStreams[_id] = result;
+    final result = StreamController<Round>();
+    _roundStreams[_id] = result;
     return result.stream;
   }
 
   void _handle(Message message) {
-    if (message.isWatchResponse) {
-      final response = message.asWatchResponse;
-      _deltaStreams[response.id]!.add(response.delta);
+    if (message.isRound) {
+      final response = message.asRound;
+      _roundStreams[response.id]!.add(response);
     } else {
       messagesController.add(message);
     }
   }
 
-  @override
-  Service get service => this;
-
   @override
   Future<void> augment({
     required QualifiedName macro,
+    required int round,
     required Map<String, String> augmentationsByUri,
   }) async {
-    socket.writeln(json.encode(
-        AugmentRequest(macro: macro, augmentationsByUri: augmentationsByUri)));
+    socket.writeln(json.encode(AugmentRequest(
+        macro: macro, round: round, augmentationsByUri: augmentationsByUri)));
   }
 }

working/macros/dart_model/macro_host/bin/main.dart

@@ -25,7 +25,7 @@ Future<void> main(List<String> arguments) async {
       contextRoot:
           ContextLocator().locateRoots(includedPaths: [workspace]).first);
   final host = DartModelAnalyzerService(context: analysisContext);
-  await MacroHost(host, (uri) {
+  await MacroHost(workspace, host, (uri) {
     final path = analysisContext.currentSession.uriConverter.uriToPath(uri);
     if (path == null) return null;
     return File(path);