Синхронизация потоковых выходов

Question

Я работаю над загрузкой изображений разных размеров в хранилище amazon s3.Всякий раз, когда я загружаю изображение, я создаю три отдельных читаемых потока изображения и передаю его по конвейеру своим функциям изменения размера и функции загрузки s3.

Каждый из потоков загружает изображение разного размера в облако s3.Я хотел бы синхронизировать выходные данные потоков, чтобы можно было отправлять URL-адреса загруженных фотографий после завершения загрузки во всех трех потоках.

Я не могу знать, когда все мои потоки имеютзакончился.Или как мне дождаться окончания всех моих потоков и вернуть URL всех загруженных изображений?

Вот мой код

resize.js

import fs from 'fs';
import sharp from 'sharp';

export const resizeThumbnail = function (path, format) {
  const readStream = fs.createReadStream(path);
  let transform = sharp();
  if (format) {
    transform = transform.toFormat(format);
  }
  transform = transform.resize(100);
  return readStream.pipe(transform);
};

export const resizeLowRes = function (path, format) {
  const readStream = fs.createReadStream(path);
  let transform = sharp();
  if (format) {
    transform = transform.toFormat(format);
  }
  transform = transform.resize(500);
  return readStream.pipe(transform);
};

export const resizeHighRes = function (path, format) {
  const readStream = fs.createReadStream(path);
  let transform = sharp();
  if (format) {
    transform = transform.toFormat(format);
  }
  transform = transform.resize(1080);
  return readStream.pipe(transform);
};

resizer.js

import { resizeThumbnail, resizeLowRes, resizeHighRes } from './resize';
import awsConfig from '../config/awsConfig';
import { uploadFromStream } from './amazonS3';

// receives a list of images
export const resizer = (images, id, callback) => {
  const imageList = images.split(',');
  const store = [];
  imageList.forEach((imagePath, i) => {
    sizeCreator(imagePath, id, i, (err, localstore) => {
      if (!err) {
        store.push(localstore);
        console.log(store);
      }
      callback(err);
    });
  });
};

const sizeCreator = (imagePath, id, i, callback) => {
  const localstore = {};
  resizeThumbnail(imagePath).pipe(uploadFromStream(awsConfig, id, i, 'thumbnail', (err, data) => {
    localstore.thumbnail = {
      url: data.Location,
      filename: data.key,
    };
  }));
  resizeLowRes(imagePath).pipe(uploadFromStream(awsConfig, id, i, 'lowres', (err, data) => {
    localstore.lowres = {
      url: data.Location,
      filename: data.key,
    };
  }));
  resizeHighRes(imagePath).pipe(uploadFromStream(awsConfig, id, i, 'highres', (err, data) => {
    localstore.highres = {
      url: data.Location,
      filename: data.key,
    };
    callback(err, localstore);
  }));
};

amazonS3.js

import stream from 'stream';

export const uploadFromStream = (s3, id, index, res, callback) => {
  const pass = new stream.PassThrough();
  const params = {
    Bucket: 'Bucket',
    Key: `${id}_${res}_${index}`,
    Body: pass,
    ContentType: 'image/jpeg',
  };
  s3.upload(params, (err, data) => {
    callback(err, data);
  });
  return pass;
};

Answer 1

Вы можете считать начатый поток и обновить счетчик на событии завершения потока.Когда последний поток закончен, счетчик должен иметь значение 0. Затем вы можете инициировать другое событие, которое отправляет URL.