Merge branch 'features/deits' into deits/engine-tests

packages/core/data-transfer/lib/__tests__/test-utils.ts

@@ -6,12 +6,13 @@ import { Readable } from 'stream';
 export const collect = <T = unknown>(stream: Readable): Promise<T[]> => {
   const chunks: T[] = [];
 
-  return new Promise((resolve) => {
-    stream.on('data', (chunk) => chunks.push(chunk));
-    stream.on('end', () => {
-      stream.destroy();
-      resolve(chunks);
-    });
+  return new Promise((resolve, reject) => {
+    stream
+      .on('data', (chunk) => chunks.push(chunk))
+      .on('close', () => {
+        resolve(chunks);
+      })
+      .on('error', reject);
   });
 };
 

packages/core/data-transfer/lib/providers/local-strapi-source-provider/__tests__/links.test.ts

@@ -1,4 +1,4 @@
-import { Duplex } from 'stream';
+import { Readable } from 'stream';
 
 import { createLinksStream } from '../links';
 import { getDeepPopulateQuery } from '../links/utils';
@@ -15,7 +15,7 @@ describe('Local Strapi Source Provider - Entities Streaming', () => {
 
       const stream = createLinksStream(strapi);
 
-      expect(stream).toBeInstanceOf(Duplex);
+      expect(stream).toBeInstanceOf(Readable);
 
       const links = await collect(stream);
 
@@ -60,7 +60,7 @@ describe('Local Strapi Source Provider - Entities Streaming', () => {
 
       const stream = createLinksStream(strapi);
 
-      expect(stream).toBeInstanceOf(Duplex);
+      expect(stream).toBeInstanceOf(Readable);
 
       const links = await collect(stream);
 
@@ -140,7 +140,7 @@ describe('Local Strapi Source Provider - Entities Streaming', () => {
 
       const stream = createLinksStream(strapi);
 
-      expect(stream).toBeInstanceOf(Duplex);
+      expect(stream).toBeInstanceOf(Readable);
 
       const links = await collect(stream);
 
@@ -238,7 +238,7 @@ describe('Local Strapi Source Provider - Entities Streaming', () => {
 
       const stream = createLinksStream(strapi);
 
-      expect(stream).toBeInstanceOf(Duplex);
+      expect(stream).toBeInstanceOf(Readable);
 
       const links = await collect(stream);
 

packages/core/data-transfer/lib/providers/local-strapi-source-provider/configuration.ts

@@ -1,11 +1,11 @@
-import { Duplex } from 'stream';
 import { chain } from 'stream-chain';
+import { Readable } from 'stream';
 import { set } from 'lodash/fp';
 
 /**
  * Create a readable stream that export the Strapi app configuration
  */
-export const createConfigurationStream = (strapi: Strapi.Strapi): Duplex => {
+export const createConfigurationStream = (strapi: Strapi.Strapi): Readable => {
   // Core Store
   const coreStoreStream = chain([
     strapi.db.queryBuilder('strapi::core-store').stream(),
@@ -22,13 +22,15 @@ export const createConfigurationStream = (strapi: Strapi.Strapi): Duplex => {
   const streams = [coreStoreStream, webhooksStream];
 
   // Readable configuration stream
-  return Duplex.from(async function* () {
-    for (const stream of streams) {
-      for await (const item of stream) {
-        yield item;
+  return Readable.from(
+    (async function* () {
+      for (const stream of streams) {
+        for await (const item of stream) {
+          yield item;
+        }
       }
-    }
-  });
+    })()
+  );
 };
 
 const wrapConfigurationItem = (type: 'core-store' | 'webhook') => (value: unknown) => ({

packages/core/data-transfer/lib/providers/local-strapi-source-provider/entities.ts

@@ -1,11 +1,11 @@
 import type { ContentTypeSchema } from '@strapi/strapi';
 
-import { Readable, Duplex, PassThrough } from 'stream';
+import { Readable, PassThrough } from 'stream';
 
 /**
  * Generate and consume content-types streams in order to stream each entity individually
  */
-export const createEntitiesStream = (strapi: Strapi.Strapi): Duplex => {
+export const createEntitiesStream = (strapi: Strapi.Strapi): Readable => {
   const contentTypes: ContentTypeSchema[] = Object.values(strapi.contentTypes);
 
   async function* contentTypeStreamGenerator() {
@@ -22,15 +22,17 @@ export const createEntitiesStream = (strapi: Strapi.Strapi): Duplex => {
     }
   }
 
-  return Duplex.from(async function* () {
-    for await (const { stream, contentType } of contentTypeStreamGenerator()) {
-      for await (const entity of stream) {
-        yield { entity, contentType };
-      }
+  return Readable.from(
+    (async function* () {
+      for await (const { stream, contentType } of contentTypeStreamGenerator()) {
+        for await (const entity of stream) {
+          yield { entity, contentType };
+        }
 
-      stream.destroy();
-    }
-  });
+        stream.destroy();
+      }
+    })()
+  );
 };
 
 /**

packages/core/data-transfer/lib/providers/local-strapi-source-provider/links/index.ts

@@ -1,6 +1,6 @@
 import type { ContentTypeSchema, GetAttributesValues, RelationsType } from '@strapi/strapi';
 
-import { Duplex } from 'stream';
+import { Readable } from 'stream';
 import { castArray } from 'lodash/fp';
 
 import { getDeepPopulateQuery, parseEntityLinks } from './utils';
@@ -8,7 +8,7 @@ import { getDeepPopulateQuery, parseEntityLinks } from './utils';
 /**
  * Create a Duplex instance which will stream all the links from a Strapi instance
  */
-export const createLinksStream = (strapi: Strapi.Strapi): Duplex => {
+export const createLinksStream = (strapi: Strapi.Strapi): Readable => {
   const schemas: ContentTypeSchema[] = Object.values(strapi.contentTypes);
 
   // Destroy the Duplex stream instance
@@ -19,25 +19,27 @@ export const createLinksStream = (strapi: Strapi.Strapi): Duplex => {
   };
 
   // Async generator stream that returns every link from a Strapi instance
-  const stream = Duplex.from(async function* () {
-    for (const schema of schemas) {
-      const populate = getDeepPopulateQuery(schema, strapi);
-      const query = { fields: ['id'], populate };
+  const stream = Readable.from(
+    (async function* () {
+      for (const schema of schemas) {
+        const populate = getDeepPopulateQuery(schema, strapi);
+        const query = { fields: ['id'], populate };
 
-      // TODO: Replace with the DB stream API
-      const results = await strapi.entityService.findMany(schema.uid, query);
+        // TODO: Replace with the DB stream API
+        const results = await strapi.entityService.findMany(schema.uid, query);
 
-      for (const entity of castArray(results)) {
-        const links = parseEntityLinks(entity, populate, schema, strapi);
+        for (const entity of castArray(results)) {
+          const links = parseEntityLinks(entity, populate, schema, strapi);
 
-        for (const link of links) {
-          yield link;
+          for (const link of links) {
+            yield link;
+          }
         }
       }
-    }
 
-    destroy();
-  });
+      destroy();
+    })()
+  );
 
   return stream;
 };