Skip to content

Commit 1450dc8

Browse files
zubairovzubairov
committed
Tests almost work
1 parent 923ef22 commit 1450dc8

File tree

5 files changed

+116
-263
lines changed

5 files changed

+116
-263
lines changed

‎lib/read.js‎

Lines changed: 50 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -1,86 +1,66 @@
11
var _ = require('underscore');
22
var util = require('util');
33
var csv = require('csv');
4-
var messages = require('../messages.js');
4+
var messages = require('elasticio-node').messages;
55
var moment = require('moment');
6-
var s3client = require('s3').getClient(process.env.S3_BUCKET);
7-
var HeartBeatStream = require('../heartbeat-stream.js').HeartBeatStream;
6+
var debug = require('debug')('csv');
7+
var request = require('request');
88

9-
exports.process = function (msg, cfg) {
10-
var attachments = msg.attachments, that = this;
9+
function processRead(msg, cfg) {
10+
var csvURL = cfg.url;
11+
var that = this;
12+
var index = 0;
13+
var separator = cfg.reader ? cfg.reader.separator || "," : ",";
14+
var startRow = cfg.reader ? cfg.reader.startRow || 0 : 0;
1115

12-
if (!attachments || Object.keys(attachments).length == 0) {
13-
this.info('No attachments found in incoming message, CSV Read component will terminate this flow');
14-
return this.emit('end');
15-
}
16-
var csvs = Object.keys(attachments).filter(function (name) {
17-
var att = attachments[name];
18-
var type = att["content-type"];
1916

20-
if (!att.s3) {
21-
console.log("Only S3 attachments supported yet");
22-
return false;
17+
if (!csvURL || csvURL.length == 0) {
18+
console.error('URL of the CSV is missing');
19+
that.emit('error', 'URL of the CSV is missing');
20+
return that.emit('end');
21+
}
22+
var parser = csv.parse({delimiter: separator});
23+
24+
parser.on('readable', function() {
25+
while(record = parser.read()){
26+
debug('Have got a row #%s', index);
27+
if (index >= startRow) {
28+
var msg = createRowMessage(record, cfg.reader.columns);
29+
that.emit('data', msg);
30+
} else {
31+
debug('Row #%s is skipped based on configuration', index);
32+
}
33+
index++;
2334
}
24-
25-
return true;
2635
});
2736

28-
if (csvs.length === 0) {
29-
this.info('Can not find suitable attachments to parse');
30-
return this.emit('end');
31-
}
32-
33-
this.debug('Will parse following CSV files [%s]', csvs.join(','));
34-
35-
csvs.forEach(function(name) {
36-
parseAttachment(that, s3client.getEncrypted(attachments[name].s3), cfg);
37+
parser.on('finish', function fireEnd() {
38+
debug('Number of lines: ' + index);
39+
that.emit('end');
3740
});
38-
};
39-
40-
var parseAttachment = function (that, attachmentPromise, cfg) {
41-
42-
var separator = cfg.reader.separator || ",";
4341

44-
var startRow = cfg.reader.startRow || 0;
45-
46-
attachmentPromise
47-
.then(function(res) {
48-
49-
new HeartBeatStream()
50-
.on('heartbeat', function() {
51-
that.emit('heartbeat');
52-
}).start(res);
53-
54-
csv().from(res,{ delimiter: separator})
55-
.on('record', function (row, index) {
56-
57-
if (index >= startRow) {
58-
59-
var msg = createRowMessage(row, cfg.reader.columns);
60-
61-
that.emit('data', msg);
62-
}
63-
})
64-
.on('end', function (count) {
65-
that.debug('Number of lines: ' + count);
42+
parser.on('error', function emitError(error) {
43+
debug('Error reported by CSV read', error);
44+
that.emit('error', error);
45+
});
6646

67-
that.emit('end');
68-
})
69-
.on('error', function (error) {
70-
console.log(error.stack);
71-
});
47+
debug('Sending GET request to url=%s', csvURL);
48+
request.get(csvURL)
49+
.on('response', function (response) {
50+
debug('Have got response status=%s headers=%j', response.statusCode, response.headers);
51+
if (response.statusCode != 200) {
52+
that.emit('error', 'Unexpected response code code=' + response.statusCode);
53+
throw Error('Unexpected response code code=' + response.statusCode);
54+
}
7255
})
73-
.fail(function(e) {
74-
console.log(e);
75-
})
76-
.done();
77-
};
56+
.pipe(parser);
57+
}
7858

7959
var createRowMessage = function (row, columns) {
8060

8161
var body = {};
8262

83-
_.each(row, function(value, index) {
63+
_.each(row, function (value, index) {
8464
var col = columns[index];
8565

8666
if (col) {
@@ -96,21 +76,21 @@ var createRowMessage = function (row, columns) {
9676
var formatValue = function (value, col) {
9777
var type = col.type || "string";
9878

99-
var formatter = formatters[type] || function(value, col) {
100-
return value;
101-
};
79+
var formatter = formatters[type] || function (value, col) {
80+
return value;
81+
};
10282

10383
return formatter(value, col);
10484
};
10585

10686
var formatters = {};
10787

108-
formatters["date"] = function(value, col) {
88+
formatters["date"] = function (value, col) {
10989

11090
return moment(value, col.format).toDate();
11191
};
11292

113-
formatters["number"] = function(value, col) {
93+
formatters["number"] = function (value, col) {
11494

11595
if (col.format === "dec_comma") {
11696
//123.456.78,9 => 12345678.9
@@ -125,3 +105,4 @@ formatters["number"] = function(value, col) {
125105
return parseFloat(value);
126106
};
127107

108+
exports.process = processRead;

‎package.json‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,16 +21,23 @@
2121
"url": "https://github.com/elasticio/csv-component/issues"
2222
},
2323
"devDependencies": {
24+
"code-quality-js": "git+https://github.com/elasticio/code-quality-js.git#master",
25+
"git-guppy": "~1.0.0",
2426
"gulp": "^3.8.11",
2527
"gulp-jasmine": "^2.0.1",
2628
"gulp-jscs": "~1.6.0",
27-
"git-guppy": "~1.0.0",
28-
"code-quality-js": "git+https://github.com/elasticio/code-quality-js.git#master"
29+
"nock": "^2.17.0"
2930
},
3031
"homepage": "https://github.com/elasticio/csv-component#readme",
3132
"dependencies": {
3233
"csv": "^0.4.6",
3334
"debug": "^2.2.0",
34-
"elasticio-sailor-nodejs": "^1.0.1"
35+
"elasticio-node": "0.0.7",
36+
"elasticio-sailor-nodejs": "^1.0.1",
37+
"moment": "^2.10.6",
38+
"node-uuid": "^1.4.3",
39+
"q": "^1.4.1",
40+
"request": "^2.65.0",
41+
"underscore": "^1.8.3"
3542
}
3643
}

‎spec/read.integration.spec.js‎

Lines changed: 11 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,34 +1,18 @@
1-
describe('CSV Read component integration tests', function () {
2-
process.env.S3_SECRET = process.env.S3_SECRET || 'my-secret';
3-
process.env.S3_KEY = process.env.S3_KEY || 'my-key';
4-
process.env.S3_REGION = process.env.S3_REGION || 'eu-west-1';
5-
process.env.S3_BUCKET = process.env.S3_BUCKET || 'my-bucket';
6-
process.env.S3_CRYPTO_PASSWORD = 'all-attachment-password';
7-
process.env.S3_CRYPTO_ALGORITHM = 'aes-256-cbc';
1+
var nock = require('nock');
82

9-
var csv = require('../../../lib/components/csv/read.js');
3+
describe('CSV Read component integration tests', function () {
4+
var csv = require('../lib/read.js');
105
var runTest = require('./testrunner.js').runTest;
116
var fs = require('fs');
127
var Q = require('q');
13-
var s3 = require('s3').getClient(process.env.S3_BUCKET);
148

15-
it('winter12.csv', function () {
16-
spyOn(s3, 'getEncrypted').andCallFake(function() {
17-
return Q(fs.createReadStream(__dirname + '/test/winter12.csv'));
18-
});
19-
20-
var msg = {
21-
id:12345,
22-
attachments:{
23-
foo : {
24-
s3 : 'http://loremipsum.fubar',
25-
'content-type' : 'text/csv'
26-
}
27-
},
28-
body:{}
29-
};
9+
nock('http://test.env.mock')
10+
.get('/winter12.csv')
11+
.replyWithFile(200, __dirname + '/test/winter12.csv');
3012

3113

14+
it('winter12.csv', function (done) {
15+
var msg = {};
3216

3317
var cfg = {
3418
reader: {
@@ -124,7 +108,8 @@ describe('CSV Read component integration tests', function () {
124108
type : 'string'
125109
}
126110
]
127-
}
111+
},
112+
url: "http://test.env.mock/winter12.csv"
128113
};
129114

130115
runTest(csv.process, msg, cfg, function(runner) {
@@ -180,6 +165,7 @@ describe('CSV Read component integration tests', function () {
180165

181166
expect(runner.errors.length).toEqual(0);
182167
expect(runner.snapshot).toBeUndefined();
168+
done();
183169
});
184170
});
185171

0 commit comments

Comments
 (0)