Use synchronous file writing in BlobStreamReference (#24)

Doing so allows us to avoid dealing in futures all the way back
to `encode()`. This becomes important when implementing replay
since replay uses `noSuchMethod`, which can't await futures.

This also extracts out a few constants in preparation for their
use in replay.

Part of flutter#11

Author: tvolkert

lib/src/backends/record_replay/common.dart

@@ -2,12 +2,64 @@
 // 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 'events.dart';
+
 /// Encoded value of the file system in a recording.
 const String kFileSystemEncodedValue = '__fs__';
 
 /// The name of the recording manifest file.
 const String kManifestName = 'MANIFEST.txt';
 
+/// The key in a serialized [InvocationEvent] map that is used to store the
+/// type of invocation.
+///
+/// See also:
+///   - [kGetType]
+///   - [kSetType]
+///   - [kInvokeType]
+const String kManifestTypeKey = 'type';
+
+/// The key in a serialized [InvocationEvent] map that is used to store the
+/// target of the invocation.
+const String kManifestObjectKey = 'object';
+
+/// The key in a serialized [InvocationEvent] map that is used to store the
+/// result (return value) of the invocation.
+const String kManifestResultKey = 'result';
+
+/// The key in a serialized [InvocationEvent] map that is used to store the
+/// timestamp of the invocation.
+const String kManifestTimestampKey = 'timestamp';
+
+/// The key in a serialized [PropertyGetEvent] or [PropertySetEvent] map that
+/// is used to store the property that was accessed or mutated.
+const String kManifestPropertyKey = 'property';
+
+/// The key in a serialized [PropertySetEvent] map that is used to store the
+/// value to which the property was set.
+const String kManifestValueKey = 'value';
+
+/// The key in a serialized [MethodEvent] map that is used to store the name of
+/// the method that was invoked.
+const String kManifestMethodKey = 'method';
+
+/// The key in a serialized [MethodEvent] map that is used to store the
+/// positional arguments that were passed to the method.
+const String kManifestPositionalArgumentsKey = 'positionalArguments';
+
+/// The key in a serialized [MethodEvent] map that is used to store the
+/// named arguments that were passed to the method.
+const String kManifestNamedArgumentsKey = 'namedArguments';
+
+/// The serialized [kManifestTypeKey] for property retrievals.
+const String kGetType = 'get';
+
+/// The serialized [kManifestTypeKey] for property mutations.
+const String kSetType = 'set';
+
+/// The serialized [kManifestTypeKey] for method invocations.
+const String kInvokeType = 'invoke';
+
 /// Gets an id guaranteed to be unique on this isolate for objects within this
 /// library.
 int newUid() => _nextUid++;

lib/src/backends/record_replay/encoding.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 'dart:async';
 import 'dart:convert';
 
 import 'package:file/file.dart';
@@ -20,7 +19,7 @@ import 'result_reference.dart';
 
 /// Encodes an object into a JSON-ready representation.
 ///
-/// It is legal for an encoder to return a future value.
+/// Must return one of {number, boolean, string, null, list, or map}.
 typedef dynamic _Encoder(dynamic object);
 
 /// Known encoders. Types not covered here will be encoded using
@@ -35,8 +34,8 @@ const Map<TypeMatcher<dynamic>, _Encoder> _kEncoders =
   const TypeMatcher<bool>(): _encodeRaw,
   const TypeMatcher<String>(): _encodeRaw,
   const TypeMatcher<Null>(): _encodeRaw,
-  const TypeMatcher<Iterable<dynamic>>(): encodeIterable,
-  const TypeMatcher<Map<dynamic, dynamic>>(): encodeMap,
+  const TypeMatcher<Iterable<dynamic>>(): _encodeIterable,
+  const TypeMatcher<Map<dynamic, dynamic>>(): _encodeMap,
   const TypeMatcher<Symbol>(): getSymbolName,
   const TypeMatcher<DateTime>(): _encodeDateTime,
   const TypeMatcher<Uri>(): _encodeUri,
@@ -59,47 +58,41 @@ const Map<TypeMatcher<dynamic>, _Encoder> _kEncoders =
 /// Encodes an arbitrary [object] into a JSON-ready representation (a number,
 /// boolean, string, null, list, or map).
 ///
-/// Returns a future that completes with a value suitable for conversion into
-/// JSON using [JsonEncoder] without the need for a `toEncodable` argument.
-Future<dynamic> encode(dynamic object) async {
+/// Returns a value suitable for conversion into JSON using [JsonEncoder]
+/// without the need for a `toEncodable` argument.
+dynamic encode(dynamic object) {
   _Encoder encoder = _encodeDefault;
   for (TypeMatcher<dynamic> matcher in _kEncoders.keys) {
     if (matcher.matches(object)) {
       encoder = _kEncoders[matcher];
       break;
     }
   }
-  return await encoder(object);
+  return encoder(object);
 }
 
 /// Default encoder (used for types not covered in [_kEncoders]).
 String _encodeDefault(dynamic object) => object.runtimeType.toString();
 
-/// Pass-through encoder.
+/// Pass-through encoder (used on `num`, `bool`, `String`, and `Null`).
 dynamic _encodeRaw(dynamic object) => object;
 
 /// Encodes the specified [iterable] into a JSON-ready list of encoded items.
-///
-/// Returns a future that completes with a list suitable for conversion into
-/// JSON using [JsonEncoder] without the need for a `toEncodable` argument.
-Future<List<dynamic>> encodeIterable(Iterable<dynamic> iterable) async {
+List<dynamic> _encodeIterable(Iterable<dynamic> iterable) {
   List<dynamic> encoded = <dynamic>[];
   for (dynamic element in iterable) {
-    encoded.add(await encode(element));
+    encoded.add(encode(element));
   }
   return encoded;
 }
 
 /// Encodes the specified [map] into a JSON-ready map of encoded key/value
 /// pairs.
-///
-/// Returns a future that completes with a map suitable for conversion into
-/// JSON using [JsonEncoder] without the need for a `toEncodable` argument.
-Future<Map<String, dynamic>> encodeMap(Map<dynamic, dynamic> map) async {
+Map<String, dynamic> _encodeMap(Map<dynamic, dynamic> map) {
   Map<String, dynamic> encoded = <String, dynamic>{};
   for (dynamic key in map.keys) {
-    String encodedKey = await encode(key);
-    encoded[encodedKey] = await encode(map[key]);
+    String encodedKey = encode(key);
+    encoded[encodedKey] = encode(map[key]);
   }
   return encoded;
 }
@@ -115,10 +108,10 @@ Map<String, String> _encodePathContext(p.Context context) {
   };
 }
 
-Future<dynamic> _encodeResultReference(ResultReference<dynamic> reference) =>
+dynamic _encodeResultReference(ResultReference<dynamic> reference) =>
     reference.serializedValue;
 
-Future<Map<String, dynamic>> _encodeEvent(LiveInvocationEvent<dynamic> event) =>
+Map<String, dynamic> _encodeEvent(LiveInvocationEvent<dynamic> event) =>
     event.serialize();
 
 String _encodeFileSystem(FileSystem fs) => kFileSystemEncodedValue;

lib/src/backends/record_replay/events.dart

@@ -103,11 +103,11 @@ abstract class LiveInvocationEvent<T> implements InvocationEvent<T> {
   }
 
   /// Returns this event as a JSON-serializable object.
-  Future<Map<String, dynamic>> serialize() async {
+  Map<String, dynamic> serialize() {
     return <String, dynamic>{
-      'object': await encode(object),
-      'result': await encode(_result),
-      'timestamp': timestamp,
+      kManifestObjectKey: encode(object),
+      kManifestResultKey: encode(_result),
+      kManifestTimestampKey: timestamp,
     };
   }
 
@@ -126,11 +126,11 @@ class LivePropertyGetEvent<T> extends LiveInvocationEvent<T>
   final Symbol property;
 
   @override
-  Future<Map<String, dynamic>> serialize() async {
+  Map<String, dynamic> serialize() {
     return <String, dynamic>{
-      'type': 'get',
-      'property': getSymbolName(property),
-    }..addAll(await super.serialize());
+      kManifestTypeKey: kGetType,
+      kManifestPropertyKey: getSymbolName(property),
+    }..addAll(super.serialize());
   }
 }
 
@@ -148,12 +148,12 @@ class LivePropertySetEvent<T> extends LiveInvocationEvent<Null>
   final T value;
 
   @override
-  Future<Map<String, dynamic>> serialize() async {
+  Map<String, dynamic> serialize() {
     return <String, dynamic>{
-      'type': 'set',
-      'property': getSymbolName(property),
-      'value': await encode(value),
-    }..addAll(await super.serialize());
+      kManifestTypeKey: kSetType,
+      kManifestPropertyKey: getSymbolName(property),
+      kManifestValueKey: encode(value),
+    }..addAll(super.serialize());
   }
 }
 
@@ -185,12 +185,12 @@ class LiveMethodEvent<T> extends LiveInvocationEvent<T>
   final Map<Symbol, dynamic> namedArguments;
 
   @override
-  Future<Map<String, dynamic>> serialize() async {
+  Map<String, dynamic> serialize() {
     return <String, dynamic>{
-      'type': 'invoke',
-      'method': getSymbolName(method),
-      'positionalArguments': await encodeIterable(positionalArguments),
-      'namedArguments': await encodeMap(namedArguments),
-    }..addAll(await super.serialize());
+      kManifestTypeKey: kInvokeType,
+      kManifestMethodKey: getSymbolName(method),
+      kManifestPositionalArgumentsKey: encode(positionalArguments),
+      kManifestNamedArgumentsKey: encode(namedArguments),
+    }..addAll(super.serialize());
   }
 }

lib/src/backends/record_replay/mutable_recording.dart

@@ -46,8 +46,7 @@ class MutableRecording implements LiveRecording {
             .timeout(awaitPendingResults, onTimeout: () {});
       }
       Directory dir = destination;
-      List<dynamic> encodedEvents = await encode(_events);
-      String json = new JsonEncoder.withIndent('  ').convert(encodedEvents);
+      String json = new JsonEncoder.withIndent('  ').convert(encode(_events));
       String filename = dir.fileSystem.path.join(dir.path, kManifestName);
       await dir.fileSystem.file(filename).writeAsString(json, flush: true);
     } finally {

lib/src/backends/record_replay/recording_file.dart

@@ -20,6 +20,7 @@ import 'result_reference.dart';
 ///
 /// See also:
 ///   - [_BlobReference]
+///   - [_BlobStreamReference]
 typedef void _BlobDataSyncWriter<T>(File file, T data);
 
 /// Callback responsible for asynchronously writing result [data] to the
@@ -29,13 +30,6 @@ typedef void _BlobDataSyncWriter<T>(File file, T data);
 ///   - [_BlobFutureReference]
 typedef Future<Null> _BlobDataAsyncWriter<T>(File file, T data);
 
-/// Callback responsible writing streaming result [data] to the specified
-/// [sink].
-///
-/// See also:
-///   - [_BlobStreamReference]
-typedef void _BlobDataStreamWriter<T>(IOSink sink, T data);
-
 /// [File] implementation that records all invocation activity to its file
 /// system's recording.
 class RecordingFile extends RecordingFileSystemEntity<File> implements File {
@@ -93,8 +87,8 @@ class RecordingFile extends RecordingFileSystemEntity<File> implements File {
     return new _BlobStreamReference<List<int>>(
       file: _newRecordingFile(),
       stream: delegate.openRead(start, end),
-      writer: (IOSink sink, List<int> bytes) {
-        sink.add(bytes);
+      writer: (File file, List<int> bytes) {
+        file.writeAsBytesSync(bytes, mode: FileMode.APPEND, flush: true);
       },
     );
   }
@@ -209,7 +203,7 @@ class _BlobReference<T> extends ResultReference<T> {
   T get recordedValue => _value;
 
   @override
-  Future<String> get serializedValue async => '!${_file.basename}';
+  String get serializedValue => '!${_file.basename}';
 }
 
 /// A [FutureReference] that serializes its value data to a separate file.
@@ -235,64 +229,28 @@ class _BlobFutureReference<T> extends FutureReference<T> {
   }
 
   @override
-  Future<String> get serializedValue async => '!${_file.basename}';
+  String get serializedValue => '!${_file.basename}';
 }
 
 /// A [StreamReference] that serializes its value data to a separate file.
 class _BlobStreamReference<T> extends StreamReference<T> {
   final File _file;
-  final _BlobDataStreamWriter<T> _writer;
-  IOSink _sink;
-  Future<dynamic> _pendingFlush;
+  final _BlobDataSyncWriter<T> _writer;
 
   _BlobStreamReference({
     @required File file,
     @required Stream<T> stream,
-    @required _BlobDataStreamWriter<T> writer,
+    @required _BlobDataSyncWriter<T> writer,
   })
       : _file = file,
         _writer = writer,
-        _sink = file.openWrite(),
         super(stream);
 
   @override
   void onData(T event) {
-    if (_pendingFlush == null) {
-      _writer(_sink, event);
-    } else {
-      // It's illegal to write to an IOSink while a flush is pending.
-      // https://github.com/dart-lang/sdk/issues/28635
-      _pendingFlush.whenComplete(() {
-        _writer(_sink, event);
-      });
-    }
-  }
-
-  @override
-  void onDone() {
-    if (_sink != null) {
-      _sink.close();
-    }
-  }
-
-  @override
-  Future<String> get serializedValue async {
-    if (_pendingFlush != null) {
-      await _pendingFlush;
-    } else {
-      _pendingFlush = _sink.flush();
-      try {
-        await _pendingFlush;
-      } finally {
-        _pendingFlush = null;
-      }
-    }
-
-    return '!${_file.basename}';
+    _writer(_file, event);
   }
 
-  // TODO(tvolkert): remove `.then()` once Dart 1.22 is in stable
   @override
-  Future<Null> get complete =>
-      Future.wait(<Future<dynamic>>[super.complete, _sink.done]).then((_) {});
+  String get serializedValue => '!${_file.basename}';
 }

lib/src/backends/record_replay/result_reference.dart

@@ -49,7 +49,7 @@ abstract class ResultReference<T> {
   /// actually a byte array that was read from a file). In this case, the
   /// method can return a `ResultReference` to the list, and it will have a
   /// hook into the serialization process.
-  Future<dynamic> get serializedValue => encode(recordedValue);
+  dynamic get serializedValue => encode(recordedValue);
 
   /// A [Future] that completes when [value] has completed.
   ///
@@ -139,7 +139,6 @@ class StreamReference<T> extends ResultReference<Stream<T>> {
         _controller.addError(error, stackTrace);
       },
       onDone: () {
-        onDone();
         _completer.complete();
         _controller.close();
       },
@@ -153,13 +152,6 @@ class StreamReference<T> extends ResultReference<Stream<T>> {
   @protected
   void onData(T event) {}
 
-  /// Called when the underlying delegate stream fires a "done" event.
-  ///
-  /// Subclasses may override this method to be notified when the underlying
-  /// stream is done.
-  @protected
-  void onDone() {}
-
   @override
   Stream<T> get value => _controller.stream;