fix(classes): cleanup resources, add cancellable timeouts, and fix bugs in several core utility classes

ts/classes.backpressuredarray.ts

@@ -5,6 +5,7 @@ export class BackpressuredArray<T> {
   private highWaterMark: number;
   public hasSpace = new plugins.smartrx.rxjs.Subject<'hasSpace'>();
   private itemsAvailable = new plugins.smartrx.rxjs.Subject<'itemsAvailable'>();
+  private isDestroyed = false;
 
   constructor(highWaterMark: number = 16) {
     this.data = [];
@@ -14,7 +15,7 @@ export class BackpressuredArray<T> {
   push(item: T): boolean {
     this.data.push(item);
     this.itemsAvailable.next('itemsAvailable');
-    
+
     const spaceAvailable = this.checkSpaceAvailable();
     if (spaceAvailable) {
       this.hasSpace.next('hasSpace');
@@ -40,12 +41,17 @@ export class BackpressuredArray<T> {
 
   waitForSpace(): Promise<void> {
     return new Promise<void>((resolve) => {
-      if (this.checkSpaceAvailable()) {
+      if (this.checkSpaceAvailable() || this.isDestroyed) {
         resolve();
       } else {
-        const subscription = this.hasSpace.subscribe(() => {
-          subscription.unsubscribe();
-          resolve();
+        const subscription = this.hasSpace.subscribe({
+          next: () => {
+            subscription.unsubscribe();
+            resolve();
+          },
+          complete: () => {
+            resolve();
+          },
         });
       }
     });
@@ -53,14 +59,28 @@ export class BackpressuredArray<T> {
 
   waitForItems(): Promise<void> {
     return new Promise<void>((resolve) => {
-      if (this.data.length > 0) {
+      if (this.data.length > 0 || this.isDestroyed) {
         resolve();
       } else {
-        const subscription = this.itemsAvailable.subscribe(() => {
-          subscription.unsubscribe();
-          resolve();
+        const subscription = this.itemsAvailable.subscribe({
+          next: () => {
+            subscription.unsubscribe();
+            resolve();
+          },
+          complete: () => {
+            resolve();
+          },
         });
       }
     });
   }
+
+  /**
+   * destroys the BackpressuredArray, completing all subjects
+   */
+  public destroy() {
+    this.isDestroyed = true;
+    this.hasSpace.complete();
+    this.itemsAvailable.complete();
+  }
 }