var inherits = require('util').inherits;
var Readable = require('stream').Readable;
var util = require('./util');

// Stream of operations. Subscribe returns one of these
function OpStream() {
  Readable.call(this, {objectMode: true});

  this.id = null;
  this.backend = null;
  this.agent = null;
  this.projection = null;

  this.open = true;
}
module.exports = OpStream;

inherits(OpStream, Readable);

// This function is for notifying us that the stream is empty and needs data.
// For now, we'll just ignore the signal and assume the reader reads as fast
// as we fill it. I could add a buffer in this function, but really I don't
// think that is any better than the buffer implementation in nodejs streams
// themselves.
OpStream.prototype._read = util.doNothing;

OpStream.prototype.initProjection = function(backend, agent, projection) {
  this.backend = backend;
  this.agent = agent;
  this.projection = projection;
};

OpStream.prototype.pushOp = function(op) {
  // Ignore any messages after unsubscribe
  if (!this.open) return;

  if (this.backend) {
    var stream = this;
    this.backend._sanitizeOp(this.agent, this.projection, op.c, op.d, op, function(err) {
      stream.push(err ? {error: err} : op);
    });
  } else {
    this.push(op);
  }
};

OpStream.prototype.pushOps = function(ops) {
  for (var i = 0; i < ops.length; i++) {
    this.pushOp(ops[i]);
  }
};

OpStream.prototype.destroy = function() {
  if (!this.open) return;
  this.open = false;

  this.push(null);
  this.emit('close');
};

@@ -973,78 +973,8 @@
 ) %7B%0A
-  // Ignore any messages after unsubscribe%0A  if (!this.open) return;%0A%0A
   if
@@ -987,24 +987,24 @@
 .backend) %7B%0A
+
     var stre
@@ -1100,24 +1100,56 @@
 tion(err) %7B%0A
+      if (!stream.open) return;%0A
       stream
@@ -1188,16 +1188,16 @@
     %7D);%0A
-
   %7D else
@@ -1199,16 +1199,89 @@
  else %7B%0A
+    // Ignore any messages after unsubscribe%0A    if (!this.open) return;%0A
     this

fix race condition with closing op stream and sanitizing op