Add support for transformer clusters to barback.

pkg/barback/lib/src/phase_forwarder.dart

+// 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.phase_forwarder;
+
+import 'dart:async';
+
+import 'asset_node.dart';
+import 'asset_forwarder.dart';
+import 'phase_output.dart';
+
+/// A class that takes care of forwarding assets within a phase.
+///
+/// An asset should be forwarded if and only if it's not consumed by any
+/// transformers or clusters in a phase. Even if an asset is forwarded, it
+/// should be marked as [AssetState.DIRTY] if any of the intermediate forwarded
+/// assets are themselves marked dirty. It should only be available if all

[Bob Nystrom, 2013/10/04 21:47:50]

Can you clarify what "intermediate forwarded assets" means here?

Also, "It" -> "The forwarded asset", I think?

[nweiz, 2013/10/07 23:21:31]

Rewritten this whole thing to hopefully make it clearer. I've added specific
terms for assets in different stages of the process, and updated the other
documentation/names to use those terms.

+/// intermediate assets are available.
+///
+/// In order to determine whether to forward an asset, this compares the number
+/// of inputs to [channels]. Only if each channel has forwarded the asset does
+/// [this] emit the asset.
+class PhaseForwarder {

[Bob Nystrom, 2013/10/04 21:47:50]

"AssetForwarder"? I know it's for a Phase, but the thing it forwards is Assets.

[nweiz, 2013/10/07 23:21:31]

That's already a class that exists, and it's generally applicable to all assets
so  I think the name makes more sense for it.

The intention was this name was to follow the pattern of PhaseInput/PhaseOutput.

+  /// Sets the number of channels through which the asset may have been
+  /// forwarded.

[Bob Nystrom, 2013/10/04 21:47:50]

I think this should just be documented like a field, so: "The number of..."

[nweiz, 2013/10/07 23:21:31]

Done.

+  ///
+  /// Each cluster is a channel, along with one channel for the [PhaseInput]

[Bob Nystrom, 2013/10/04 21:47:50]

This makes me wonder if we could stratify things where Phase only contains
clusters and the "unclustered" loose transformers in a phase get put into their
own implicit cluster. Would that simplify things?

[nweiz, 2013/10/07 23:21:31]

I considered that, but it produces an infinite recursion problem. Imagine you
have a phase with a single transformer. That transformer is represented as a
cluster, which contains a phase. What does that phase contain? It can't just
contain a transformer, it has to contain a cluster! It needs to bottom out at
some point, and the "raw" transformers provide a convenient way to do that.

We could add a class that exposes the same API as ClusterRunner but runs a set
of transformers directly. I didn't explore that route mostly because this
resulted in less code churn. It's a good avenue to explore for future refactors.

[Bob Nystrom, 2013/10/07 23:58:39]

Oh, right. :)

+  /// that handles all the transformers.
+  set channels(int value) {

[Bob Nystrom, 2013/10/04 21:47:50]

"channels" -> "numChannels"

[nweiz, 2013/10/07 23:21:31]

Done.

+    _channels = value;
+    _adjustOutput();
+  }
+  int _channels;

[Bob Nystrom, 2013/10/04 21:47:50]

_numChannels

[nweiz, 2013/10/07 23:21:31]

Done.

+
+  /// The input assets.
+  final _assets = new Set<AssetNode>();
+
+  /// The forwarded output asset.
+  ///
+  /// This will be null if the asset is not being forwarded.
+  AssetNode get output => _outputController.node;
+  AssetNodeController _outputController;
+
+  /// A stream that emits an event whenever [this] starts forwarding an asset.
+  ///
+  /// Whenever this stream emits an event, the value will be identical to
+  /// [output].
+  Stream<AssetNode> get onForwarding => _onForwardingController.stream;
+  final _onForwardingController =
+      new StreamController<AssetNode>.broadcast(sync: true);
+
+  PhaseForwarder(this._channels);
+
+  /// Adds a forwarded asset to [this].
+  ///
+  /// [asset] must have the same origin as all other forwarded assets.
+  void add(AssetNode asset) {
+    if (_assets.isNotEmpty) assert(asset.origin == _assets.first.origin);
+
+    _assets.add(asset);
+
+    asset.onStateChange.listen((state) {
+      if (state.isRemoved) _assets.remove(asset);
+      _adjustOutput();
+    });
+
+    _adjustOutput();
+  }
+
+  /// Mark this forwarder as removed.
+  ///
+  /// This will remove [output] if it exists.
+  void remove() {
+    if (_outputController != null) {
+      _outputController.setRemoved();
+      _outputController = null;
+    }
+    _onForwardingController.close();
+  }
+
+  /// Adjusts [output] to ensure that it accurately reflects the current state
+  /// of the forwarded nodes.
+  void _adjustOutput() {
+    assert(_assets.length <= _channels);
+    assert(!_assets.any((asset) => asset.state.isRemoved));
+
+    // If there are any channels that haven't forwarded an asset, we shouldn't
+    // emit an asset. If we are currently, remove it.
+    if (_assets.length < _channels) {
+      if (_outputController == null) return;
+      _outputController.setRemoved();
+      _outputController = null;
+      return;
+    }
+
+    // If there isn't an asset being emitted yet, we should emit one. It should
+    // be dirty iff any of the source assets are dirty.
+    if (_outputController == null) {
+      var asset = _assets.firstWhere((asset) => asset.state.isDirty,
+          orElse: () => _assets.first);
+      _outputController = new AssetNodeController.from(asset);
+      _onForwardingController.add(output);
+      return;
+    }
+
+    // If we're already emitting an asset, set it dirty iff any of the source
+    // assets are dirty.
+    if (_assets.any((asset) => asset.state.isDirty)) {
+      if (!_outputController.node.state.isDirty) _outputController.setDirty();
+    } else {
+      if (!_outputController.node.state.isAvailable) {
+        _outputController.setAvailable(_assets.first.asset);
+      }
+    }
+  }
+}