CODEWITHSUNDEEP
javascriptrxjs
ste9206
ste9206

Reputation: 1892

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?

Upvotes: 1

Views: 50

Answers (2)

siva636
siva636

Reputation: 16451

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

 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.

Upvotes: 0

martin
martin

Reputation: 96979

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)
  )

Upvotes: 1

Related Questions