Feature: Entities are refreshing every 10 seconds automatically

app/components/application/browse/application-list.component.ts

@@ -53,7 +53,7 @@ export class ApplicationListComponent extends ListOrTableBase implements OnInit,
 
         this.status = this.data.status;
         this._subs.push(applicationService.onApplicationAdded.subscribe((applicationId) => {
-            this.data.loadNewItem(applicationService.get(applicationId));
+            this.data.loadNewItem(applicationService.getOnce(applicationId));
         }));
     }
 

app/components/job/action/delete/delete-job-action.ts

@@ -47,23 +47,17 @@ export class WaitForDeleteJobPoller {
     @autobind()
     public start(progress: BehaviorSubject<any>): Observable<any> {
         const obs = new AsyncSubject();
-        const data = this.jobService.get(this.jobId);
-        const errorCallback = (e) => {
-            progress.next(100);
-            clearInterval(interval);
-            obs.complete();
-        };
-
         let interval = setInterval(() => {
-            data.fetch().subscribe({
-                error: errorCallback,
+            this.jobService.getOnce(this.jobId).subscribe({
+                error: (e) => {
+                    progress.next(100);
+                    clearInterval(interval);
+                    obs.complete();
+                },
             });
         }, 5000);
 
         progress.next(-1);
-        data.item.subscribe({
-            error: errorCallback,
-        });
 
         return obs;
     }

app/components/job/browse/job-list.component.ts

@@ -73,7 +73,7 @@ export class JobListComponent extends ListOrTableBase implements OnInit, OnDestr
         this.data = this.jobService.list(this._baseOptions);
         this.status = this.data.status;
         this._onJobAddedSub = jobService.onJobAdded.subscribe((jobId) => {
-            this.data.loadNewItem(jobService.get(jobId));
+            this.data.loadNewItem(jobService.getOnce(jobId));
         });
     }
 

app/components/job/details/job-details.component.ts

@@ -73,6 +73,7 @@ export class JobDetailsComponent implements OnInit, OnDestroy {
 
     public ngOnDestroy() {
         this._paramsSubscriber.unsubscribe();
+        this.data.dispose();
     }
 
     public get filterPlaceholderText() {

app/components/job/details/job-progress-status/job-progress-status.component.ts

@@ -73,6 +73,7 @@ export class JobProgressStatusComponent implements OnChanges, OnDestroy {
     public ngOnDestroy() {
         this._polls.forEach(x => x.destroy());
         this._subs.forEach(x => x.unsubscribe());
+        this.poolData.dispose();
     }
 
     public countRunningTasks() {

app/components/node/details/node-details.component.ts

@@ -77,6 +77,8 @@ export class NodeDetailsComponent implements OnInit, OnDestroy {
 
     public ngOnDestroy() {
         this._paramsSubscribers.forEach(x => x.unsubscribe());
+        this.poolData.dispose();
+        this.data.dispose();
     }
 
     @autobind()

app/components/pool/action/delete/delete-pool-task.ts

@@ -7,7 +7,7 @@ import { PoolService } from "app/services";
 import { LongRunningDeleteAction } from "app/services/core";
 
 export class DeletePoolTask extends LongRunningDeleteAction {
-    constructor(private poolService: PoolService,  poolIds: string[]) {
+    constructor(private poolService: PoolService, poolIds: string[]) {
         super("pool", poolIds);
     }
 
@@ -53,31 +53,25 @@ export class WaitForDeletePoolPollTask {
     @autobind()
     public start(progress: BehaviorSubject<any>): Observable<any> {
         const obs = new AsyncSubject();
-        const data = this.poolService.get(this.poolId);
         let interval;
 
-        const errorCallback = (e) => {
-            progress.next(100);
-            clearInterval(interval);
-            obs.complete();
-        };
-
         interval = setInterval(() => {
-            data.fetch().subscribe({
-                error: errorCallback,
+            this.poolService.getOnce(this.poolId).subscribe({
+                next: (pool: Pool) => {
+                    if (pool) {
+                        const currentNodes = pool.currentNodes;
+                        progress.next(this._getProgress(currentNodes));
+                    }
+                },
+                error: (e) => {
+                    progress.next(100);
+                    clearInterval(interval);
+                    obs.complete();
+                },
             });
         }, this.refreshRate);
 
         progress.next(10);
-        data.item.subscribe({
-            next: (pool: Pool) => {
-                if (pool) {
-                    const currentNodes = pool.currentNodes;
-                    progress.next(this._getProgress(currentNodes));
-                }
-            },
-            error: errorCallback,
-        });
 
         return obs;
     }

app/components/pool/browse/pool-list.component.ts

@@ -75,7 +75,7 @@ export class PoolListComponent extends ListOrTableBase implements OnInit, OnDest
         this.data = this.poolService.list();
         this.status = this.data.status;
         this._subs.push(poolService.onPoolAdded.subscribe((poolId) => {
-            this.data.loadNewItem(poolService.get(poolId));
+            this.data.loadNewItem(poolService.getOnce(poolId));
         }));
         this._subs.push(this.data.items.subscribe((pools) => {
             this.pools = List<PoolDecorator>(pools.map(x => new PoolDecorator(x)));

app/components/pool/details/pool-details.component.ts

@@ -69,6 +69,7 @@ export class PoolDetailsComponent implements OnInit, OnDestroy {
 
     public ngOnDestroy() {
         this._paramsSubscriber.unsubscribe();
+        this.data.dispose();
     }
 
     public get filterPlaceholderText() {

app/components/task/action/delete/delete-task-action.ts

@@ -48,24 +48,17 @@ export class WaitForDeleteTaskPoller {
     @autobind()
     public start(progress: BehaviorSubject<any>): Observable<any> {
         const obs = new AsyncSubject();
-        const data = this.taskService.get(this.jobId, this.taskId);
-        const errorCallback = (e) => {
-            progress.next(100);
-            clearInterval(interval);
-            obs.complete();
-        };
-
         let interval = setInterval(() => {
-            data.fetch().subscribe({
-                error: errorCallback,
+            this.taskService.getOnce(this.jobId, this.taskId).subscribe({
+                error: (e) => {
+                    progress.next(100);
+                    clearInterval(interval);
+                    obs.complete();
+                },
             });
         }, 5000);
 
         progress.next(-1);
-        data.item.subscribe({
-            error: errorCallback,
-        });
-
         return obs;
     }
 }

app/components/task/browse/task-list.component.ts

@@ -68,7 +68,7 @@ export class TaskListComponent extends SelectableList implements OnInit {
         super();
 
         this._onTaskAddedSub = taskService.onTaskAdded.subscribe((item: TaskParams) => {
-            this.data.loadNewItem(taskService.get(item.jobId, item.id));
+            this.data.loadNewItem(taskService.getOnce(item.jobId, item.id));
         });
     }
 

app/components/task/details/task-details.component.ts

@@ -82,6 +82,8 @@ export class TaskDetailsComponent implements OnInit, OnDestroy {
 
     public ngOnDestroy() {
         this._paramsSubscribers.forEach(x => x.unsubscribe());
+        this.jobData.dispose();
+        this.data.dispose();
     }
 
     @autobind()

app/services/core/rx-entity-proxy.ts

@@ -3,10 +3,16 @@ import { AsyncSubject, BehaviorSubject, Observable } from "rxjs";
 
 import { LoadingStatus } from "app/components/base/loading";
 import { ServerError } from "app/models";
+import { PollObservable } from "app/services/core";
 import { HttpCode } from "app/utils/constants";
 import { RxProxyBase, RxProxyBaseConfig } from "./rx-proxy-base";
 
 export interface RxEntityProxyConfig<TParams, TEntity> extends RxProxyBaseConfig<TParams, TEntity> {
+    /**
+     * If you want to have the entity proxy poll automatically for you every given milliseconds.
+     * @default Disabled
+     */
+    poll?: number;
 }
 
 export abstract class RxEntityProxy<TParams, TEntity> extends RxProxyBase<TParams, any, TEntity> {
@@ -17,6 +23,7 @@ export abstract class RxEntityProxy<TParams, TEntity> extends RxProxyBase<TParam
     public item: Observable<TEntity>;
 
     private _itemKey = new BehaviorSubject<string>(null);
+    private _pollTracker: PollObservable;
 
     /**
      * @param _type Class for TEntity used to instantiate
@@ -27,11 +34,15 @@ export abstract class RxEntityProxy<TParams, TEntity> extends RxProxyBase<TParam
     constructor(type: Type<TEntity>, config: RxEntityProxyConfig<TParams, TEntity>) {
         super(type, config);
         this.params = config.initialParams || {} as TParams;
-        this.item = this._itemKey.map((key) => {
+        this.item = this._itemKey.distinctUntilChanged().map((key) => {
             return this.cache.items.map((items) => {
                 return items.get(key);
             });
-        }).switch();
+        }).switch().distinctUntilChanged().takeUntil(this.isDisposed);
+
+        if (config.poll) {
+            this._pollTracker = this.startPoll(5000);
+        }
     }
 
     /**
@@ -59,10 +70,31 @@ export abstract class RxEntityProxy<TParams, TEntity> extends RxProxyBase<TParam
         });
     }
 
+    /**
+     * @see RxProxyBase#dispose()
+     */
+    public dispose() {
+        super.dispose();
+        this._itemKey.complete();
+        this.stopPoll();
+    }
+
     public refresh(): Observable<any> {
         return this.fetch();
     }
 
+    /**
+     * Stop the automatically started poll if applicable
+     */
+    public stopPoll() {
+        if (this._pollTracker) {
+            this._pollTracker.destroy();
+        }
+    }
+
+    /**
+     * Abstract method implementation of what to do when the polling calls.
+     */
     protected pollRefresh() {
         return this.refresh();
     }
@@ -83,22 +115,38 @@ export abstract class RxEntityProxy<TParams, TEntity> extends RxProxyBase<TParam
 
 export function getOnceProxy<TEntity>(getProxy: RxEntityProxy<any, TEntity>): Observable<TEntity> {
     const obs = new AsyncSubject<TEntity>();
+    getProxy.stopPoll();
 
     getProxy.fetch().subscribe({
         next: () => {
-            getProxy.item.first().subscribe((item: TEntity) => {
-                if (item) {
-                    obs.next(item);
-                    obs.complete();
-                    getProxy.dispose();
-                }
+            getProxy.item.subscribe((item: TEntity) => {
+                obs.next(item);
+                obs.complete();
+                getProxy.dispose();
             });
         },
         error: (e) => {
             obs.error(e);
             obs.complete();
+            getProxy.dispose();
         },
     });
 
     return obs.asObservable();
 }
+
+// const sub = new BehaviorSubject(0);
+// const until = new AsyncSubject();
+// let i = 0;
+// setInterval(() => {
+//     sub.next(i++);
+//     if (i === 4) {
+//         until.next(true);
+//         until.complete();
+//     }
+// }, 2000);
+
+// const obs = sub.takeUntil(until);
+// obs.subscribe((x) => {
+//     console.log("New value", x);
+// });

app/services/core/rx-list-proxy.ts

@@ -5,7 +5,6 @@ import { AsyncSubject, BehaviorSubject, Observable } from "rxjs";
 import { LoadingStatus } from "app/components/base/loading";
 import { log } from "app/utils";
 import { ListOptions, ListOptionsAttributes } from "./list-options";
-import { RxEntityProxy } from "./rx-entity-proxy";
 import { RxProxyBase, RxProxyBaseConfig } from "./rx-proxy-base";
 
 export interface RxListProxyConfig<TParams, TEntity> extends RxProxyBaseConfig<TParams, TEntity> {
@@ -129,18 +128,16 @@ export abstract class RxListProxy<TParams, TEntity> extends RxProxyBase<TParams,
      * It should not add the new item if already present.
      * The cache system will handle updating it already.
      */
-    public loadNewItem(entityProxy: RxEntityProxy<any, TEntity>): Observable<any> {
-        const obs = entityProxy.fetch().flatMap(() => entityProxy.item.first());
-        obs.subscribe({
+    public loadNewItem(getOnceObs: Observable<TEntity>): Observable<any> {
+        getOnceObs.subscribe({
             next: (newItem) => {
                 this._addItemToList(newItem);
             }, error: (error) => {
                 log.error("Error loading new item into RxListProxy",
                     { error, params: this._params, options: this._options });
             },
         });
-
-        return obs;
+        return getOnceObs;
     }
 
     /**

app/services/core/rx-proxy-base.ts

@@ -1,5 +1,5 @@
 import { Type } from "@angular/core";
-import { BehaviorSubject, Observable, Subject, Subscription } from "rxjs";
+import { AsyncSubject, BehaviorSubject, Observable, Subject, Subscription } from "rxjs";
 
 import { LoadingStatus } from "app/components/base/loading";
 import { ServerError } from "app/models";
@@ -62,6 +62,11 @@ export abstract class RxProxyBase<TParams, TOptions extends ProxyOptions, TEntit
      */
     public newDataStatus: Observable<LoadingStatus>;
 
+    /**
+     * Sets to   after calling dispose()
+     */
+    public isDisposed: AsyncSubject<boolean>;
+
     protected _status = new BehaviorSubject<LoadingStatus>(LoadingStatus.Loading);
     protected _newDataStatus = new BehaviorSubject<LoadingStatus>(LoadingStatus.Loading);
     protected _error = new BehaviorSubject<ServerError>(null);
@@ -86,6 +91,7 @@ export abstract class RxProxyBase<TParams, TOptions extends ProxyOptions, TEntit
         this.status = this._status.asObservable();
         this.newDataStatus = this._newDataStatus.asObservable();
         this.error = this._error.asObservable();
+        this.isDisposed = new AsyncSubject();
 
         this.status.subscribe((status) => {
             if (status === LoadingStatus.Loading) {
@@ -163,7 +169,12 @@ export abstract class RxProxyBase<TParams, TOptions extends ProxyOptions, TEntit
      * otherwise internal subscribe will never get cleared and the list porxy will not get GC
      */
     public dispose() {
+        this.isDisposed.next(true);
+        this.isDisposed.complete();
         this._clearDeleteSub();
+        this._deleted.complete();
+        this._status.complete();
+        this._error.complete();
     }
 
     protected set cache(cache: DataCache<TEntity>) {
@@ -265,7 +276,7 @@ export abstract class RxProxyBase<TParams, TOptions extends ProxyOptions, TEntit
 
     private _key() {
         const paramsKey = ObjectUtils.serialize(this._params);
-        const optionsKey = ObjectUtils.serialize(this._options.original);
+        const optionsKey = this._options && ObjectUtils.serialize(this._options.original);
         return `${paramsKey}|${optionsKey}`;
     }
 }