Index: mojo/public/dart/third_party/barback/lib/src/graph/transform_node.dart |
diff --git a/mojo/public/dart/third_party/barback/lib/src/graph/transform_node.dart b/mojo/public/dart/third_party/barback/lib/src/graph/transform_node.dart |
new file mode 100644 |
index 0000000000000000000000000000000000000000..61bab3f258943d925ef0be951fec42fe2f1156e5 |
--- /dev/null |
+++ b/mojo/public/dart/third_party/barback/lib/src/graph/transform_node.dart |
@@ -0,0 +1,886 @@ |
+// Copyright (c) 2013, 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. |
+ |
+library barback.graph.transform_node; |
+ |
+import 'dart:async'; |
+ |
+import '../asset/asset.dart'; |
+import '../asset/asset_id.dart'; |
+import '../asset/asset_node.dart'; |
+import '../asset/asset_node_set.dart'; |
+import '../errors.dart'; |
+import '../log.dart'; |
+import '../transformer/aggregate_transform.dart'; |
+import '../transformer/aggregate_transformer.dart'; |
+import '../transformer/declaring_aggregate_transform.dart'; |
+import '../transformer/declaring_aggregate_transformer.dart'; |
+import '../transformer/lazy_aggregate_transformer.dart'; |
+import '../utils.dart'; |
+import 'node_status.dart'; |
+import 'node_streams.dart'; |
+import 'phase.dart'; |
+import 'transformer_classifier.dart'; |
+ |
+/// Describes a transform on a set of assets and its relationship to the build |
+/// dependency graph. |
+/// |
+/// Keeps track of whether it's dirty and needs to be run and which assets it |
+/// depends on. |
+class TransformNode { |
+ /// The aggregate key for this node. |
+ final String key; |
+ |
+ /// The [TransformerClassifier] that [this] belongs to. |
+ final TransformerClassifier classifier; |
+ |
+ /// The [Phase] that this transform runs in. |
+ Phase get phase => classifier.phase; |
+ |
+ /// The [AggregateTransformer] to apply to this node's inputs. |
+ final AggregateTransformer transformer; |
+ |
+ /// The primary asset nodes this transform runs on. |
+ final _primaries = new AssetNodeSet(); |
+ |
+ /// A string describing the location of [this] in the transformer graph. |
+ final String _location; |
+ |
+ /// The subscription to the [_primaries]' [AssetNode.onStateChange] streams. |
+ final _primarySubscriptions = new Map<AssetId, StreamSubscription>(); |
+ |
+ /// The subscription to [phase]'s [Phase.onAsset] stream. |
+ StreamSubscription<AssetNode> _phaseAssetSubscription; |
+ |
+ /// The subscription to [phase]'s [Phase.onStatusChange] stream. |
+ StreamSubscription<NodeStatus> _phaseStatusSubscription; |
+ |
+ /// How far along [this] is in processing its assets. |
+ NodeStatus get status { |
+ if (_state == _State.APPLIED || _state == _State.DECLARED) { |
+ return NodeStatus.IDLE; |
+ } |
+ |
+ if (_declaring && _state != _State.DECLARING && |
+ _state != _State.NEEDS_DECLARE) { |
+ return NodeStatus.MATERIALIZING; |
+ } else { |
+ return NodeStatus.RUNNING; |
+ } |
+ } |
+ |
+ /// The [TransformInfo] describing this node. |
+ /// |
+ /// [TransformInfo] is the publicly-visible representation of a transform |
+ /// node. |
+ TransformInfo get info => new TransformInfo(transformer, |
+ new AssetId(phase.cascade.package, key)); |
+ |
+ /// Whether this is a declaring transform. |
+ /// |
+ /// This is usually identical to `transformer is |
+ /// DeclaringAggregateTransformer`, but if a declaring and non-lazy |
+ /// transformer emits an error during `declareOutputs` it's treated as though |
+ /// it wasn't declaring. |
+ bool get _declaring => transformer is DeclaringAggregateTransformer && |
+ (_state == _State.DECLARING || _declaredOutputs != null); |
+ |
+ /// Whether this transform has been forced since it last finished applying. |
+ /// |
+ /// A transform being forced means it should run until it generates outputs |
+ /// and is no longer dirty. This is always true for non-declaring |
+ /// transformers, since they always need to eagerly generate outputs. |
+ bool _forced; |
+ |
+ /// The subscriptions to each secondary input's [AssetNode.onStateChange] |
+ /// stream. |
+ final _secondarySubscriptions = new Map<AssetId, StreamSubscription>(); |
+ |
+ /// The subscriptions to the [AssetCascade.onAsset] stream for cascades that |
+ /// might generate assets in [_missingInputs]. |
+ final _missingExternalInputSubscriptions = |
+ new Map<String, StreamSubscription>(); |
+ |
+ /// The controllers for the asset nodes emitted by this node. |
+ final _outputControllers = new Map<AssetId, AssetNodeController>(); |
+ |
+ /// The ids of inputs the transformer tried and failed to read last time it |
+ /// ran. |
+ final _missingInputs = new Set<AssetId>(); |
+ |
+ /// The controllers that are used to pass each primary input through [this] if |
+ /// it's not consumed or overwritten. |
+ /// |
+ /// This needs an intervening controller to ensure that the output can be |
+ /// marked dirty when determining whether [this] will consume or overwrite it, |
+ /// and be marked removed if it does. No pass-through controller will exist |
+ /// for primary inputs that are not being passed through. |
+ final _passThroughControllers = new Map<AssetId, AssetNodeController>(); |
+ |
+ /// The asset node for this transform. |
+ final _streams = new NodeStreams(); |
+ Stream<NodeStatus> get onStatusChange => _streams.onStatusChange; |
+ Stream<AssetNode> get onAsset => _streams.onAsset; |
+ Stream<LogEntry> get onLog => _streams.onLog; |
+ |
+ /// The current state of [this]. |
+ var _state = _State.DECLARED; |
+ |
+ /// Whether [this] has been marked as removed. |
+ bool get _isRemoved => _streams.onAssetController.isClosed; |
+ |
+ // If [transformer] is declaring but not lazy and [primary] is available, we |
+ // can run [apply] even if [force] hasn't been called, since [transformer] |
+ // should run eagerly if possible. |
+ bool get _canRunDeclaringEagerly => |
+ _declaring && transformer is! LazyAggregateTransformer && |
+ _primaries.every((input) => input.state.isAvailable); |
+ |
+ /// Which primary inputs the most recent run of this transform has declared |
+ /// that it consumes. |
+ /// |
+ /// This starts out `null`, indicating that the transform hasn't declared |
+ /// anything yet. This is not meaningful unless [_state] is [_State.APPLIED] |
+ /// or [_State.DECLARED]. |
+ Set<AssetId> _consumedPrimaries; |
+ |
+ /// The set of output ids that [transformer] declared it would emit. |
+ /// |
+ /// This is only non-null if [transformer] is a |
+ /// [DeclaringAggregateTransformer] and its [declareOutputs] has been run |
+ /// successfully. |
+ Set<AssetId> _declaredOutputs; |
+ |
+ /// The controller for the currently-running |
+ /// [DeclaringAggregateTransformer.declareOutputs] call's |
+ /// [DeclaringAggregateTransform]. |
+ /// |
+ /// This will be non-`null` when |
+ /// [DeclaringAggregateTransformer.declareOutputs] is running. This means that |
+ /// it's always non-`null` when [_state] is [_State.DECLARING], sometimes |
+ /// non-`null` when it's [_State.NEEDS_DECLARE], and always `null` otherwise. |
+ DeclaringAggregateTransformController _declareController; |
+ |
+ /// The controller for the currently-running [AggregateTransformer.apply] |
+ /// call's [AggregateTransform]. |
+ /// |
+ /// This will be non-`null` when [AggregateTransform.apply] is running, which |
+ /// means that it's always non-`null` when [_state] is [_State.APPLYING] or |
+ /// [_State.NEEDS_APPLY], sometimes non-`null` when it's |
+ /// [_State.NEEDS_DECLARE], and always `null` otherwise. |
+ AggregateTransformController _applyController; |
+ |
+ /// The number of secondary inputs that have been requested but not yet |
+ /// produced. |
+ int _pendingSecondaryInputs = 0; |
+ |
+ /// A stopwatch that tracks the total time spent in a transformer's `apply` |
+ /// function. |
+ final _timeInTransformer = new Stopwatch(); |
+ |
+ /// A stopwatch that tracks the time in a transformer's `apply` function spent |
+ /// waiting for [getInput] calls to complete. |
+ final _timeAwaitingInputs = new Stopwatch(); |
+ |
+ TransformNode(this.classifier, this.transformer, this.key, this._location) { |
+ _forced = transformer is! DeclaringAggregateTransformer; |
+ |
+ _phaseAssetSubscription = phase.previous.onAsset.listen((node) { |
+ if (!_missingInputs.contains(node.id)) return; |
+ if (_forced) node.force(); |
+ _dirty(); |
+ }); |
+ |
+ _phaseStatusSubscription = phase.previous.onStatusChange.listen((status) { |
+ if (status == NodeStatus.RUNNING) return; |
+ |
+ _maybeFinishDeclareController(); |
+ _maybeFinishApplyController(); |
+ }); |
+ |
+ classifier.onDoneClassifying.listen((_) { |
+ _maybeFinishDeclareController(); |
+ _maybeFinishApplyController(); |
+ }); |
+ |
+ _run(); |
+ } |
+ |
+ /// Adds [input] as a primary input for this node. |
+ void addPrimary(AssetNode input) { |
+ _primaries.add(input); |
+ if (_forced) input.force(); |
+ |
+ _primarySubscriptions[input.id] = input.onStateChange |
+ .listen((_) => _onPrimaryStateChange(input)); |
+ |
+ if (_state == _State.DECLARING && !_declareController.isDone) { |
+ // If we're running `declareOutputs` and its id stream isn't closed yet, |
+ // pass this in as another id. |
+ _declareController.addId(input.id); |
+ _maybeFinishDeclareController(); |
+ } else if (_state == _State.APPLYING) { |
+ // If we're running `apply`, we need to wait until [input] is available |
+ // before we pass it into the stream. If it's available now, great; if |
+ // not, [_onPrimaryStateChange] will handle it. |
+ if (!input.state.isAvailable) { |
+ // If we started running eagerly without being forced, abort that run if |
+ // a new unavailable asset comes in. |
+ if (input.isLazy && !_forced) _restartRun(); |
+ return; |
+ } |
+ |
+ _onPrimaryStateChange(input); |
+ _maybeFinishApplyController(); |
+ } else { |
+ // Otherwise, a new input means we'll need to re-run `declareOutputs`. |
+ _restartRun(); |
+ } |
+ } |
+ |
+ /// Marks this transform as removed. |
+ /// |
+ /// This causes all of the transform's outputs to be marked as removed as |
+ /// well. Normally this will be automatically done internally based on events |
+ /// from the primary input, but it's possible for a transform to no longer be |
+ /// valid even if its primary input still exists. |
+ void remove() { |
+ _streams.close(); |
+ _phaseAssetSubscription.cancel(); |
+ _phaseStatusSubscription.cancel(); |
+ if (_declareController != null) _declareController.cancel(); |
+ if (_applyController != null) _applyController.cancel(); |
+ _clearSecondarySubscriptions(); |
+ _clearOutputs(); |
+ |
+ for (var subscription in _primarySubscriptions.values) { |
+ subscription.cancel(); |
+ } |
+ _primarySubscriptions.clear(); |
+ |
+ for (var controller in _passThroughControllers.values) { |
+ controller.setRemoved(); |
+ } |
+ _passThroughControllers.clear(); |
+ } |
+ |
+ /// If [this] is deferred, ensures that its concrete outputs will be |
+ /// generated. |
+ void force() { |
+ if (_forced || _state == _State.APPLIED) return; |
+ for (var input in _primaries) { |
+ input.force(); |
+ } |
+ |
+ _forced = true; |
+ if (_state == _State.DECLARED) _apply(); |
+ } |
+ |
+ /// Marks this transform as dirty. |
+ /// |
+ /// Specifically, this should be called when one of the transform's inputs' |
+ /// contents change, or when a secondary input is removed. Primary inputs |
+ /// being added or removed are handled by [addInput] and |
+ /// [_onPrimaryStateChange]. |
+ void _dirty() { |
+ if (_state == _State.DECLARING || _state == _State.NEEDS_DECLARE || |
+ _state == _State.NEEDS_APPLY) { |
+ // If we already know that [_apply] needs to be run, there's nothing to do |
+ // here. |
+ return; |
+ } |
+ |
+ if (!_forced && !_canRunDeclaringEagerly) { |
+ // [forced] should only ever be false for a declaring transformer. |
+ assert(_declaring); |
+ |
+ // If we've finished applying, transition to DECLARED, indicating that we |
+ // know what outputs [apply] will emit but we're waiting to emit them |
+ // concretely until [force] is called. If we're still applying, we'll |
+ // transition to DECLARED once we finish. |
+ if (_state == _State.APPLIED) _state = _State.DECLARED; |
+ for (var controller in _outputControllers.values) { |
+ controller.setLazy(force); |
+ } |
+ _emitDeclaredOutputs(); |
+ return; |
+ } |
+ |
+ if (_state == _State.APPLIED) { |
+ if (_declaredOutputs != null) _emitDeclaredOutputs(); |
+ _apply(); |
+ } else if (_state == _State.DECLARED) { |
+ _apply(); |
+ } else { |
+ _state = _State.NEEDS_APPLY; |
+ } |
+ } |
+ |
+ /// The callback called when [input]'s state changes. |
+ void _onPrimaryStateChange(AssetNode input) { |
+ if (input.state.isRemoved) { |
+ _primarySubscriptions.remove(input.id); |
+ |
+ if (_primaries.isEmpty) { |
+ // If there are no more primary inputs, there's no more use for this |
+ // node in the graph. It will be re-created by its |
+ // [TransformerClassifier] if a new input with [key] is added. |
+ remove(); |
+ return; |
+ } |
+ |
+ // Any change to the number of primary inputs requires that we re-run the |
+ // transformation. |
+ _restartRun(); |
+ } else if (input.state.isAvailable) { |
+ if (_state == _State.DECLARED) { |
+ // If we're passing through this input and its contents don't matter, |
+ // update the pass-through controller. |
+ var controller = _passThroughControllers[input.id]; |
+ if (controller != null) controller.setAvailable(input.asset); |
+ } |
+ |
+ if (_state == _State.DECLARED && _canRunDeclaringEagerly) { |
+ // If [this] is fully declared but hasn't started applying, this input |
+ // becoming available may mean that all inputs are available, in which |
+ // case we can run apply eagerly. |
+ _apply(); |
+ return; |
+ } |
+ |
+ // If we're not actively passing concrete assets to the transformer, the |
+ // distinction between a dirty asset and an available one isn't relevant. |
+ if (_state != _State.APPLYING) return; |
+ |
+ if (_applyController.isDone) { |
+ // If we get a new asset after we've closed the asset stream, we need to |
+ // re-run declare and then apply. |
+ _restartRun(); |
+ } else { |
+ // If the new asset comes before the asset stream is done, we can just |
+ // pass it to the stream. |
+ _applyController.addInput(input.asset); |
+ _maybeFinishApplyController(); |
+ } |
+ } else { |
+ if (_forced) input.force(); |
+ |
+ var controller = _passThroughControllers[input.id]; |
+ if (controller != null) controller.setDirty(); |
+ |
+ if (_state == _State.APPLYING && !_applyController.addedId(input.id) && |
+ (_forced || !input.isLazy)) { |
+ // If the input hasn't yet been added to the transform's input stream, |
+ // there's no need to consider the transformation dirty. However, if the |
+ // input is lazy and we're running eagerly, we need to restart the |
+ // transformation. |
+ return; |
+ } |
+ _dirty(); |
+ } |
+ } |
+ |
+ /// Run the entire transformation, including both `declareOutputs` (if |
+ /// applicable) and `apply`. |
+ void _run() { |
+ assert(_state != _State.DECLARING); |
+ assert(_state != _State.APPLYING); |
+ |
+ _markOutputsDirty(); |
+ _declareOutputs(() { |
+ if (_forced || _canRunDeclaringEagerly) { |
+ _apply(); |
+ } else { |
+ _state = _State.DECLARED; |
+ _streams.changeStatus(NodeStatus.IDLE); |
+ } |
+ }); |
+ } |
+ |
+ /// Restart the entire transformation, including `declareOutputs` if |
+ /// applicable. |
+ void _restartRun() { |
+ if (_state == _State.DECLARED || _state == _State.APPLIED) { |
+ // If we're currently idle, we can restart the transformation immediately. |
+ _run(); |
+ return; |
+ } |
+ |
+ // If we're actively running `declareOutputs` or `apply`, cancel the |
+ // transforms and transition to `NEEDS_DECLARE`. Once the transformer's |
+ // method returns, we'll transition to `DECLARING`. |
+ if (_declareController != null) _declareController.cancel(); |
+ if (_applyController != null) _applyController.cancel(); |
+ _state = _State.NEEDS_DECLARE; |
+ } |
+ |
+ /// Runs [transform.declareOutputs] and emits the resulting assets as dirty |
+ /// assets. |
+ /// |
+ /// Calls [callback] when it's finished. This doesn't return a future so that |
+ /// [callback] is called synchronously if there are no outputs to declare. If |
+ /// [this] is removed while inputs are being declared, [callback] will not be |
+ /// called. |
+ void _declareOutputs(void callback()) { |
+ if (transformer is! DeclaringAggregateTransformer) { |
+ callback(); |
+ return; |
+ } |
+ |
+ _state = _State.DECLARING; |
+ var controller = new DeclaringAggregateTransformController(this); |
+ _declareController = controller; |
+ _streams.onLogPool.add(controller.onLog); |
+ for (var primary in _primaries) { |
+ controller.addId(primary.id); |
+ } |
+ _maybeFinishDeclareController(); |
+ |
+ syncFuture(() { |
+ return (transformer as DeclaringAggregateTransformer) |
+ .declareOutputs(controller.transform); |
+ }).whenComplete(() { |
+ // Cancel the controller here even if `declareOutputs` wasn't interrupted. |
+ // Since the declaration is finished, we want to close out the |
+ // controller's streams. |
+ controller.cancel(); |
+ _declareController = null; |
+ }).then((_) { |
+ if (_isRemoved) return; |
+ if (_state == _State.NEEDS_DECLARE) { |
+ _declareOutputs(callback); |
+ return; |
+ } |
+ |
+ if (controller.loggedError) { |
+ // If `declareOutputs` fails, fall back to treating a declaring |
+ // transformer as though it were eager. |
+ if (transformer is! LazyAggregateTransformer) _forced = true; |
+ callback(); |
+ return; |
+ } |
+ |
+ _consumedPrimaries = controller.consumedPrimaries; |
+ _declaredOutputs = controller.outputIds; |
+ var invalidIds = _declaredOutputs |
+ .where((id) => id.package != phase.cascade.package).toSet(); |
+ for (var id in invalidIds) { |
+ _declaredOutputs.remove(id); |
+ // TODO(nweiz): report this as a warning rather than a failing error. |
+ phase.cascade.reportError(new InvalidOutputException(info, id)); |
+ } |
+ |
+ for (var primary in _primaries) { |
+ if (_declaredOutputs.contains(primary.id)) continue; |
+ _passThrough(primary.id); |
+ } |
+ _emitDeclaredOutputs(); |
+ callback(); |
+ }).catchError((error, stackTrace) { |
+ if (_isRemoved) return; |
+ if (transformer is! LazyAggregateTransformer) _forced = true; |
+ phase.cascade.reportError(_wrapException(error, stackTrace)); |
+ callback(); |
+ }); |
+ } |
+ |
+ /// Emits a dirty asset node for all outputs that were declared by the |
+ /// transformer. |
+ /// |
+ /// This won't emit any outputs for which there already exist output |
+ /// controllers. It should only be called for transforms that have declared |
+ /// their outputs. |
+ void _emitDeclaredOutputs() { |
+ assert(_declaredOutputs != null); |
+ for (var id in _declaredOutputs) { |
+ if (_outputControllers.containsKey(id)) continue; |
+ var controller = _forced |
+ ? new AssetNodeController(id, this) |
+ : new AssetNodeController.lazy(id, force, this); |
+ _outputControllers[id] = controller; |
+ _streams.onAssetController.add(controller.node); |
+ } |
+ } |
+ |
+ //// Mark all emitted and passed-through outputs of this transform as dirty. |
+ void _markOutputsDirty() { |
+ for (var controller in _passThroughControllers.values) { |
+ controller.setDirty(); |
+ } |
+ for (var controller in _outputControllers.values) { |
+ if (_forced) { |
+ controller.setDirty(); |
+ } else { |
+ controller.setLazy(force); |
+ } |
+ } |
+ } |
+ |
+ /// Applies this transform. |
+ void _apply() { |
+ assert(!_isRemoved); |
+ |
+ _markOutputsDirty(); |
+ _clearSecondarySubscriptions(); |
+ _state = _State.APPLYING; |
+ _streams.changeStatus(status); |
+ _runApply().then((hadError) { |
+ if (_isRemoved) return; |
+ |
+ if (_state == _State.DECLARED) return; |
+ |
+ if (_state == _State.NEEDS_DECLARE) { |
+ _run(); |
+ return; |
+ } |
+ |
+ // If an input's contents changed while running `apply`, retry unless the |
+ // transformer is deferred and hasn't been forced. |
+ if (_state == _State.NEEDS_APPLY) { |
+ if (_forced || _canRunDeclaringEagerly) { |
+ _apply(); |
+ } else { |
+ _state = _State.DECLARED; |
+ } |
+ return; |
+ } |
+ |
+ if (_declaring) _forced = false; |
+ |
+ assert(_state == _State.APPLYING); |
+ if (hadError) { |
+ _clearOutputs(); |
+ // If the transformer threw an error, we don't want to emit the |
+ // pass-through assets in case they'll be overwritten by the |
+ // transformer. However, if the transformer declared that it wouldn't |
+ // overwrite or consume a pass-through asset, we can safely emit it. |
+ if (_declaredOutputs != null) { |
+ for (var input in _primaries) { |
+ if (_consumedPrimaries.contains(input.id) || |
+ _declaredOutputs.contains(input.id)) { |
+ _consumePrimary(input.id); |
+ } else { |
+ _passThrough(input.id); |
+ } |
+ } |
+ } |
+ } |
+ |
+ _state = _State.APPLIED; |
+ _streams.changeStatus(NodeStatus.IDLE); |
+ }); |
+ } |
+ |
+ /// Gets the asset for an input [id]. |
+ /// |
+ /// If an input with [id] cannot be found, throws an [AssetNotFoundException]. |
+ Future<Asset> getInput(AssetId id) { |
+ _timeAwaitingInputs.start(); |
+ _pendingSecondaryInputs++; |
+ return phase.previous.getOutput(id).then((node) { |
+ // Throw if the input isn't found. This ensures the transformer's apply |
+ // is exited. We'll then catch this and report it through the proper |
+ // results stream. |
+ if (node == null) { |
+ _missingInputs.add(id); |
+ |
+ // If this id is for an asset in another package, subscribe to that |
+ // package's asset cascade so when it starts emitting the id we know to |
+ // re-run the transformer. |
+ if (id.package != phase.cascade.package) { |
+ var stream = phase.cascade.graph.onAssetFor(id.package); |
+ if (stream != null) { |
+ _missingExternalInputSubscriptions.putIfAbsent(id.package, () { |
+ return stream.listen((node) { |
+ if (!_missingInputs.contains(node.id)) return; |
+ if (_forced) node.force(); |
+ _dirty(); |
+ }); |
+ }); |
+ } |
+ } |
+ |
+ throw new AssetNotFoundException(id); |
+ } |
+ |
+ _secondarySubscriptions.putIfAbsent(node.id, () { |
+ return node.onStateChange.listen((_) => _dirty()); |
+ }); |
+ |
+ return node.asset; |
+ }).whenComplete(() { |
+ _pendingSecondaryInputs--; |
+ if (_pendingSecondaryInputs != 0) _timeAwaitingInputs.stop(); |
+ }); |
+ } |
+ |
+ /// Run [AggregateTransformer.apply]. |
+ /// |
+ /// Returns whether or not an error occurred while running the transformer. |
+ Future<bool> _runApply() { |
+ var controller = new AggregateTransformController(this); |
+ _applyController = controller; |
+ _streams.onLogPool.add(controller.onLog); |
+ for (var primary in _primaries) { |
+ if (!primary.state.isAvailable) continue; |
+ controller.addInput(primary.asset); |
+ } |
+ _maybeFinishApplyController(); |
+ |
+ return syncFuture(() { |
+ _timeInTransformer.reset(); |
+ _timeAwaitingInputs.reset(); |
+ _timeInTransformer.start(); |
+ return transformer.apply(controller.transform); |
+ }).whenComplete(() { |
+ _timeInTransformer.stop(); |
+ _timeAwaitingInputs.stop(); |
+ |
+ // Cancel the controller here even if `apply` wasn't interrupted. Since |
+ // the apply is finished, we want to close out the controller's streams. |
+ controller.cancel(); |
+ _applyController = null; |
+ }).then((_) { |
+ assert(_state != _State.DECLARED); |
+ assert(_state != _State.DECLARING); |
+ assert(_state != _State.APPLIED); |
+ |
+ if (!_forced && _primaries.any((node) => !node.state.isAvailable)) { |
+ _state = _State.DECLARED; |
+ _streams.changeStatus(NodeStatus.IDLE); |
+ return false; |
+ } |
+ |
+ if (_isRemoved) return false; |
+ if (_state == _State.NEEDS_APPLY) return false; |
+ if (_state == _State.NEEDS_DECLARE) return false; |
+ if (controller.loggedError) return true; |
+ |
+ // If the transformer took long enough, log its duration in fine output. |
+ // That way it's not always visible, but users running with "pub serve |
+ // --verbose" can see it. |
+ var ranLong = _timeInTransformer.elapsed > new Duration(seconds: 1); |
+ var ranLongLocally = |
+ _timeInTransformer.elapsed - _timeAwaitingInputs.elapsed > |
+ new Duration(milliseconds: 200); |
+ |
+ // Report the transformer's timing information if it spent more than 0.2s |
+ // doing things other than waiting for its secondary inputs or if it spent |
+ // more than 1s in total. |
+ if (ranLongLocally || ranLong) { |
+ _streams.onLogController.add(new LogEntry( |
+ info, info.primaryId, LogLevel.FINE, |
+ "Took ${niceDuration(_timeInTransformer.elapsed)} " |
+ "(${niceDuration(_timeAwaitingInputs.elapsed)} awaiting " |
+ "secondary inputs).", |
+ null)); |
+ } |
+ |
+ _handleApplyResults(controller); |
+ return false; |
+ }).catchError((error, stackTrace) { |
+ // If the transform became dirty while processing, ignore any errors from |
+ // it. |
+ if (_state == _State.NEEDS_APPLY || _isRemoved) return false; |
+ |
+ // Catch all transformer errors and pipe them to the results stream. This |
+ // is so a broken transformer doesn't take down the whole graph. |
+ phase.cascade.reportError(_wrapException(error, stackTrace)); |
+ return true; |
+ }); |
+ } |
+ |
+ /// Handle the results of running [Transformer.apply]. |
+ /// |
+ /// [controller] should be the controller for the [AggegateTransform] passed |
+ /// to [AggregateTransformer.apply]. |
+ void _handleApplyResults(AggregateTransformController controller) { |
+ _consumedPrimaries = controller.consumedPrimaries; |
+ |
+ var newOutputs = controller.outputs; |
+ // Any ids that are for a different package are invalid. |
+ var invalidIds = newOutputs |
+ .map((asset) => asset.id) |
+ .where((id) => id.package != phase.cascade.package) |
+ .toSet(); |
+ for (var id in invalidIds) { |
+ newOutputs.removeId(id); |
+ // TODO(nweiz): report this as a warning rather than a failing error. |
+ phase.cascade.reportError(new InvalidOutputException(info, id)); |
+ } |
+ |
+ // Remove outputs that used to exist but don't anymore. |
+ for (var id in _outputControllers.keys.toList()) { |
+ if (newOutputs.containsId(id)) continue; |
+ _outputControllers.remove(id).setRemoved(); |
+ } |
+ |
+ // Emit or stop emitting pass-through assets between removing and adding |
+ // outputs to ensure there are no collisions. |
+ for (var id in _primaries.map((node) => node.id)) { |
+ if (_consumedPrimaries.contains(id) || newOutputs.containsId(id)) { |
+ _consumePrimary(id); |
+ } else { |
+ _passThrough(id); |
+ } |
+ } |
+ |
+ // Store any new outputs or new contents for existing outputs. |
+ for (var asset in newOutputs) { |
+ var controller = _outputControllers[asset.id]; |
+ if (controller != null) { |
+ controller.setAvailable(asset); |
+ } else { |
+ var controller = new AssetNodeController.available(asset, this); |
+ _outputControllers[asset.id] = controller; |
+ _streams.onAssetController.add(controller.node); |
+ } |
+ } |
+ } |
+ |
+ /// Cancels all subscriptions to secondary input nodes and other cascades. |
+ void _clearSecondarySubscriptions() { |
+ _missingInputs.clear(); |
+ for (var subscription in _secondarySubscriptions.values) { |
+ subscription.cancel(); |
+ } |
+ for (var subscription in _missingExternalInputSubscriptions.values) { |
+ subscription.cancel(); |
+ } |
+ _secondarySubscriptions.clear(); |
+ _missingExternalInputSubscriptions.clear(); |
+ } |
+ |
+ /// Removes all output assets. |
+ void _clearOutputs() { |
+ // Remove all the previously-emitted assets. |
+ for (var controller in _outputControllers.values) { |
+ controller.setRemoved(); |
+ } |
+ _outputControllers.clear(); |
+ } |
+ |
+ /// Emit the pass-through node for the primary input [id] if it's not being |
+ /// emitted already. |
+ void _passThrough(AssetId id) { |
+ assert(!_outputControllers.containsKey(id)); |
+ |
+ if (_consumedPrimaries.contains(id)) return; |
+ var controller = _passThroughControllers[id]; |
+ var primary = _primaries[id]; |
+ if (controller == null) { |
+ controller = new AssetNodeController.from(primary); |
+ _passThroughControllers[id] = controller; |
+ _streams.onAssetController.add(controller.node); |
+ } else if (primary.state.isDirty) { |
+ controller.setDirty(); |
+ } else if (!controller.node.state.isAvailable) { |
+ controller.setAvailable(primary.asset); |
+ } |
+ } |
+ |
+ /// Stops emitting the pass-through node for the primary input [id] if it's |
+ /// being emitted. |
+ void _consumePrimary(AssetId id) { |
+ var controller = _passThroughControllers.remove(id); |
+ if (controller == null) return; |
+ controller.setRemoved(); |
+ } |
+ |
+ /// If `declareOutputs` is running and all previous phases have declared their |
+ /// outputs, mark [_declareController] as done. |
+ void _maybeFinishDeclareController() { |
+ if (_declareController == null) return; |
+ if (classifier.isClassifying) return; |
+ if (phase.previous.status == NodeStatus.RUNNING) return; |
+ _declareController.done(); |
+ } |
+ |
+ /// If `apply` is running, all previous phases have declared their outputs, |
+ /// and all primary inputs are available and thus have been passed to the |
+ /// transformer, mark [_applyController] as done. |
+ void _maybeFinishApplyController() { |
+ if (_applyController == null) return; |
+ if (classifier.isClassifying) return; |
+ if (_primaries.any((input) => !input.state.isAvailable)) return; |
+ if (phase.previous.status == NodeStatus.RUNNING) return; |
+ _applyController.done(); |
+ } |
+ |
+ BarbackException _wrapException(error, StackTrace stackTrace) { |
+ if (error is! AssetNotFoundException) { |
+ return new TransformerException(info, error, stackTrace); |
+ } else { |
+ return new MissingInputException(info, error.id); |
+ } |
+ } |
+ |
+ String toString() => |
+ "transform node in $_location for $transformer on ${info.primaryId} " |
+ "($_state, $status, ${_forced ? '' : 'un'}forced)"; |
+} |
+ |
+/// The enum of states that [TransformNode] can be in. |
+class _State { |
+ /// The transform is running [DeclaringAggregateTransformer.declareOutputs]. |
+ /// |
+ /// If the set of primary inputs changes while in this state, it will |
+ /// transition to [NEEDS_DECLARE]. If the [TransformNode] is still in this |
+ /// state when `declareOutputs` finishes running, it will transition to |
+ /// [APPLYING] if the transform is non-lazy and all of its primary inputs are |
+ /// available, and [DECLARED] otherwise. |
+ /// |
+ /// Non-declaring transformers will transition out of this state and into |
+ /// [APPLYING] immediately. |
+ static const DECLARING = const _State._("declaring outputs"); |
+ |
+ /// The transform is running [AggregateTransformer.declareOutputs] or |
+ /// [AggregateTransform.apply], but a primary input was added or removed after |
+ /// it started, so it will need to re-run `declareOutputs`. |
+ /// |
+ /// The [TransformNode] will transition to [DECLARING] once `declareOutputs` |
+ /// or `apply` finishes running. |
+ static const NEEDS_DECLARE = const _State._("needs declare"); |
+ |
+ /// The transform is deferred and has run |
+ /// [DeclaringAggregateTransformer.declareOutputs] but hasn't yet been forced. |
+ /// |
+ /// The [TransformNode] will transition to [APPLYING] when one of the outputs |
+ /// has been forced or if the transformer is non-lazy and all of its primary |
+ /// inputs become available. |
+ static const DECLARED = const _State._("declared"); |
+ |
+ /// The transform is running [AggregateTransformer.apply]. |
+ /// |
+ /// If an input's contents change or a secondary input is added or removed |
+ /// while in this state, the [TransformNode] will transition to [NEEDS_APPLY]. |
+ /// If a primary input is added or removed, it will transition to |
+ /// [NEEDS_DECLARE]. If it's still in this state when `apply` finishes |
+ /// running, it will transition to [APPLIED]. |
+ static const APPLYING = const _State._("applying"); |
+ |
+ /// The transform is running [AggregateTransformer.apply], but an input's |
+ /// contents changed or a secondary input was added or removed after it |
+ /// started, so it will need to re-run `apply`. |
+ /// |
+ /// If a primary input is added or removed while in this state, the |
+ /// [TranformNode] will transition to [NEEDS_DECLARE]. If it's still in this |
+ /// state when `apply` finishes running, it will transition to [APPLYING]. |
+ static const NEEDS_APPLY = const _State._("needs apply"); |
+ |
+ /// The transform has finished running [AggregateTransformer.apply], whether |
+ /// or not it emitted an error. |
+ /// |
+ /// If an input's contents change or a secondary input is added or removed, |
+ /// the [TransformNode] will transition to [DECLARED] if the transform is |
+ /// declaring and [APPLYING] otherwise. If a primary input is added or |
+ /// removed, this will transition to [DECLARING]. |
+ static const APPLIED = const _State._("applied"); |
+ |
+ final String name; |
+ |
+ const _State._(this.name); |
+ |
+ String toString() => name; |
+} |