How to create only one subscription from this two observables?

I have a function that reads data from a readable stream and then saves the result in Parse:

const { from } = require('rxjs');
const Observable = require('rxjs').Observable;
const jsArr = ;

fromStream(fs.createReadStream(req.file.path, 'utf-8').pipe(csv()))
.map((data) => data.USERNAME)
.subscribe(
(data) => jsArr.push(data),
(error) => console.log(error),
() => {
const CSVData = Parse.Object.extend('CSVData');
const csv = new CSVData();
csv.set('CSV',jsArr);

from (csv.save())
.subscribe(
() => res.json(serialize({type:'success'})),
(error) => console.log(error)
);

}
);


it works really well, but I'd like to know if there is a way to do it in a more cleaner way, using only one subscriber.

Is it possible?

2 Answers
2

I guess you could do it like this but for obvious reasons I haven't tested it.

fromStream(fs.createReadStream(req.file.path, 'utf-8').pipe(csv()))
.map((data) => data.USERNAME)
.toArray(),
.concatMap(jsArr => {
const CSVData = Parse.Object.extend('CSVData');
const csv = new CSVData();
csv.set('CSV',jsArr);

return from(csv.save());
})
.subscribe(
() => res.json(serialize({type:'success'})),
(error) => console.log(error)
)


First reduce and get the result to save, and save the result:

reduce

fromStream(fs.createReadStream(req.file.path, 'utf-8').pipe(csv()) )
.map((data) => data.USERNAME)
.reduce((acc, curr) => acc.push(curr), )
.switchMap(resultArray => {
const CSVData = Parse.Object.extend('CSVData');
const csv = new CSVData();
csv.set('CSV',resultArray);
return from (csv.save());

}).subscribe(x =>{})


The above approach may be easier and follows good functional programming concepts, here you do not need to use the property jsArr.

jsArr

By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.