chore(build): fix gulp async task completion

ReactiveX · Jul 10, 2024 · c68e97c · c68e97c
1 parent 512e370
commit c68e97c
Show file tree

Hide file tree

Showing 4 changed files with 11 additions and 12 deletions.
diff --git a/gulp/copy-main-task.js b/gulp/copy-main-task.js
@@ -6,8 +6,7 @@ import { mkdirp } from 'mkdirp';
 import gulpRename from 'gulp-rename';
 import gulpReplace from 'gulp-replace';
 import { memoizeTask } from './memoize-task.js';
-import { ReplaySubject, forkJoin as ObservableForkJoin } from 'rxjs';
-import { share } from 'rxjs/operators/index.js';
+import { forkJoin as ObservableForkJoin } from 'rxjs';
 import { pipeline } from 'stream/promises';
 
 export const copyMainTask = ((cache) => memoizeTask(cache, function copyMain(target) {
@@ -31,7 +30,7 @@ export const copyMainTask = ((cache) => memoizeTask(cache, function copyMain(tar
         observableFromStreams(gulp.src(esmGlob), gulpRename((p) => { p.extname = '.mjs'; }), gulpReplace(`.js'`, `.mjs'`), gulpReplace(`.js.map`, `.mjs.map`), gulp.dest(out)), // copy es2015 esm files and rename to `.mjs`
         observableFromStreams(gulp.src(es2015UmdGlob), gulpRename((p) => { p.basename += `.es2015.min`; }), gulp.dest(out)), // copy es2015 umd files and add `.es2015.min`
         observableFromStreams(gulp.src(esnextUmdGlob), gulpRename((p) => { p.basename += `.esnext.min`; }), gulp.dest(out)), // copy esnext umd files and add `.esnext.min`
-    ]).pipe(share({ connector: () => new ReplaySubject(), resetOnError: false, resetOnComplete: false, resetOnRefCountZero: false }));
+    ]).toPromise();
 }))({});
 
 export const copyTSTask = ((cache) => memoizeTask(cache, async function copyTS(target, format) {

diff --git a/gulp/package-task.js b/gulp/package-task.js
@@ -2,8 +2,7 @@ import { metadataFiles, packageJSONFields, npmPkgName, npmOrgName, targetDir, pa
 
 import gulp from 'gulp';
 import { memoizeTask } from './memoize-task.js';
-import { ReplaySubject, EMPTY as ObservableEmpty, forkJoin as ObservableForkJoin } from 'rxjs';
-import { share } from 'rxjs/operators/index.js';
+import { EMPTY as ObservableEmpty, forkJoin as ObservableForkJoin } from 'rxjs';
 import gulpJsonTransform from 'gulp-json-transform';
 
 export const packageTask = ((cache) => memoizeTask(cache, function bundle(target, format) {
@@ -16,7 +15,7 @@ export const packageTask = ((cache) => memoizeTask(cache, function bundle(target
     return ObservableForkJoin([
         observableFromStreams(gulp.src(metadataFiles), gulp.dest(out)), // copy metadata files
         observableFromStreams(gulp.src(`package.json`), jsonTransform, gulp.dest(out)) // write packageJSONs
-    ]).pipe(share({ connector: () => new ReplaySubject(), resetOnError: false, resetOnComplete: false, resetOnRefCountZero: false }));
+    ]).toPromise();
 }))({});
 
 export default packageTask;

diff --git a/gulp/typescript-task.js b/gulp/typescript-task.js
@@ -6,8 +6,8 @@ import tsc from 'typescript';
 import ts from 'gulp-typescript';
 import sourcemaps from 'gulp-sourcemaps';
 import { memoizeTask } from './memoize-task.js';
-import { ReplaySubject, forkJoin as ObservableForkJoin, defer as ObservableDefer } from 'rxjs';
-import { takeLast, share, concat } from 'rxjs/operators/index.js';
+import { forkJoin as ObservableForkJoin } from 'rxjs';
+import { takeLast } from 'rxjs/operators/index.js';
 
 export const typescriptTask = ((cache) => memoizeTask(cache, function typescript(target, format) {
     if (shouldRunInChildProcess(target, format)) {
@@ -18,7 +18,7 @@ export const typescriptTask = ((cache) => memoizeTask(cache, function typescript
     const tsconfigPath = path.join(`tsconfig`, `tsconfig.${tsconfigName(target, format)}.json`);
     return compileTypescript(out, tsconfigPath)
         .pipe(takeLast(1))
-        .pipe(share({ connector: () => new ReplaySubject(), resetOnError: false, resetOnComplete: false, resetOnRefCountZero: false }))
+        .toPromise();
 }))({});
 
 export default typescriptTask;

diff --git a/gulp/util.js b/gulp/util.js
@@ -6,8 +6,8 @@ import util from 'node:util';
 import asyncDoneSync from 'async-done';
 const pump = stream.pipeline;
 import { targets, modules } from './argv.js';
-import { ReplaySubject, empty as ObservableEmpty, throwError as ObservableThrow, fromEvent as ObservableFromEvent } from 'rxjs';
-import { share, flatMap, takeUntil, defaultIfEmpty, merge } from 'rxjs/operators/index.js';
+import { empty as ObservableEmpty, throwError as ObservableThrow, fromEvent as ObservableFromEvent } from 'rxjs';
+import { flatMap, takeUntil, defaultIfEmpty, merge, publishReplay } from 'rxjs/operators/index.js';
 const asyncDone = util.promisify(asyncDoneSync);
 import { createRequire } from 'node:module';
 
@@ -107,10 +107,11 @@ function observableFromStreams(...streams) {
         merge(fromEvent(`error`).pipe(flatMap((e) => ObservableThrow(e)))),
         takeUntil(fromEvent(`end`).pipe(merge(fromEvent(`close`)))),
         defaultIfEmpty(`empty stream`),
-        share({ connector: () => new ReplaySubject(), resetOnError: false, resetOnComplete: false, resetOnRefCountZero: false })
+        publishReplay()
     );
     streamObs.stream = pumped;
     streamObs.observable = streamObs;
+    streamObs.subscription = streamObs.connect();
     return streamObs;
 }