riak書き込みテスト

gistfile1.txt

"use strict";
var os = require('os');

var fs = require("fs");
var readline = require('readline');
var RiakClient = require("riak");
var async = require('async');


var parse = function(line){
    var head = line.split(",", 2);
    var jsondata = JSON.parse(line.replace(head.join(",")+",", ""))
    if(jsondata.protocol_name === null){
        jsondata.protocol_name = "TASK";
    }
    return [head[0], jsondata];
}
var fileReadLine = function( filepath ){
    var rs = fs.ReadStream(filepath);
    var rl = readline.createInterface({'input': rs, 'output': {}});
    return rl;
}
var client = new RiakClient(["127.0.0.1:8098"], "client_id", "pool_name");

client.debug_mode = false;

client.on("metrics", function (type, key, val) {
    console.log("metric: " + type + ", " + key + "=" + val);
});


var filepath = "/disk1/info2013-07-12-13409.log";

var list = [];

var rl = fileReadLine(filepath).on('line', function (line) {
    var obj = parse(line.trim());

    var key=obj[1].log.chara_id.low_;

    list.push({key:key, val:obj});

    rl.pause();

}).on('close', function(){
    console.log("end");
});
rl.pause();

var update = function(){
    setTimeout(function(){
        exec(list, function(){
            update();
        });
    },0);
}

var exec = function(list, callback){
    if(list.length === 0){
        rl.resume();
        callback();
        return;
    }
    var bucket="users_log1";
            
    var w = list.shift();
    client.append(bucket, w.key, JSON.stringify(w.val), {}, function(err, obj){
        if(err) throw err;
        callback();
    });
}
update();