Converted Observatory heap-snapshot element

runtime/observatory/lib/src/repositories/heap_snapshot.dart

+// Copyright (c) 2016, the Dart project authors.  Please see the AUTHORS file
+// 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
+
+part of repositories;
+
+class HeapSnapshotLoadingProgressEvent
+  implements M.HeapSnapshotLoadingProgressEvent {
+  final HeapSnapshotLoadingProgress progress;
+  HeapSnapshotLoadingProgressEvent(this.progress);
+}
+
+class HeapSnapshotLoadingProgress extends M.HeapSnapshotLoadingProgress {
+  StreamController<HeapSnapshotLoadingProgressEvent> _onProgress =
+      new StreamController<HeapSnapshotLoadingProgressEvent>.broadcast();
+  Stream<HeapSnapshotLoadingProgressEvent> get onProgress =>
+      _onProgress.stream;
+
+  final S.Isolate isolate;
+  final bool gc;
+
+  M.HeapSnapshotLoadingStatus _status = M.HeapSnapshotLoadingStatus.fetching;
+  String _stepDescription = '';
+  double _progress = 0.0;
+  final Stopwatch _fetchingTime = new Stopwatch();
+  final Stopwatch  _loadingTime = new Stopwatch();
+  HeapSnapshot _snapshot;
+
+  M.HeapSnapshotLoadingStatus get status => _status;
+  String get stepDescription => _stepDescription;
+  double get progress => _progress;
+  Duration get fetchingTime => _fetchingTime.elapsed;
+  Duration get loadingTime => _loadingTime.elapsed;
+  HeapSnapshot get snapshot => _snapshot;
+
+  HeapSnapshotLoadingProgress(this.isolate, this.gc) {
+      _run();
+  }
+
+  Future _run() async {
+    _fetchingTime.start();
+    try {
+      _status = M.HeapSnapshotLoadingStatus.fetching;
+      _triggerOnProgress();
+
+      await isolate.getClassRefs();
+
+      final stream = isolate.fetchHeapSnapshot(gc);
+
+      stream.listen((status) {
+        if (status is List) {
+          _progress = status[0] * 100.0 / status[1];
+          _stepDescription = 'Receiving snapshot chunk ${status[0] + 1}'
+                            ' of ${status[1]}...';
+          _triggerOnProgress();
+        }
+      });
+
+      final response = await stream.last;
+
+      _fetchingTime.stop();
+      _loadingTime.start();
+      _status = M.HeapSnapshotLoadingStatus.loading;
+      _stepDescription = '';
+      _triggerOnProgress();
+
+      HeapSnapshot snapshot = new HeapSnapshot();
+
+      Stream<List> progress = snapshot.loadProgress(isolate, response);
+      progress.listen((value) {
+        _stepDescription = value[0];
+        _progress = value[1];
+        _triggerOnProgress();
+      });
+
+      await progress.drain();
+
+      _snapshot = snapshot;
+
+      _loadingTime.stop();
+      _status = M.HeapSnapshotLoadingStatus.loaded;
+      _triggerOnProgress();
+    } finally {
+      _onProgress.close();
+    }
+  }
+
+  void _triggerOnProgress() {
+    _onProgress.add(new HeapSnapshotLoadingProgressEvent(this));
+  }
+
+  void reuse() {
+    _onProgress =
+      new StreamController<HeapSnapshotLoadingProgressEvent>.broadcast();
+    (() async {
+      _triggerOnProgress();
+      _onProgress.close();
+    }());
+  }
+}
+
+class HeapSnapshotRepository
+    implements M.HeapSnapshotRepository {
+
+  Stream<HeapSnapshotLoadingProgressEvent> get(M.IsolateRef i,
+                                               {bool gc: false}) {
+    S.Isolate isolate = i as S.Isolate;
+    assert(isolate != null);
+    assert(gc != null);
+    return new HeapSnapshotLoadingProgress(isolate, gc).onProgress;
+  }
+}