send progressing information

src/client/js/components/Admin/ImportData/GrowiZipImportForm.jsx

@@ -5,7 +5,8 @@ import * as toastr from 'toastr';
 
 import { createSubscribedElement } from '../../UnstatedUtils';
 import AppContainer from '../../../services/AppContainer';
-// import { toastSuccess, toastError } from '../../../util/apiNotification';
+import WebsocketContainer from '../../../services/WebsocketContainer';
+import { toastSuccess, toastError } from '../../../util/apiNotification';
 
 const GROUPS_PAGE = [
   'pages', 'revisions', 'tags', 'pagetagrelations',
@@ -25,6 +26,10 @@ class GrowiImportForm extends React.Component {
     super(props);
 
     this.initialState = {
+      isImporting: false,
+      isImported: false,
+      progressList: [],
+
       collectionNameToFileNameMap: {},
       selectedCollections: new Set(),
       schema: {
@@ -58,6 +63,36 @@ class GrowiImportForm extends React.Component {
     return Object.keys(this.state.collectionNameToFileNameMap);
   }
 
+  componentWillMount() {
+    this.setupWebsocketEventHandler();
+  }
+
+  setupWebsocketEventHandler() {
+    const socket = this.props.websocketContainer.getWebSocket();
+
+    // websocket event
+    // eslint-disable-next-line object-curly-newline
+    socket.on('admin:onProgressForImport', ({ currentCount, totalCount, progressList, appendedErrors }) => {
+      console.log(progressList);
+      console.log(appendedErrors);
+
+      this.setState({
+        isImporting: true,
+        progressList,
+      });
+    });
+
+    // websocket event
+    socket.on('admin:onTerminateForImport', () => {
+      this.setState({
+        isImporting: false,
+        isImported: true,
+      });
+
+      toastSuccess(undefined, 'Import process has terminated.');
+    });
+  }
+
   async toggleCheckbox(e) {
     const { target } = e;
     const { name, checked } = target;
@@ -340,6 +375,7 @@ class GrowiImportForm extends React.Component {
 GrowiImportForm.propTypes = {
   t: PropTypes.func.isRequired, // i18next
   appContainer: PropTypes.instanceOf(AppContainer).isRequired,
+  websocketContainer: PropTypes.instanceOf(WebsocketContainer).isRequired,
 
   fileName: PropTypes.string,
   innerFileStats: PropTypes.arrayOf(PropTypes.object).isRequired,
@@ -351,7 +387,7 @@ GrowiImportForm.propTypes = {
  * Wrapper component for using unstated
  */
 const GrowiImportFormWrapper = (props) => {
-  return createSubscribedElement(GrowiImportForm, props, [AppContainer]);
+  return createSubscribedElement(GrowiImportForm, props, [AppContainer, WebsocketContainer]);
 };
 
 export default withTranslation()(GrowiImportFormWrapper);

src/server/routes/apiv3/import.js

@@ -45,6 +45,16 @@ module.exports = (crowi) => {
   const adminRequired = require('../../middleware/admin-required')(crowi);
   const csrf = require('../../middleware/csrf')(crowi);
 
+  this.adminEvent = crowi.event('admin');
+
+  // setup event
+  this.adminEvent.on('onProgressForImport', (data) => {
+    crowi.getIo().sockets.emit('admin:onProgressForImport', data);
+  });
+  this.adminEvent.on('onTerminateForImport', (data) => {
+    crowi.getIo().sockets.emit('admin:onTerminateForImport', data);
+  });
+
   const uploads = multer({
     storage: multer.diskStorage({
       destination: (req, file, cb) => {
@@ -68,15 +78,13 @@ module.exports = (crowi) => {
    * all imported documents are overwriten by this value
    * each value can be any value or a function (_value, { _document, key, schema }) { return newValue }
    *
-   * @param {object} Model instance of mongoose model
+   * @param {string} collectionName MongoDB collection name
    * @param {object} req request object
    * @return {object} document to be persisted
    */
-  const overwriteParamsFn = async(Model, schema, req) => {
-    const collectionName = Model.collection.name;
-
+  const overwriteParamsFn = (collectionName, schema, req) => {
     /* eslint-disable no-case-declarations */
-    switch (Model.collection.collectionName) {
+    switch (collectionName) {
       case 'pages':
         // TODO: use schema and req to generate overwriteParams
         // e.g. { creator: schema.creator === 'me' ? ObjectId(req.user._id) : importService.keepOriginal }
@@ -101,7 +109,7 @@ module.exports = (crowi) => {
       //   return {};
       // ... add more cases
       default:
-        throw new Error(`cannot find a model for collection name "${collectionName}"`);
+        return {};
     }
     /* eslint-enable no-case-declarations */
   };
@@ -159,33 +167,55 @@ module.exports = (crowi) => {
     const { fileName, collections, schema } = req.body;
     const zipFile = importService.getFile(fileName);
 
-    // unzip
-    await importService.unzip(zipFile);
-    // eslint-disable-next-line no-unused-vars
-    const { meta, fileStats, innerFileStats } = await growiBridgeService.parseZipFile(zipFile);
+    /*
+     * unzip, parse
+     */
+    let meta = null;
+    let fileStatsToImport = null;
+    try {
+      // unzip
+      await importService.unzip(zipFile);
 
-    // filter innerFileStats
-    const filteredInnerFileStats = innerFileStats.filter(({ fileName, collectionName, size }) => {
-      return collections.includes(collectionName);
-    });
+      // eslint-disable-next-line no-unused-vars
+      const { meta: parsedMeta, fileStats, innerFileStats } = await growiBridgeService.parseZipFile(zipFile);
+      meta = parsedMeta;
 
+      // filter innerFileStats
+      fileStatsToImport = innerFileStats.filter(({ fileName, collectionName, size }) => {
+        return collections.includes(collectionName);
+      });
+    }
+    catch (err) {
+      logger.error(err);
+      return res.apiv3Err(err, 500);
+    }
+
+    /*
+     * validate with meta.json
+     */
     try {
-      // validate with meta.json
       importService.validate(meta);
+    }
+    catch (err) {
+      logger.error(err);
+      return res.apiv3Err(err);
+    }
 
-      filteredInnerFileStats.map(async({ fileName, collectionName, size }) => {
-        const Model = growiBridgeService.getModelFromCollectionName(collectionName);
-        const jsonFile = importService.getFile(fileName);
-
-        let overwriteParams;
-        if (overwriteParamsFn[collectionName] != null) {
-          // await in case overwriteParamsFn[collection] is a Promise
-          overwriteParams = await overwriteParamsFn(Model, schema[collectionName], req);
-        }
+    // generate maps to import
+    const jsonFileNamesMap = {};
+    const overwriteParamsMap = {};
+    fileStatsToImport.forEach(({ fileName, collectionName }) => {
+      jsonFileNamesMap[collectionName] = fileName;
 
-        importService.import(Model, jsonFile, overwriteParams);
-      });
+      const overwriteParams = overwriteParamsFn(collectionName, schema[collectionName], req);
+      overwriteParamsMap[collectionName] = overwriteParams;
+    });
 
+    /*
+     * import
+     */
+    try {
+      importService.import(collections, jsonFileNamesMap, overwriteParamsMap);
       return res.apiv3();
     }
     catch (err) {

src/server/service/import.js

@@ -10,6 +10,8 @@ const unzipper = require('unzipper');
 const { ObjectId } = require('mongoose').Types;
 
 const { createBatchStream } = require('../util/batch-stream');
+const CollectionProgressingStatus = require('../models/vo/collection-progressing-status');
+
 
 const BULK_IMPORT_SIZE = 100;
 
@@ -22,6 +24,8 @@ class ImportService {
     this.baseDir = path.join(crowi.tmpDir, 'imports');
     this.keepOriginal = this.keepOriginal.bind(this);
 
+    this.adminEvent = crowi.event('admin');
+
     // { pages: { _id: ..., path: ..., ...}, users: { _id: ..., username: ..., }, ... }
     this.convertMap = {};
     this.initConvertMap(crowi.models);
@@ -100,24 +104,57 @@ class ImportService {
     };
   }
 
+  /**
+   * import collections from json
+   *
+   * @param {string} collections MongoDB collection name
+   * @param {object} jsonFileNamesMap key: collection name, value: json file name
+   * @param {object} overwriteParamsMap key: collection name, value: overwrite each document with unrelated value. e.g. { creator: req.user }
+   */
+  async import(collections, jsonFileNamesMap, overwriteParamsMap) {
+    // init status object
+    this.currentProgressingStatus = new CollectionProgressingStatus(collections);
+
+    try {
+      const promises = collections.map((collectionName) => {
+        const jsonFileName = jsonFileNamesMap[collectionName];
+        const overwriteParams = overwriteParamsMap[collectionName];
+        return this.importCollection(collections, jsonFileName, overwriteParams);
+      });
+      await Promise.all(promises);
+    }
+    finally {
+      this.currentProgressingStatus = null;
+    }
+  }
+
   /**
    * import a collection from json
    *
    * @memberOf ImportService
-   * @param {object} Model instance of mongoose model
-   * @param {string} jsonFile absolute path to the jsonFile being imported
+   * @param {string} collectionName MongoDB collection name
+   * @param {string} jsonFileName json file name
    * @param {object} overwriteParams overwrite each document with unrelated value. e.g. { creator: req.user }
    * @return {insertedIds: Array.<string>, failedIds: Array.<string>}
    */
-  async import(Model, jsonFile, overwriteParams = {}) {
+  async importCollection(collectionName, jsonFileName, overwriteParams = {}) {
     // prepare functions invoked from custom streams
     const convertDocuments = this.convertDocuments.bind(this);
     const execUnorderedBulkOpSafely = this.execUnorderedBulkOpSafely.bind(this);
+    const emitProgressEvent = this.emitProgressEvent.bind(this);
+    const emitTerminateEvent = this.emitTerminateEvent.bind(this);
 
+    const Model = this.growiBridgeService.getModelFromCollectionName(collectionName);
+    const jsonFile = this.getFile(jsonFileName);
+    const collectionProgress = this.currentProgressingStatus.progressMap[collectionName];
+
+    // stream 1
     const readStream = fs.createReadStream(jsonFile, { encoding: this.growiBridgeService.getEncoding() });
 
+    // stream 2
     const jsonStream = JSONStream.parse('*');
 
+    // stream 3
     const convertStream = new Transform({
       objectMode: true,
       transform(doc, encoding, callback) {
@@ -127,8 +164,10 @@ class ImportService {
       },
     });
 
+    // stream 4
     const batchStream = createBatchStream(BULK_IMPORT_SIZE);
 
+    // stream 5
     const writeStream = new Writable({
       objectMode: true,
       async write(batch, encoding, callback) {
@@ -140,16 +179,18 @@ class ImportService {
         });
 
         // exec
-        // eslint-disable-next-line no-unused-vars
-        const { insertedIds, failedIds } = await execUnorderedBulkOpSafely(unorderedBulkOp);
+        const { insertedCount, errors } = await execUnorderedBulkOpSafely(unorderedBulkOp);
+        logger.debug(`Importing ${collectionName}. Inserted: ${insertedCount}. Failed: ${errors.length}.`);
+        collectionProgress.currentCount += insertedCount;
 
-        // TODO: emit event
+        emitProgressEvent(errors);
 
         callback();
       },
       final(callback) {
         // TODO: logger.info
-        // TODO: emit event
+
+        emitTerminateEvent();
 
         callback();
       },
@@ -167,6 +208,30 @@ class ImportService {
     fs.unlinkSync(jsonFile);
   }
 
+  /**
+   * emit progress event
+   * @param {object} appendedErrors key: collection name, value: array of error object
+   */
+  emitProgressEvent(appendedErrors) {
+    const { currentCount, totalCount, progressList } = this.currentProgressingStatus;
+    const data = {
+      currentCount,
+      totalCount,
+      progressList,
+      appendedErrors,
+    };
+
+    // send event (in progress in global)
+    this.adminEvent.emit('onProgressForImport', data);
+  }
+
+  /**
+   * emit terminate event
+   */
+  emitTerminateEvent() {
+    this.adminEvent.emit('onTerminateForImport');
+  }
+
   /**
    * extract a zip file
    *
@@ -204,38 +269,27 @@ class ImportService {
    *
    * @memberOf ImportService
    * @param {object} unorderedBulkOp result of Model.collection.initializeUnorderedBulkOp()
-   * @return {{nInserted: number, failed: Array.<string>}} number of docuemnts inserted and failed
+   * @return {object} e.g. { insertedCount: 10, errors: [...] }
    */
   async execUnorderedBulkOpSafely(unorderedBulkOp) {
-    // keep the number of documents inserted and failed for logger
-    let insertedIds = [];
-    let failedIds = [];
+    let insertedCount = 0;
+    let errors = [];
 
     // try catch to skip errors
     try {
       const log = await unorderedBulkOp.execute();
-      const _insertedIds = log.result.insertedIds.map(op => op._id);
-      insertedIds = [...insertedIds, ..._insertedIds];
+      insertedCount = log.result.insertedIds.length;
     }
     catch (err) {
-      const collectionName = unorderedBulkOp.s.namespace;
-
-      for (const error of err.result.result.writeErrors) {
-        logger.error(`${collectionName}: ${error.errmsg}`);
-      }
-
-      const _failedIds = err.result.result.writeErrors.map(err => err.err.op._id);
-      const _insertedIds = err.result.result.insertedIds.filter(op => !_failedIds.includes(op._id)).map(op => op._id);
-
-      failedIds = [...failedIds, ..._failedIds];
-      insertedIds = [...insertedIds, ..._insertedIds];
+      errors = err.writeErrors.map((err) => {
+        const moreDetailErr = err.err;
+        return { _id: moreDetailErr.op._id, message: err.errmsg };
+      });
     }
 
-    logger.debug(`Importing ${unorderedBulkOp.s.collection.s.name}. Inserted: ${insertedIds.length}. Failed: ${failedIds.length}.`);
-
     return {
-      insertedIds,
-      failedIds,
+      insertedCount,
+      errors,
     };
   }