diff --git a/sub.js b/sub.js
index <HASH>..<HASH> 100644
--- a/sub.js
+++ b/sub.js
@@ -1,6 +1,10 @@
 var inherits = require('inherits')
 var Stream = require('./stream')
 
+function isError (end) {
+  return end && end !== true
+}
+
 module.exports = Sub
 
 inherits(Sub, Stream)
@@ -9,14 +13,16 @@ function Sub (parent, id) {
   this.parent = parent
   this.id = id
   Stream.call(this)
+  this.paused = false
 }
 
 Sub.prototype.write = function (data) {
   this.parent._write({req: this.id, value: data, stream: true, end: false})
+  this.paused = !this.parent.sink || this.parent.sink.paused
 }
 
 Sub.prototype.end = function (err) {
-  this.parent._write({req: this.id, value: data, stream: true, end: true})
+  this.parent._write({req: this.id, value: err, stream: true, end: true})
   delete this.parent.subs[this.id]
   //if we errored this stream, kill it immediately.
   if(isError(this.ended)) this._end(err)
@@ -26,5 +32,3 @@ Sub.prototype.end = function (err) {
 
 
 
-
-
sub should pause if the parent's sink is paused, or there is no sink!