Support pipable cancelOnDestroy and tag operators

kaltura-common/src/rxjs/operators/cancel-on-destroy.ts

@@ -0,0 +1,49 @@
+import { Observable, Subject } from 'rxjs';
+import { OnDestroy } from '@angular/core';
+import { merge } from 'rxjs/observable/merge';
+
+interface EnhancedOnDestroy extends OnDestroy {
+    __ngOnDestroySource__: Subject<string>;
+    __ngOnDestroy__: () => void;
+}
+
+export function cancelOnDestroy<T>(instance: OnDestroy, manualDestroy?: Observable<any>): (source: Observable<T>) => Observable<T> {
+    return (source: Observable<T>) => {
+        if (!instance.ngOnDestroy) {
+            return source;
+        }
+
+        return Observable.create(observer => {
+            let subscription = source.subscribe(observer);
+
+            if (!(<EnhancedOnDestroy>instance).__ngOnDestroySource__) {
+                (<EnhancedOnDestroy>instance).__ngOnDestroySource__ = new Subject();
+                (<EnhancedOnDestroy>instance).__ngOnDestroy__ = instance.ngOnDestroy;
+
+                instance.ngOnDestroy = function (this: EnhancedOnDestroy) {
+                    this.__ngOnDestroy__.apply(this, arguments);
+                    this.__ngOnDestroySource__.next();
+                    this.__ngOnDestroySource__.complete();
+                };
+            }
+
+            const sources = manualDestroy
+                ? merge(manualDestroy, (<EnhancedOnDestroy>instance).__ngOnDestroySource__)
+                : (<EnhancedOnDestroy>instance).__ngOnDestroySource__.asObservable();
+
+            sources.subscribe(() => {
+                if (subscription) {
+                    subscription.unsubscribe();
+                    subscription = null;
+                }
+            });
+
+            return () => {
+                if (subscription) {
+                    subscription.unsubscribe();
+                    subscription = null;
+                }
+            }
+        });
+    }
+}

kaltura-common/src/rxjs/operators/index.ts

@@ -0,0 +1,2 @@
+export { cancelOnDestroy } from './cancel-on-destroy';
+export { tag } from './tag';

kaltura-common/src/rxjs/operators/tag.ts

@@ -0,0 +1,44 @@
+import { Observable } from 'rxjs';
+import { OperationTagStoreMediator } from '../../operation-tag/operation-tag-store-mediator';
+
+export function tag<T>(action: string): (source: Observable<T>) => Observable<T> {
+    let isDecreased = false;
+    let closed = false;
+
+    return (source: Observable<T>) => Observable.create(observer => {
+        OperationTagStoreMediator.increase(action);
+
+        let subscription = source.subscribe(
+            (value) => {
+                observer.next(value);
+            },
+            error => {
+                if (action && !isDecreased) {
+                    isDecreased = true;
+                    OperationTagStoreMediator.decrease(action);
+                }
+                observer.error(error);
+            },
+            () => {
+                if (action && !isDecreased) {
+                    isDecreased = true;
+                    OperationTagStoreMediator.decrease(action);
+                }
+                observer.complete();
+            }
+        );
+
+        return () => {
+            if (!closed && action && !isDecreased) {
+                isDecreased = true;
+                OperationTagStoreMediator.decrease(action);
+            }
+
+            if (subscription) {
+                subscription.unsubscribe();
+                subscription = null;
+                closed = true;
+            }
+        }
+    });
+}