Writing tests to check streaming (result chunks) of stream deletions

Author: Perki

components/api-server/src/Result.js

@@ -196,7 +196,8 @@ class Result {
     const streams = [];
     for (let i = 0; i < streamsArray.length; i++) {
       const s = streamsArray[i];
-      streams.push(s.stream.pipe(s.isArray ? new ArraySerializationStream(s.name) : new SingleObjectSerializationStream(s.name)));
+      const serializedStream = s.stream.pipe(s.isArray ? new ArraySerializationStream(s.name) : new SingleObjectSerializationStream(s.name));
+      streams.push(serializedStream);
     }
 
     return new MultiStream(streams)

components/api-server/src/methods/streams/ArraySerializationStream.js

@@ -6,7 +6,10 @@
  */
 const Transform = require('stream').Transform;
 
-const SERIALIZATION_STACK_SIZE = 2048;
+// serialize every n objects
+const OBJECT_BUFFER_SIZE = 100;
+// event if OBJECT_BUFFER_SIZE is not reach, serialize if MAX_WAIT_MS is reached
+const MAX_WAIT_MS = 100;
 
 /**
  * Stream that encapsulates the items it receives in a stringified array.
@@ -19,20 +22,22 @@ module.exports = class ArraySerializationStream extends Transform {
     super({ writableObjectMode: true });
     this.isStart = true;
     this.prefix = '"' + arrayName + '":';
-    this.size = SERIALIZATION_STACK_SIZE;
+    this.size = OBJECT_BUFFER_SIZE;
     this.stack = [];
+    this.lastSerialization = Date.now();
   }
 
   _transform (item, encoding, callback) {
     this.stack.push(item);
 
-    if (this.stack.length >= this.size) {
+    if (this.stack.length >= this.size || (Date.now() - this.lastSerialization) > MAX_WAIT_MS) {
       if (this.isStart) {
         this.isStart = false;
         this.push((this.prefix + JSON.stringify(this.stack)).slice(0, -1));
       } else {
         this.push(',' + (JSON.stringify(this.stack)).slice(1, -1));
       }
+      this.lastSerialization = Date.now();
       this.stack = [];
     }
     callback();

components/api-server/test/events-streaming.test.js

@@ -12,6 +12,8 @@ const { produceMongoConnection, context } = require('./test-helpers');
 const { databaseFixture } = require('test-helpers');
 
 const http = require('http');
+const superagent = require('superagent');
+const { promisify } = require('util');
 
 const N_ITEMS = 2000;
 describe('events streaming with ' + N_ITEMS + ' entries', function () {
@@ -66,13 +68,16 @@ describe('events streaming with ' + N_ITEMS + ' entries', function () {
       method: 'GET'
     };
 
+    let lastChunkRecievedAt = Date.now();
     http.request(options, function (res) {
       assert.equal(res.headers['content-type'], 'application/json');
       assert.equal(res.headers['transfer-encoding'], 'chunked');
       res.setEncoding('utf8');
       let jsonString = '';
       let chunkCount = 0;
       res.on('data', function (chunk) {
+        if (Date.now() - lastChunkRecievedAt > 500) throw new Error('It took more that 500ms between chunks');
+        lastChunkRecievedAt = Date.now();
         chunkCount++;
         jsonString += chunk;
       });
@@ -86,4 +91,43 @@ describe('events streaming with ' + N_ITEMS + ' entries', function () {
       });
     }).end();
   });
+
+  it('[XZGB] Streams deleted in sent as chunked', async function () {
+    const options = {
+      host: apiServer.host,
+      port: apiServer.port,
+      path: '/' + username + '/streams/' + streamId + '?mergeEventsWithParent=false&auth=' + appAccessToken,
+      method: 'DELETE'
+    };
+
+    const resultTrash = await superagent.delete(`http://${options.host}:${options.port}${options.path}`);
+    assert.isTrue(resultTrash.body?.stream?.trashed);
+
+    let lastChunkRecievedAt = Date.now();
+
+    await promisify(function (callback) {
+      http.request(options, function (res) {
+        assert.equal(res.headers['content-type'], 'application/json');
+        assert.equal(res.headers['transfer-encoding'], 'chunked');
+        res.setEncoding('utf8');
+        let jsonString = '';
+        let chunkCount = 0;
+        res.on('data', function (chunk) {
+          if (Date.now() - lastChunkRecievedAt > 2000) throw new Error('It took more that 2000ms between chunks');
+          lastChunkRecievedAt = Date.now();
+          chunkCount++;
+          jsonString += chunk;
+        });
+        res.on('end', () => {
+          lastChunkRecievedAt = -1;
+          assert.equal(JSON.parse(jsonString).updatedEvents.length, N_ITEMS);
+          assert.isAtLeast(chunkCount, 3, 'Should receive at least 3 chunks');
+          callback();
+        });
+        res.on('error', function (error) {
+          callback(error);
+        });
+      }).end();
+    })();
+  });
 });

components/api-server/test/streams.test.js

@@ -664,7 +664,7 @@ describe('[STRE] streams', function () {
     });
   });
 
-  describe('DELETE /<id>', function () {
+  describe('[STRD] DELETE /<id>', function () {
     this.timeout(5000);
 
     beforeEach(resetData);
@@ -833,6 +833,8 @@ describe('[STRE] streams', function () {
       const deletedEventWithAtt = deletedEvents[0];
       let deletionTime;
 
+      const ADD_N_EVENTS = 100;
+
       async.series([
         function addEventAttachment (stepDone) {
           request.post('/' + user.username + '/events/' + deletedEventWithAtt.id)
@@ -844,6 +846,22 @@ describe('[STRE] streams', function () {
               stepDone();
             });
         },
+        async function fillStreamWithALotOfEvent () {
+          const mall = await getMall();
+          for (let i = 0; i < ADD_N_EVENTS; i++) {
+            await mall.events.create(user.id, {
+              id: 'cxxxxxxx' + i,
+              type: 'note/txt',
+              streamIds: [testData.streams[8].id],
+              content: '' + i,
+              time: timestamp.now(),
+              created: timestamp.now(),
+              createdBy: 'test',
+              modified: timestamp.now(),
+              modifiedBy: 'test'
+            });
+          }
+        },
         async function trashStream () {
           await mall.streams.update(user.id, { id, trashed: true });
         },
@@ -870,7 +888,7 @@ describe('[STRE] streams', function () {
           const separatedEvents = validation.separateAccountStreamsAndOtherEvents(events);
           events = separatedEvents.events;
           const eventsWithoutHistory = testData.events.filter(e => e.headId == null);
-          (events.length + foundDeletedEvents.length).should.eql(eventsWithoutHistory.length, 'events');
+          (events.length + foundDeletedEvents.length).should.eql(eventsWithoutHistory.length + ADD_N_EVENTS, 'events');
 
           // validate account streams events
           const actualAccountStreamsEvents = separatedEvents.accountStreamsEvents;