feat: added WorkerPipe

Author: IKatsuba

projects/demo/src/app/app.browser.module.ts

@@ -7,7 +7,7 @@ import {
 import {NgModule} from '@angular/core';
 import {FormsModule} from '@angular/forms';
 import {BrowserModule} from '@angular/platform-browser';
-import {WebWorkerModule} from '@ng-web-apis/workers';
+import {WorkerModule} from '@ng-web-apis/workers';
 import {AppComponent} from './app.component';
 import {AppRoutingModule} from './app.routes';
 
@@ -18,7 +18,7 @@ import {AppRoutingModule} from './app.routes';
         FormsModule,
         BrowserModule.withServerTransition({appId: 'demo'}),
         AppRoutingModule,
-        WebWorkerModule,
+        WorkerModule,
     ],
     declarations: [AppComponent],
     providers: [

projects/demo/src/app/app.component.html

@@ -4,4 +4,4 @@
     <button type="button" (click)="workerThread.complete()">Complete</button>
 </form>
 
-{{ workerThread | async | json }}
+{{ workerThread | async | waWorker: oneMoreFn | async | json }}

projects/demo/src/app/app.component.ts

@@ -1,5 +1,5 @@
 import {ChangeDetectionStrategy, Component} from '@angular/core';
-import {WebWorker, WebWorkerExecutor} from '@ng-web-apis/workers';
+import {WebWorker, WorkerExecutor} from '@ng-web-apis/workers';
 
 @Component({
     selector: 'main',
@@ -10,9 +10,13 @@ import {WebWorker, WebWorkerExecutor} from '@ng-web-apis/workers';
 export class AppComponent {
     public workerThread: WebWorker<string, string>;
 
-    constructor(webWorkerExecutor: WebWorkerExecutor) {
+    constructor(webWorkerExecutor: WorkerExecutor) {
         this.workerThread = webWorkerExecutor.createWorker((result: string) =>
             Promise.resolve(`Message from worker: ${result}`),
         );
     }
+
+    oneMoreFn(data: any): Promise<any> {
+        return Promise.resolve().then(() => data);
+    }
 }

projects/workers/src/public-api.ts

@@ -1,7 +1,7 @@
 /**
  * Public API Surface of @ng-web-apis/workers
  */
-export * from './web-worker/web-worker.module';
-export * from './web-worker/types/web-worker-function';
-export * from './web-worker/services/web-worker-executor.service';
-export * from './web-worker/classes/web-worker';
+export * from './worker/worker.module';
+export * from './worker/types/worker-function';
+export * from './worker/services/worker-executor.service';
+export * from './worker/classes/web-worker';

projects/workers/src/web-worker/services/web-worker-executor.service.ts

@@ -1,12 +1,12 @@
 import {fromEvent} from 'rxjs';
 import {takeWhile} from 'rxjs/operators';
-import {WebWorkerFunction} from '../types/web-worker-function';
+import {WorkerFunction} from '../types/worker-function';
 import {AnyNextSubject} from './any-next-subject';
 
-export class WebWorker<T, R> extends AnyNextSubject<R> {
+export class WebWorker<T = any, R = any> extends AnyNextSubject<R> {
     private worker!: Worker;
 
-    private static createFnUrl(fn: WebWorkerFunction): string {
+    private static createFnUrl(fn: WorkerFunction): string {
         const script = `
 self.addEventListener('message', function(e) {
     var result = ((${fn.toString()}).call(null, e.data));
@@ -28,7 +28,7 @@ self.addEventListener('message', function(e) {
     }
 
     public static fromFunction<T, R>(
-        fn: WebWorkerFunction<T, R>,
+        fn: WorkerFunction<T, R>,
         options?: WorkerOptions,
     ): WebWorker<T, R> {
         return new WebWorker<T, R>(WebWorker.createFnUrl(fn), options);

projects/workers/src/web-worker/types/web-worker-function.ts

@@ -0,0 +1,10 @@
+import {WorkerExecutor} from '../services/worker-executor.service';
+import {WorkerPipe} from './worker.pipe';
+
+describe('WorkerPipe', () => {
+    it('create an instance', () => {
+        const pipe = new WorkerPipe(new WorkerExecutor());
+
+        expect(pipe).toBeTruthy();
+    });
+});

projects/workers/src/web-worker/web-worker.module.ts

@@ -0,0 +1,25 @@
+import {Pipe, PipeTransform} from '@angular/core';
+import {WebWorker} from '../classes/web-worker';
+import {WorkerExecutor} from '../services/worker-executor.service';
+import {WorkerFunction} from '../types/worker-function';
+
+@Pipe({
+    name: 'waWorker',
+})
+export class WorkerPipe implements PipeTransform {
+    private workers = new WeakMap<WorkerFunction, WebWorker>();
+
+    constructor(private workerExecutor: WorkerExecutor) {}
+
+    transform(value: any, fn: WorkerFunction): any {
+        const worker = this.workers.has(fn)
+            ? (this.workers.get(fn) as WebWorker)
+            : this.workerExecutor.createWorker(fn);
+
+        this.workers.set(fn, worker);
+
+        worker.next(value);
+
+        return worker;
+    }
+}

projects/workers/src/web-worker/classes/any-next-subject.ts renamed to projects/workers/src/worker/classes/any-next-subject.ts

@@ -1,12 +1,12 @@
 import {TestBed} from '@angular/core/testing';
 
-import {WebWorkerExecutor} from './web-worker-executor.service';
+import {WorkerExecutor} from './worker-executor.service';
 
 describe('WebWorkerExecutorService', () => {
     beforeEach(() => TestBed.configureTestingModule({}));
 
     it('should be created', () => {
-        const service: WebWorkerExecutor = TestBed.get(WebWorkerExecutor);
+        const service: WorkerExecutor = TestBed.get(WorkerExecutor);
 
         expect(service).toBeTruthy();
     });

projects/workers/src/web-worker/classes/web-worker.ts renamed to projects/workers/src/worker/classes/web-worker.ts

@@ -0,0 +1,23 @@
+import {Injectable} from '@angular/core';
+import {WebWorker} from '../classes/web-worker';
+import {WorkerFunction} from '../types/worker-function';
+
+@Injectable()
+export class WorkerExecutor {
+    execute<T, R>(fn: WorkerFunction<T, R>, data?: T): Promise<R> {
+        const worker = this.createWorker(fn);
+        const promise = worker.toPromise();
+
+        worker.next(data);
+        worker.complete();
+
+        return promise;
+    }
+
+    createWorker<T, R>(
+        fn: WorkerFunction<T, R>,
+        options?: WorkerOptions,
+    ): WebWorker<T, R> {
+        return WebWorker.fromFunction<T, R>(fn, options);
+    }
+}

projects/workers/src/worker/pipes/worker.pipe.spec.ts

@@ -0,0 +1 @@
+export type WorkerFunction<T = any, R = any> = (data: T) => R | PromiseLike<R>;

projects/workers/src/worker/pipes/worker.pipe.ts

@@ -0,0 +1,21 @@
+import {CommonModule} from '@angular/common';
+import {inject, InjectFlags, NgModule} from '@angular/core';
+import {WorkerPipe} from './pipes/worker.pipe';
+import {WorkerExecutor} from './services/worker-executor.service';
+
+@NgModule({
+    imports: [CommonModule],
+    declarations: [WorkerPipe],
+    exports: [WorkerPipe],
+    providers: [
+        {
+            provide: WorkerExecutor,
+            useFactory(): WorkerExecutor {
+                const instance = inject(WorkerExecutor, InjectFlags.Optional);
+
+                return instance || new WorkerExecutor();
+            },
+        },
+    ],
+})
+export class WorkerModule {}