<commit_before>#!/usr/bin/env node

const fs = require('fs');
const LimitedParallelStream = require('./src/modules/limited_parallel_stream').LimitedParallelStream;
const byline = require('byline'); //@todo replace this module
const riak = require('./riak');
const Riak = require('basho-riak-client');
const program = require('commander');

program
	.version('0.0.1')
	.usage('[options] bucketName')
	.option('-H, --host [host]', 'specify the host (default: http://127.0.0.1)', 'http://127.0.0.1')
	.option('-p, --port [port]', 'specify the post (default: 8098)', 8098)
	.option('-c, --concurrency [concurrency]','specify the concurrency (default: 100)', 100)
	.parse(process.argv);

if(!program.args.length) {
	console.log('\nBucket is required!');
	program.help();
}

const bucket = program.args[0];

let stream = fs.createReadStream(__dirname + '/dump');
stream = byline.createStream(stream);

console.info('Import started');

riak.onReady(program.host, program.port, (err, client) =>{
	if(err) {
		throw err
	}

	stream
		.pipe(new LimitedParallelStream(program.concurrency, function(row, enc, done){
			row = row.toString().split('\t');
			const value = new Riak.Commands.KV.RiakObject();
			value.setContentType(row[2]);
			value.setValue(row[1]);

			client.storeValue({
				bucket: bucket,
				key: row[0],
				value: value
			}, (err) =>{
				if(err) {<commit_msg>Add -f option to import<commit_after>#!/usr/bin/env node

const fs = require('fs');
const LimitedParallelStream = require('./src/modules/limited_parallel_stream').LimitedParallelStream;
const byline = require('byline'); //@todo replace this module
const riak = require('./riak');
const Riak = require('basho-riak-client');
const program = require('commander');

program
	.version('0.0.1')
	.usage('[options] bucketName')
	.option('-H, --host [host]', 'specify the host (default: http://127.0.0.1)', 'http://127.0.0.1')
	.option('-p, --port [port]', 'specify the post (default: 8098)', 8098)
	.option('-c, --concurrency [concurrency]','specify the concurrency (default: 100)', 100)
	.option('-f, --file [file]', 'specify the folder of dump (default: __dirname + /dump)', __dirname + '/dump')
	.parse(process.argv);

if(!program.args.length) {
	console.log('\nBucket is required!');
	program.help();
}

const bucket = program.args[0];

let stream = fs.createReadStream(program.file);
stream = byline.createStream(stream);

console.info('Import started');

riak.onReady(program.host, program.port, (err, client) =>{
	if(err) {
		throw err
	}

	stream
		.pipe(new LimitedParallelStream(program.concurrency, function(row, enc, done){
			row = row.toString().split('\t');
			const value = new Riak.Commands.KV.RiakObject();
			value.setContentType(row[2]);
			value.setValue(row[1]);

			client.storeValue({
				bucket: bucket,
				key: row[0],
				value: value
			}, (err) =>{
				if(err) {
{'fraction_non_alphanumeric': 0.19180959047952398, 'fraction_numerical': 0.023101155057752887, 'mean_word_length': 4.560311284046692, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6042704', 'n_tokens_mistral': 1071, 'n_tokens_neox': 988, 'n_words': 218}