Merge pull request #20 from WorldMaker/artisinal-subscribers

Ensure all of our subscribers are artisanal with proper closures
WorldMaker · Feb 17, 2022 · 44e6974 · 44e6974
2 parents 8b95eae + 672a27c
commit 44e6974
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 17 deletions.
diff --git a/projects/angular-pharkas/package.json b/projects/angular-pharkas/package.json
@@ -1,6 +1,6 @@
 {
   "name": "angular-pharkas",
-  "version": "0.5.0",
+  "version": "0.5.1",
   "peerDependencies": {
     "@angular/common": "^11.1.1",
     "@angular/core": "^11.1.1"

diff --git a/projects/angular-pharkas/src/pharkas.component.ts b/projects/angular-pharkas/src/pharkas.component.ts
@@ -53,6 +53,17 @@ interface PharkasComponentState<T> extends Observable<T> {
   [pharkas]: boolean
 }
 
+function bindSubject<T>(observable: Observable<T>, subject: Subject<T>) {
+  // Zone's monkey patching of RxJS breaks the obvious binding (observable.subscribe(subject))
+  // as RxJS doesn't think it "safe" so we need to do this the hard way to support migrating
+  // out of Zone apps
+  return observable.subscribe({
+    next: (value: T) => subject.next(value),
+    error: (err: any) => subject.error(err),
+    complete: () => subject.complete(),
+  })
+}
+
 /**
  * Pharkas Base Component
  *
@@ -97,7 +108,7 @@ export class PharkasComponent<TViewModel> implements OnInit, OnDestroy {
         }
         bound = true
         if (isObservable(value)) {
-          this[subscription].add(value.subscribe(subject))
+          this[subscription].add(bindSubject(value, subject))
         } else {
           subject.next(value)
         }
@@ -195,7 +206,7 @@ export class PharkasComponent<TViewModel> implements OnInit, OnDestroy {
       throw new Error(`${name} is already bound`)
     }
     const subject = new BehaviorSubject(defaultValue)
-    this[subscription].add(observable.subscribe(subject))
+    this[subscription].add(bindSubject(observable, subject))
     this[props].set(name, {
       type: 'display',
       name,
@@ -222,7 +233,7 @@ export class PharkasComponent<TViewModel> implements OnInit, OnDestroy {
       throw new Error(`${name} is already bound`)
     }
     const subject = new BehaviorSubject(defaultValue)
-    this[subscription].add(observable.subscribe(subject))
+    this[subscription].add(bindSubject(observable, subject))
     this[props].set(name, {
       type: 'display',
       name,
@@ -243,7 +254,7 @@ export class PharkasComponent<TViewModel> implements OnInit, OnDestroy {
     eventEmitter: EventEmitter<T>,
     observable: Observable<T>
   ) {
-    this[subscription].add(observable.subscribe(eventEmitter))
+    this[subscription].add(bindSubject(observable, eventEmitter))
   }
 
   //#region *** Callbacks ***
@@ -382,7 +393,7 @@ export class PharkasComponent<TViewModel> implements OnInit, OnDestroy {
     }
     const subject = this[state].get(localState)
     if (subject) {
-      this[subscription].add(observable.subscribe(subject))
+      this[subscription].add(bindSubject(observable, subject))
       localState[pharkas] = true
     } else {
       throw new Error('Unknown local component state')
@@ -410,9 +421,10 @@ export class PharkasComponent<TViewModel> implements OnInit, OnDestroy {
     const subject = this[state].get(localState) as BehaviorSubject<T>
     if (subject) {
       this[subscription].add(
-        observable
-          .pipe(map((value) => reducer(subject.value, value)))
-          .subscribe(subject)
+        bindSubject(
+          observable.pipe(map((value) => reducer(subject.value, value))),
+          subject
+        )
       )
       localState[pharkas] = true
     } else {
@@ -442,12 +454,13 @@ export class PharkasComponent<TViewModel> implements OnInit, OnDestroy {
     const subject = this[state].get(localState) as BehaviorSubject<T>
     if (subject) {
       this[subscription].add(
-        from(observables)
-          .pipe(
+        bindSubject(
+          from(observables).pipe(
             mergeAll(),
             map((value) => reducer(subject.value, value))
-          )
-          .subscribe(subject)
+          ),
+          subject
+        )
       )
       localState[pharkas] = true
     } else {
@@ -512,7 +525,7 @@ export class PharkasComponent<TViewModel> implements OnInit, OnDestroy {
   //#endregion
 
   ngOnInit(): void {
-    const subjects: Subject<unknown>[] = []
+    const observables: Observable<unknown>[] = []
     for (const prop of this[props].values()) {
       if (prop.type === 'display') {
         if (prop.immediate) {
@@ -522,12 +535,12 @@ export class PharkasComponent<TViewModel> implements OnInit, OnDestroy {
             })
           )
         } else {
-          subjects.push(prop.subject)
+          observables.push(prop.subject.asObservable())
         }
       }
     }
-    if (subjects.length) {
-      const displayObservable = merge(...subjects).pipe(
+    if (observables.length) {
+      const displayObservable = merge(...observables).pipe(
         throttleTime(0, animationFrameScheduler),
         share()
       )