使用 RxJS 和 Node 调整图片大小

Question

我是 RxJS 的新手，正在尝试这个（看似）简单的任务，但我就是想不通。

我要：
1. 从文件中读取图像
2. 将该图像转换为几个较小的图像
3. 保存所有图片到文件

我已将 fs.readFile 和 fs.writeFile 转换为可观察值。

const readFile$ = Rx.Observable.bindNodeCallback(fs.readFile);
const writeFile$ = Rx.Observable.bindNodeCallback(fs.writeFile);

我做了一个图片数组管道。

var pictureSizes = [
  {width: 100, size: 'thumbnail', suffix: '_t'},
  {width: 300, size: 'small', suffix: '_s'},
  {width: 600, size: 'medium', suffix: '_m'},
  {width: 1000, size: 'large', suffix: '_l'}
];

然后我使用图形魔法制作了一个 resizeImage$ 函数

function resizeImage$(picture, data) {
  return Rx.Observable.create(observer => {
      gm(data)
        .resize(picture.width)
        .toBuffer('jpg', function(err, buffer) {
          if (err) {
            console.log(err);
            observer.error(err);
          } else {
            observer.next(buffer);
            observer.complete();
          }
        });
  })
}

我认为（希望）以上是可以的。我不知道如何链接我的运算符。

  readFile$('./largeimage.jpg')
    .mergeMap(data => pictureSizes.map(picture => resizeImage$(picture, data)))
    .flatMap(picture => writeFile$('./testImages/resized.jpg', picture))
    .subscribe(
    (x) => console.log('Next', x),
    (e) => console.log('Error', e),
    (c) => console.log('Complete',c )
  )

上述损坏的数据到 jpeg 文件。（并重写该文件，因为我无法弄清楚如何将 pictureSizes.suffix 放入输出文件名中。

任何帮助！谢谢。

更新

我让它工作了，但我知道这种奇怪的多重订阅是一种可怕的反模式。主订阅在调整图像大小之前完成。我感觉这是一个 hot/cold 问题，但我不知道如何解决它。这是我现在的工作代码..

const pictureSizes = [
  {width: 100, size: 'thumbnail', suffix: '_t'},
  {width: 300, size: 'small', suffix: '_s'},
  {width: 600, size: 'medium', suffix: '_m'},
  {width: 1000, size: 'large', suffix: '_l'}
];

const image = 'truck.jpg';

function resizeImage$(binary, pictureSize) {
  return new Rx.Observable(observer => {
      gm(binary)
        .resize(pictureSize.width)
        .toBuffer('jpg', function(err, buffer) {
          console.log('BUFFER');
          if (err) {
            console.log(err);
            observer.error(err);
          } else {
            observer.next({binary: buffer, pictureSize: pictureSize});
            observer.complete('done');
          }
        });
  }).subscribe(
    (resizedImage) => {
      console.log(resizedImage);
      const binary = resizedImage.binary;
      const pictureSize = resizedImage.pictureSize;
      const fileName = image.split('.')[0];
      const fileExtension = image.split('.')[1];
      fs.writeFile(`./testImages/${fileName}${pictureSize.suffix}.${fileExtension}`, binary);
    })
}
  var readFile$ = new  Rx.Observable.bindNodeCallback(fs.readFile);
  readFile$(`./${image}`)
  .zip(Rx.Observable.of(pictureSizes), (binary, sizes) =>
        Rx.Observable.of({ binary: binary, sizes: sizes }))
  .mergeMap(x => x.value.sizes.map(pictureSize => 
       resizeImage$(x.value.binary, pictureSize)))
  .subscribe()

Answer 1

如果有人感兴趣，我有一个答案。如果有人想进一步重构它，请这样做。

剩余问题... 1. 写入文件是否应该作为可观察的来捕获错误？
2. 我不确定为什么需要 mergeAll。

var pictureSizes = [
  {width: 100, size: 'thumbnail', suffix: '_t'},
  {width: 300, size: 'small', suffix: '_s'},
  {width: 600, size: 'medium', suffix: '_m'},
  {width: 1000, size: 'large', suffix: '_l'}
];

function scaleImage$(binary, pictureSize) {
  return new Rx.Observable(observer => {
      gm(binary)
        .resize(pictureSize.width)
        .toBuffer('jpg', function(err, buffer) {
          if (err) {
            observer.error(err);
          } else {
            observer.next({ binary: buffer, pictureSize: pictureSize });
            observer.complete();
          }
        });
  })
}

function writeFile(binary, pictureSize, image) {
  const fileName = image.split('.')[0];
  const fileExtension = image.split('.')[1];
  fs.writeFile(`./resized/${fileName}${pictureSize.suffix}.${fileExtension}`, binary);
}

function resizeImage(imagePath) {
  var readFile$ = new  Rx.Observable.bindNodeCallback(fs.readFile);
  readFile$(imagePath)
    .combineLatest(Rx.Observable.of(pictureSizes),(binary,y) => y.map(pictureSize => Object.assign({}, {binary, pictureSize} )))
    .mergeMap(arr => arr.map(obj => scaleImage$(obj.binary, obj.pictureSize)))
    .mergeAll()
    .subscribe((obj) => writeFile(obj.binary, obj.pictureSize, image))

}

如果您想要同步行为（缩放图像 1 -> 写入图像 1 -> 缩放图像 2 ...），请使用 concatMap 和 concatAll。

Answer 2

根据你的例子，我认为你可以进一步简化你的回答：

var pictureSizes = [
  {width: 100, size: 'thumbnail', suffix: '_t'},
  {width: 300, size: 'small', suffix: '_s'},
  {width: 600, size: 'medium', suffix: '_m'},
  {width: 1000, size: 'large', suffix: '_l'}
];

const scaler$ = Rx.Observable.bindNodeCallback((binary, size, callback) => {
  gm(binary)
   .resize(size.width)
   .toBuffer('jpg', callback);
});

const readFile$ = Rx.Observable.bindNodeCallback(fs.readFile);
const writeFile$ = Rx.Observable.bindNodeCallback(fs.writeFile);

function scaleImage$(sizes) {
  const scales = Rx.Observable.from(sizes);

  return source => 
    source.flatMap(binary => 
      scales.flatMap(
        size => scaler$(binary, size),
        (pictureSize, binary) => ({pictureSize, binary})
      )
    );
}

function resize(imagePath, sizes) {
  return readFile$(imagePath)
    .let(scaleImage$(sizes))
    .flatMap(result => {
      const {pictureSize, binary} = result;
      const [name, ext] = image.split('.');
      return writeFile$(`./resized/${name}${pictureSize.suffix}.${ext}`, binary);
    });
}

使用：

resize(imagePath, pictureSizes)
  .subscribe();

使用 RxJS 和 Node 调整图片大小

Resizing Pictures with RxJS and Node

node.js

graphicsmagick

rxjs

rxjs5