Как создать только одну подписку из этих двух наблюдаемых?

Question

У меня есть функция, которая читает данные из читаемого потока, а затем сохраняет результат в 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)
                );

            }
        );

это работает очень хорошо, но я хотел бы знать, есть ли способ сделать это более чистым способом, используя только одного подписчика.

Возможно ли это?

Answer 1

Сначала 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 =>{})

Приведенный выше подход может быть проще и следует хорошим принципам функционального программирования, здесь вам не нужно использовать свойство jsArr.

Answer 2

Полагаю, вы могли бы сделать это так, но по понятным причинам я не проверял это.

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