Create streams only after Promise.all()

lib/resources/odata.js

@@ -64,12 +64,11 @@ module.exports = (service, endpoint) => {
  const options = QueryOptions.fromODataRequest(params, query);
  return Promise.all([
  Forms.getFields(form.def.id),
- Submissions.streamForExport(form.id, draft, undefined, options),
  ((params.table === 'Submissions') && options.hasPaging())
  ? Submissions.countByFormId(form.id, draft, options) : resolve(null)
  ])
- .then(([ fields, stream, count ]) =>
- json(rowStreamToOData(fields, params.table, env.domain, originalUrl, query, stream, count)));
+ .then(([ fields, count ]) => Submissions.streamForExport(form.id, draft, undefined, options)
+ .then((stream) => json(rowStreamToOData(fields, params.table, env.domain, originalUrl, query, stream, count))));
  })));
  };
 

lib/resources/submissions.js

@@ -7,7 +7,7 @@
 // including this file, may be copied, modified, propagated, or distributed
 // except according to the terms contained in the LICENSE file.
 
-const { always, identity } = require('ramda');
+const { always, identity, call } = require('ramda');
 const multer = require('multer');
 const sanitize = require('sanitize-filename');
 const { Blob, Form, Submission } = require('../model/frames');
@@ -269,20 +269,20 @@ module.exports = (service, endpoint) => {
  const options = QueryOptions.fromSubmissionCsvRequest(query);
  return Promise.all([
  (options.deletedFields === true) ? Forms.getMergedFields(form.id) : Forms.getFields(form.def.id),
- Submissions.streamForExport(form.id, draft, keys, options),
  (options.splitSelectMultiples !== true) ? null : Submissions.getSelectMultipleValuesForExport(form.id, draft, options),
- SubmissionAttachments.streamForExport(form.id, draft, keys, options),
- ClientAudits.streamForExport(form.id, draft, keys, options),
  draft ? null : Audits.log(auth.actor, 'form.submission.export', form)
- ]).then(([ fields, rows, selectValues, attachments, clientAudits ]) => {
+ ]).then(([ fields, selectValues ]) => {
  const filename = sanitize(form.xmlFormId);
  response.append('Content-Disposition', contentDisposition(`${filename}.zip`));
  response.append('Content-Type', 'application/zip');
  return zipStreamFromParts(
  // TODO: not 100% sure that these streams close right on crash.
- streamBriefcaseCsvs(rows, fields, form.xmlFormId, selectValues, decryptor, false, options),
- streamAttachments(attachments, decryptor),
- streamClientAudits(clientAudits, form, decryptor)
+ () => Submissions.streamForExport(form.id, draft, keys, options)
+ .then((rows) => streamBriefcaseCsvs(rows, fields, form.xmlFormId, selectValues, decryptor, false, options)),
+ () => SubmissionAttachments.streamForExport(form.id, draft, keys, options)
+ .then((attachments) => streamAttachments(attachments, decryptor)),
+ () => ClientAudits.streamForExport(form.id, draft, keys, options)
+ .then((clientAudits) => streamClientAudits(clientAudits, form, decryptor))
  );
  });
  }));
@@ -295,18 +295,18 @@ module.exports = (service, endpoint) => {
  const options = QueryOptions.fromSubmissionCsvRequest(query);
  return Promise.all([
  (options.deletedFields === true) ? Forms.getMergedFields(form.id) : Forms.getFields(form.def.id),
- Submissions.streamForExport(form.id, draft, Object.keys(passphrases), options),
  (options.splitSelectMultiples !== true) ? null : Submissions.getSelectMultipleValuesForExport(form.id, draft, options),
  Keys.getDecryptor(passphrases),
  draft ? null : Audits.log(auth.actor, 'form.submission.export', form)
  ])
- .then(([ fields, rows, selectValues, decryptor ]) => {
+ .then(([ fields, selectValues, decryptor ]) => {
  const filename = sanitize(form.xmlFormId);
  const extension = (rootOnly === true) ? 'csv' : 'csv.zip';
  response.append('Content-Disposition', contentDisposition(`${filename}.${extension}`));
  response.append('Content-Type', (rootOnly === true) ? 'text/csv' : 'application/zip');
- const envelope = (rootOnly === true) ? identity : zipStreamFromParts;
- return envelope(streamBriefcaseCsvs(rows, fields, form.xmlFormId, selectValues, decryptor, rootOnly, options));
+ const envelope = (rootOnly === true) ? call : zipStreamFromParts;
+ return envelope(() => Submissions.streamForExport(form.id, draft, Object.keys(passphrases), options)
+ .then((rows) => streamBriefcaseCsvs(rows, fields, form.xmlFormId, selectValues, decryptor, rootOnly, options)));
  });
  });
 

lib/util/zip.js

@@ -14,6 +14,7 @@
 const { Readable } = require('stream');
 const { PartialPipe } = require('./stream');
 const archiver = require('archiver');
+const { resolve } = require('./promise');
 
 // Returns an object that can add files to an archive, without having that archive
 // object directly nor knowing what else is going into it. Call append() to add a
@@ -36,8 +37,7 @@ const zipPart = () => {
 // if the final component in the pipeline emitted the error, archiver would then
 // emit it again, but if it was an intermediate component archiver wouldn't know
 // about it. by manually aborting, we always emit the error and archiver never does.
-const zipStreamFromParts = (...zipParts) => {
- let completed = 0;
+const zipStreamFromParts = (...zipPartFns) => {
  const resultStream = archiver('zip', { zlib: { level: 9 } });
 
  // track requested callbacks and call them when they are fully added to the zip.
@@ -47,29 +47,35 @@ const zipStreamFromParts = (...zipParts) => {
  if (cb != null) cb();
  });
 
- for (const part of zipParts) {
- part.stream.on('data', ({ stream, options, cb }) => {
- const s = (stream instanceof PartialPipe)
-  ? stream.pipeline((err) => { resultStream.emit('error', err); resultStream.abort(); })
-  : stream;
+ const next = () => {
+ if (zipPartFns.length === 0) {
+ resultStream.finalize();
+ return;
+ }
 
- if (cb == null) {
- resultStream.append(s, options);
- } else {
- // using the String object will result in still an empty comment, but allows
- // separate instance equality check when the entry is recorded.
- const sentinel = new String(); // eslint-disable-line no-new-wrappers
- callbacks.set(sentinel, cb);
- resultStream.append(s, Object.assign(options, { comment: sentinel }));
- }
- });
- part.stream.on('error', (err) => { resultStream.emit('error', err); });
- part.stream.on('end', () => { // eslint-disable-line no-loop-func
- completed += 1;
- if (completed === zipParts.length)
- resultStream.finalize();
- });
- }
+ resolve(zipPartFns.shift()())
+ .then((part) => {
+ part.stream.on('data', ({ stream, options, cb }) => {
+ const s = (stream instanceof PartialPipe)
+ ? stream.pipeline((err) => { resultStream.emit('error', err); resultStream.abort(); })
+ : stream;
+
+ if (cb == null) {
+ resultStream.append(s, options);
+ } else {
+ // using the String object will result in still an empty comment, but allows
+ // separate instance equality check when the entry is recorded.
+ const sentinel = new String(); // eslint-disable-line no-new-wrappers
+ callbacks.set(sentinel, cb);
+ resultStream.append(s, Object.assign(options, { comment: sentinel }));
+ }
+ });
+ part.stream.on('error', (err) => { resultStream.emit('error', err); });
+ part.stream.on('end', next);
+ })
+ .catch((err) => { resultStream.emit('error', err); });
+ };
+ next();
 
  return resultStream;
 };

package.json

@@ -35,7 +35,7 @@
  "prompt": "~1",
  "ramda": "~0",
  "sanitize-filename": "~1",
- "slonik": "~23",
+ "slonik": "npm:@getodk/slonik@23.6.0-1",
  "slonik-sql-tag-raw": "1.0.3",
  "tmp-promise": "~3",
  "uuid": "~3",