Update dataflow error handling.

src/Dataflow.js

@@ -107,6 +107,13 @@ function logMethod(method) {
   };
 }
 
+/**
+ * Logs an error message. By default, logged messages are written to console
+ * output. The message will only be logged if the current log level is high
+ * enough to permit error messages.
+ */
+prototype.error = logMethod('error');
+
 /**
  * Logs a warning message. By default, logged messages are written to console
  * output. The message will only be logged if the current log level is high
@@ -135,11 +142,3 @@ prototype.debug = logMethod('debug');
  * @return {number} - The current log level.
  */
 prototype.logLevel = logMethod('level');
-
-/**
- * Handle an error. By default, this method re-throws the input error.
- * This method can be overridden for custom error handling.
- */
-prototype.error = function(err) {
-  throw err;
-};

src/dataflow/events.js

@@ -17,8 +17,13 @@ export default function(source, type, filter, apply) {
       s = stream(filter, apply),
       send = function(e) {
         e.dataflow = df;
-        s.receive(e);
-        df.run();
+        try {
+          s.receive(e);
+        } catch (error) {
+          df.error(error);
+        } finally {
+          df.run();
+        }
       },
       sources;
 

src/dataflow/load.js

@@ -43,9 +43,9 @@ export function request(target, url, format) {
         df.ingest(target, data, format);
       },
       function(error) {
-        df.warn('Loading failed: ' + url, error);
+        df.error('Loading failed: ' + url, error);
         pending.done();
       })
     .then(pending.done)
-    .catch(function(error) { df.warn(error); });
+    .catch(function(error) { df.error(error); });
 }

src/dataflow/run.js

@@ -30,6 +30,7 @@ export function run(encode) {
 
   if (df._pulse) {
     df.error('Dataflow invoked recursively. Use the runAfter method to queue invocation.');
+    return 0;
   }
 
   if (!df._touched.length) {

test/core/dataflow-test.js

@@ -34,13 +34,10 @@ tape('Dataflow handles errors', function(test) {
   var df = new vega.Dataflow(),
       error = 0;
 
+  df.error = function() { error = 1; };
   df.add(function() { throw Error('!!!') });
 
-  try {
-    df.run();
-  } catch (err) {
-    error = 1;
-  }
+  df.run();
 
   test.equal(error, 1);
   test.equal(df._pulse, null);

test/transforms/prefacet-test.js

@@ -92,6 +92,8 @@ tape('PreFacet raises error if tuple sets are modified', function(test) {
   var tuples = util.field('tuples'),
       df = new vega.Dataflow(),
       source = df.add(Collect);
+
+  df.error = function(e) { throw e; };
   df.add(PreFacet, {subflow:subflow, field:tuples, pulse:source});
 
   // -- add