       # that is any better than the buffer implementation in nodejs streams themselves.
       stream._read = ->
 
-      opChannel = getDocOpChannel cName, docName
       redisObserver = redisLib.createClient redis.port, redis.host, redis.options
 
       open = true

        throw new Error 'Stream already closed' unless open
        stream.push null
        open = false
        redisObserver.unsubscribe opChannel
        redisObserver.quit()

      # Subscribe redis to the stream first so we don't miss out on any operations
      # while we're getting the history
      queue = []
      redisObserver.on 'message', (channel, msg) ->
        assert open
        return unless channel is opChannel
        opData = JSON.parse msg

        if queue
          queue.push opData
        else
          assert opData.v is v
          v++
          stream.push opData

      redisObserver.subscribe opChannel, (err) ->
        if err
          stream.destroy()
          return callback? err

        # Get all ops from v to current
        getOps cName, docName, v, (err, data) ->
          if err
            stream.destroy()
            return callback? err

          callback? null, stream

          # First send all the operations between v and when we called getOps
          for d in data

            if Math.random() < 0.4
              snapshotDb.setSnapshot cName, docName, {v:opData.v, data:doc} unless snapshotDb.closed

    _subscribe: (channel, callback) -> # Subscribe to a redis pubsub channel and get a nodejs stream out
      stream = new Readable objectMode:yes

      # This function is for notifying us that the stream is empty and needs data.
