I am writing a custom operator to load a csv file and emit each line as data. This operator is supposed to work like the of
operator, which is a static function to create observable. I follow the instruction of operator creation and add the operator function directly to Observable prototype.
All following code is written in JavaScript ES6.
My source code is this
import { Observable } from 'rxjs';
import { createInterface } from 'readline';
import { createReadStream } from 'fs';
function fromCSV(path, options) {
return Observable.create((subscriber) => {
let readStream = createReadStream(path);
let reader = createInterface({ input: readStream });
let isHeader = true;
let columns;
reader.on('line', (line) => {
if (isHeader) {
columns = line.split(',');
isHeader = false;
} else {
let values = line.split(',');
let valueObject = {};
for (let i = 0, n = columns.length; i < n; i++) {
valueObject[columns[i]] = values[i] || undefined;
}
subscriber.next(valueObject);
}
});
reader.on('close', () => subscriber.complete());
readStream.on('error', (error) => subscriber.error(error));
});
}
Observable.prototype.fromCSV = fromCSV;
The operator function looks totally correct, but when I try to use this operator like
import { Observable } from 'rxjs';
import './rx-from-csv';
Observable.fromCSV(testCSV)
.subscribe((row) => {
console.log(row);
});
It throws an error TypeError: _rxjs.Observable.fromCSV is not a function
. So the function binding fails and I have no idea why it happens :-( Any help is appreciated.
This particularly confuses me because I have successfully done a similar operator binding for another custom csv operator.