Support pipable cancelOnDestroy and tag operators

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

@@ -0,0 +1,50 @@
+import { Observable } from 'rxjs/Observable';
+import { Subject } from 'rxjs/Subject';
+import { OnDestroy } from '@angular/core';
+import 'rxjs/add/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 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
+                    ? Observable.merge(manualDestroy, (<EnhancedOnDestroy>instance).__ngOnDestroySource__)
+                    : (<EnhancedOnDestroy>instance).__ngOnDestroySource__.asObservable();
+
+                sources.subscribe(() => {
+                    if (subscription) {
+                        subscription.unsubscribe();
+                        subscription = null;
+                    }
+                });
+
+                return () => {
+                    if (subscription) {
+                        subscription.unsubscribe();
+                        subscription = null;
+                    }
+                }
+            });
+        } else {
+            return source;
+        }
+    }
+}

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/Observable';
+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;
+            }
+        }
+    });
+}