gistfile1.js

// This file is provided to you under the Apache License,
// Version 2.0 (the "License"); you may not use this file
// except in compliance with the License.  You may obtain
// a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

var sys = require('sys'),
  Riak = require('../lib/riak-node');
  //assert = require('assert');


var settings = {  }; //headers: { 'X-Riak-ClientId': 'nodejs'}
var db = new Riak.Client({port: 8091});
var test_bucket = 'test-bucket-1';
var test_bucket_1 = 'test-bucket-2';
var test_bucket_stats = 'test-bucket-1-stats';
var test_bucket_lineage = 'test-bucket-1-lineage';
var test_bucket_index = [];
var volume = 30000;

util = {
  
  zeroPad: function (digits, n) {
    n = n.toString();
    while (n.length < digits) 
      n = '0' + n;
    return Number(n);
  },

  randomNum: function(){

	return Math.floor(Math.random()*99999999999);
	
  },

  randomString: function (length) {
    var chars = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXTZabcdefghiklmnopqrstuvwxyz'.split('');

    if (! length) {
        length = Math.floor(Math.random() * chars.length);
    }

    var str = '';
    for (var i = 0; i < length; i++) {
        str += chars[Math.floor(Math.random() * chars.length)];
    }
    return str;
  },

  randomData: function(){

    return {rn: this.randomNum(), rs: this.randomString(10)};
	
  }

};

var mycallback = function(d,m){
	var data=JSON.stringify(d), meta=JSON.stringify(m)
	db.log('call: ' + data+meta)
	
}

var myerrback = function(d,m){
	var d=JSON.stringify(d), m=JSON.stringify(m)
	db.log('err: ' + d+m)
	
	
}


function loadData () {
	/*
	var lineageWcommas = "1,4,5,3", lineageWpipes = "1|4|5|3", lineageWcarrots = "1^4^5^3";
	db.save(test_bucket_lineage, lineageWcommas, {delim: 'commas'})();
	db.save(test_bucket_lineage, lineageWpipes, {delim: 'pipes'})();
	db.save(test_bucket_lineage, lineageWcarrots, {delim: 'carrots'})();
	*/
	
	db.log('INSERTING KEYS');
	for( var i = 1 ; i <= volume ; i++){
	  
	  var rn = util.randomNum(), 
	      //rs = '\'' + util.randomString(10) + '\'' ;
	      rs = util.randomString(10)  ;
	     
	
	  test_bucket_index.push(i);
      
      //db.log('PUT test_bucket, ' + i + ', {numVal: ' + rn + ', staticStringVal: test_string' + ', dynStringVal: ' + rs + '}');

      db.head(test_bucket, i )(
	    //callback
	    function(cd,cm){
		  var  rd = util.randomData(),
		  payload = {numVal: rd['rn'], staticStringVal: 'test_string', dynStringVal: rd['rs']} ;
		  //db.log('PUT ' + test_bucket + '/' + cm['key'] +'\n>-------d: '+ JSON.stringify(payload));
		  db.log('\n>-------d: '+ JSON.stringify(payload)
		   + '\n>-----key: +' + cm['key'] + '\n>--vclock: ' + JSON.stringify(cm['headers']['x-riak-vclock'])
		   + '\n>----etag: ' + JSON.stringify(cm['headers']['etag']));
		  
		  db.save(test_bucket, cm['key'], payload, { headers:{ 'X-Riak-Vclock': cm['headers']['x-riak-vclock'] } } )();
		  /*  */


	    }  , //callback
	    //errback
	    function(ed,em){
		  var  rd = util.randomData(),
		  payload = {numVal: rd['rn'], staticStringVal: 'test_string', dynStringVal: rd['rs']} ;
		  //db.log('PUT ' + test_bucket + '/' + em['key'] +' '+ JSON.stringify(payload));
		  //db.log('key ' + em['key'] + ' NOT found. Insert with no vClock.');
		  db.log('\n>-------d: '+ JSON.stringify(payload)
		   + '\n>-----key: -' + em['key'] + '\n>--vclock: ' + JSON.stringify(em['headers']['x-riak-vclock'])
		   + '\n>----etag: ' + JSON.stringify(em['headers']['etag']));
		  db.save(test_bucket, em['key'], payload )();
		
	    }//errback
	  
	  );//db.get

      db.log('call outer');
      //db.get(test_bucket_1, i )(mycallback,myerrback) ;
      

    } //for

    db.log('INSERTING KEYS INDEX');
	//db.log('PUT test_bucket_index, keyIndex, ' + JSON.stringify(test_bucket_index));
	db.save(test_bucket_stats, 'keyIndex', {keyIndex: test_bucket_index})();

};



// querying

var map = function(v, keydata, arg) {
  if (v.values) {
    var ret = [], a = Riak.mapValuesJson(v)[0]; 
    if ( (a.staticStringVal === arg.ssv) && ( a.dynStringVal.search(arg.dsv) != -1 ) ) {
	  a.key = v.key;
      ret.push(a);
    }
    return ret;
  } else {
    return [];
  }
};

var ssv = 'test_string',
    dsv = '9a',
    dsv2 = '88',
  query = {
    language: "javascript", source: map, arg:{ssv: ssv, dsv: dsv} //, keep: true
  };	
  query2 = {
	language: "javascript", source: map, arg:{ssv: ssv, dsv: dsv2} //, keep: true
	  };

var res_print = function(res) {
	//var r = JSON.parse(res);
  db.log('key: ' + res['key'] + ' dynStringVal: ' + res['dynStringVal']);//+ res[0]['numVal']
};

function doMR(q, style){
	

db.log("INITIATING M/R via " + style);

var start = (new Date).getTime();



db
  .map(query)
  //.map(query2)
  .run(test_bucket, {r: 1})(function(response) {
  //assert.equal(response.length, 2);
  db.log(response.length + ' matches, matching ' + ssv + ' and ' + dsv + ': ');
  //db.log(response);
  response.forEach(res_print);


  db.log("");

  var diff = (new Date).getTime() - start;

  db.log("time = " + diff);

  });
	


	
}




function colonBlow (storage, index) {
  
  db.get(index, 'keyIndex')(function(response) {
	//sys.puts(JSON.stringify(response));
	//sys.puts(response.keyIndex);
	//sys.puts(JSON.stringify(response).keys)
    response.keyIndex.forEach(function(key) {
      db.remove(storage, key)(function(d,m){
		//sys.log('REMOVED key: ' + m['key'])
		});
    //sys.puts(key);
    
     

    })

	db.remove(index, 'keyIndex')(function(){});

  })
  

};

//loadData();

//setTimeout(function () {sys.log('wait.');}, 60000);
doMR(query,'inputs:bucket');
//doMR(query,'key index')


//colonBlow(test_bucket,test_bucket_stats);

//setTimeout(function () {sys.log('wait.');}, 60000);