Merge branch 'w2p-111638_Process-admin-UI-redesign_Overview-page-tables_PR-feedback' into process-admin-ui-redesign-8.0.0-next

Author: AAwouters

src/app/collection-page/edit-collection-page/collection-source/collection-source-controls/collection-source-controls.component.ts

@@ -1,4 +1,4 @@
-import { Component, Input, OnDestroy } from '@angular/core';
+import { Component, Input, OnDestroy, OnInit } from '@angular/core';
 import { ScriptDataService } from '../../../../core/data/processes/script-data.service';
 import { ContentSource } from '../../../../core/shared/content-source.model';
 import { ProcessDataService } from '../../../../core/data/processes/process-data.service';
@@ -29,7 +29,7 @@ import { ContentSourceSetSerializer } from '../../../../core/shared/content-sour
   styleUrls: ['./collection-source-controls.component.scss'],
   templateUrl: './collection-source-controls.component.html',
 })
-export class CollectionSourceControlsComponent implements OnDestroy {
+export class CollectionSourceControlsComponent implements OnInit, OnDestroy {
 
   /**
    * Should the controls be enabled.
@@ -48,6 +48,7 @@ export class CollectionSourceControlsComponent implements OnDestroy {
 
   contentSource$: Observable<ContentSource>;
   private subs: Subscription[] = [];
+  private autoRefreshIDs: string[] = [];
 
   testConfigRunning$ = new BehaviorSubject(false);
   importRunning$ = new BehaviorSubject(false);
@@ -94,7 +95,10 @@ export class CollectionSourceControlsComponent implements OnDestroy {
       }),
       // filter out responses that aren't successful since the pinging of the process only needs to happen when the invocation was successful.
       filter((rd) => rd.hasSucceeded && hasValue(rd.payload)),
-      switchMap((rd) => this.processDataService.autoRefreshUntilCompletion(rd.payload.processId)),
+      switchMap((rd) => {
+        this.autoRefreshIDs.push(rd.payload.processId);
+        return this.processDataService.autoRefreshUntilCompletion(rd.payload.processId);
+      }),
       map((rd) => rd.payload)
     ).subscribe((process: Process) => {
       if (process.processStatus.toString() === ProcessStatus[ProcessStatus.FAILED].toString()) {
@@ -135,7 +139,10 @@ export class CollectionSourceControlsComponent implements OnDestroy {
           }
         }),
         filter((rd) => rd.hasSucceeded && hasValue(rd.payload)),
-        switchMap((rd) => this.processDataService.autoRefreshUntilCompletion(rd.payload.processId)),
+        switchMap((rd) => {
+          this.autoRefreshIDs.push(rd.payload.processId);
+          return this.processDataService.autoRefreshUntilCompletion(rd.payload.processId);
+        }),
         map((rd) => rd.payload)
       ).subscribe((process) => {
         if (process.processStatus.toString() === ProcessStatus[ProcessStatus.FAILED].toString()) {
@@ -170,7 +177,10 @@ export class CollectionSourceControlsComponent implements OnDestroy {
           }
         }),
         filter((rd) => rd.hasSucceeded && hasValue(rd.payload)),
-        switchMap((rd) => this.processDataService.autoRefreshUntilCompletion(rd.payload.processId)),
+        switchMap((rd) => {
+          this.autoRefreshIDs.push(rd.payload.processId);
+          return this.processDataService.autoRefreshUntilCompletion(rd.payload.processId);
+        }),
         map((rd) => rd.payload)
       ).subscribe((process) => {
         if (process.processStatus.toString() === ProcessStatus[ProcessStatus.FAILED].toString()) {
@@ -191,5 +201,9 @@ export class CollectionSourceControlsComponent implements OnDestroy {
         sub.unsubscribe();
       }
     });
+
+    this.autoRefreshIDs.forEach((id) => {
+      this.processDataService.stopAutoRefreshing(id);
+    });
   }
 }

src/app/core/data/processes/process-data.service.spec.ts

@@ -9,7 +9,7 @@
 import { testFindAllDataImplementation } from '../base/find-all-data.spec';
 import { ProcessDataService, TIMER_FACTORY } from './process-data.service';
 import { testDeleteDataImplementation } from '../base/delete-data.spec';
-import { waitForAsync, TestBed, fakeAsync, tick, flush } from '@angular/core/testing';
+import { waitForAsync, TestBed, fakeAsync, tick } from '@angular/core/testing';
 import { RequestService } from '../request.service';
 import { RemoteData } from '../remote-data';
 import { RequestEntryState } from '../request-entry-state.model';
@@ -27,6 +27,7 @@ import { testSearchDataImplementation } from '../base/search-data.spec';
 import { PaginatedList } from '../paginated-list.model';
 import { FindListOptions } from '../find-list-options.model';
 import { of } from 'rxjs';
+import { getMockRequestService } from '../../../shared/mocks/request.service.mock';
 
 describe('ProcessDataService', () => {
   let testScheduler;
@@ -43,7 +44,7 @@ describe('ProcessDataService', () => {
     testSearchDataImplementation(initService);
   });
 
-  let requestService;
+  let requestService = getMockRequestService();
   let processDataService;
   let remoteDataBuildService;
 
@@ -136,7 +137,7 @@ describe('ProcessDataService', () => {
         imports: [],
         providers: [
           ProcessDataService,
-          { provide: RequestService, useValue: null },
+          { provide: RequestService, useValue: requestService },
           { provide: RemoteDataBuildService, useValue: null },
           { provide: ObjectCacheService, useValue: null },
           { provide: ReducerManager, useValue: null },
@@ -152,41 +153,41 @@ describe('ProcessDataService', () => {
     }));
 
     it('should refresh after the specified interval', fakeAsync(() => {
-        const runningProcess = Object.assign(new Process(), {
-          _links: {
-            self: {
-              href: 'https://rest.api/processes/123'
-            }
+      const runningProcess = Object.assign(new Process(), {
+        _links: {
+          self: {
+            href: 'https://rest.api/processes/123'
           }
-        });
-        runningProcess.processStatus = ProcessStatus.RUNNING;
+        }
+      });
+      runningProcess.processStatus = ProcessStatus.RUNNING;
 
-        const runningProcessPagination: PaginatedList<Process> = Object.assign(new PaginatedList(), {
-          page: [runningProcess],
-          _links: {
-            self: {
-              href: 'https://rest.api/processesList/456'
-            }
+      const runningProcessPagination: PaginatedList<Process> = Object.assign(new PaginatedList(), {
+        page: [runningProcess],
+        _links: {
+          self: {
+            href: 'https://rest.api/processesList/456'
           }
-        });
+        }
+      });
 
-        const runningProcessRD = new RemoteData(0, 0, 0, RequestEntryState.Success, null, runningProcessPagination);
+      const runningProcessRD = new RemoteData(0, 0, 0, RequestEntryState.Success, null, runningProcessPagination);
 
-        spyOn(processDataService, 'invalidateByHref');
-        spyOn(processDataService, 'searchBy').and.returnValue(
-          of(runningProcessRD)
-        );
+      spyOn(processDataService, 'searchBy').and.returnValue(
+        of(runningProcessRD)
+      );
+
+      expect(processDataService.searchBy).toHaveBeenCalledTimes(0);
+      expect(requestService.setStaleByHrefSubstring).toHaveBeenCalledTimes(0);
+
+      let sub = processDataService.autoRefreshingSearchBy('id', 'byProperty', new FindListOptions(), 200).subscribe();
+      expect(processDataService.searchBy).toHaveBeenCalledTimes(1);
 
-        let sub = processDataService.autoRefreshingSearchBy('byProperty', new FindListOptions(), 200).subscribe();
-        tick(0);
-        expect(processDataService.searchBy).toHaveBeenCalledTimes(1);
-        tick(450);
-        expect(processDataService.searchBy).toHaveBeenCalledTimes(3);
-        sub.unsubscribe();
+      tick(250);
 
-        flush();
+      expect(requestService.setStaleByHrefSubstring).toHaveBeenCalledTimes(1);
 
-        expect(processDataService.invalidateByHref).toHaveBeenCalledTimes(3);
+      sub.unsubscribe();
     }));
   });
 });

src/app/core/data/processes/process-data.service.ts

@@ -5,8 +5,8 @@ import { ObjectCacheService } from '../../cache/object-cache.service';
 import { HALEndpointService } from '../../shared/hal-endpoint.service';
 import { Process } from '../../../process-page/processes/process.model';
 import { PROCESS } from '../../../process-page/processes/process.resource-type';
-import { Observable, timer as rxjsTimer, concatMap } from 'rxjs';
-import { switchMap, filter, distinctUntilChanged, find, tap } from 'rxjs/operators';
+import { Observable, Subscription } from 'rxjs';
+import { switchMap, filter, distinctUntilChanged, find } from 'rxjs/operators';
 import { PaginatedList } from '../paginated-list.model';
 import { Bitstream } from '../../shared/bitstream.model';
 import { RemoteData } from '../remote-data';
@@ -19,7 +19,7 @@ import { dataService } from '../base/data-service.decorator';
 import { DeleteData, DeleteDataImpl } from '../base/delete-data';
 import { NotificationsService } from '../../../shared/notifications/notifications.service';
 import { NoContent } from '../../shared/NoContent.model';
-import { getAllCompletedRemoteData, getFirstCompletedRemoteData } from '../../shared/operators';
+import { getAllCompletedRemoteData } from '../../shared/operators';
 import { ProcessStatus } from 'src/app/process-page/processes/process-status.model';
 import { hasValue } from '../../../shared/empty.util';
 import { SearchData, SearchDataImpl } from '../base/search-data';
@@ -41,6 +41,7 @@ export class ProcessDataService extends IdentifiableDataService<Process> impleme
   private deleteData: DeleteData<Process>;
   private searchData: SearchData<Process>;
   protected activelyBeingPolled: Map<string, NodeJS.Timeout> = new Map();
+  protected subs: Map<string, Subscription> = new Map();
 
   constructor(
     protected requestService: RequestService,
@@ -129,6 +130,9 @@ export class ProcessDataService extends IdentifiableDataService<Process> impleme
   }
 
   /**
+   * @param id                          The id for this auto-refreshing search. Used to stop
+   *                                    auto-refreshing afterwards, and ensure we're not
+   *                                    auto-refreshing the same thing multiple times.
    * @param searchMethod                The search method for the Process
    * @param options                     The FindListOptions object
    * @param pollingIntervalInMs         The interval by which the search will be repeated
@@ -137,22 +141,41 @@ export class ProcessDataService extends IdentifiableDataService<Process> impleme
    * @return {Observable<RemoteData<PaginatedList<Process>>>}
    *    Return an observable that emits a paginated list of processes every interval
    */
-  autoRefreshingSearchBy(searchMethod: string, options?: FindListOptions, pollingIntervalInMs: number = 5000, ...linksToFollow: FollowLinkConfig<Process>[]): Observable<RemoteData<PaginatedList<Process>>> {
-    // Create observable that emits every pollingInterval
-    return rxjsTimer(0, pollingIntervalInMs).pipe(
-        concatMap(() => {
-          // Every time the timer emits, request the current state of the processes
-          return this.searchBy(searchMethod, options, false, false, ...linksToFollow).pipe(
-              getFirstCompletedRemoteData(),
-              tap((processListRD: RemoteData<PaginatedList<Process>>) => {
-                // Once the response has been received, invalidate the response right before the next request
-                setTimeout(() => {
-                  this.invalidateByHref(processListRD.payload._links.self.href);
-                }, Math.max(pollingIntervalInMs - 100, 0));
-              }),
-          );
-        })
+  autoRefreshingSearchBy(id: string, searchMethod: string, options?: FindListOptions, pollingIntervalInMs: number = 5000, ...linksToFollow: FollowLinkConfig<Process>[]): Observable<RemoteData<PaginatedList<Process>>> {
+
+    const result$ = this.searchBy(searchMethod, options, true, true, ...linksToFollow).pipe(
+      getAllCompletedRemoteData()
     );
+
+    const sub = result$.pipe(
+      filter(() =>
+        !this.activelyBeingPolled.has(id)
+      )
+    ).subscribe((processListRd: RemoteData<PaginatedList<Process>>) => {
+      this.clearCurrentTimeout(id);
+      const nextTimeout = this.timer(() => {
+        this.activelyBeingPolled.delete(id);
+        this.requestService.setStaleByHrefSubstring(processListRd.payload._links.self.href);
+      }, pollingIntervalInMs);
+
+      this.activelyBeingPolled.set(id, nextTimeout);
+    });
+
+    this.subs.set(id, sub);
+
+    return result$;
+  }
+
+  /**
+   * Stop auto-refreshing the request with the given id
+   * @param id the id of the request to stop automatically refreshing
+   */
+  stopAutoRefreshing(id: string) {
+    this.clearCurrentTimeout(id);
+    if (hasValue(this.subs.get(id))) {
+      this.subs.get(id).unsubscribe();
+      this.subs.delete(id);
+    }
   }
 
   /**
@@ -181,14 +204,15 @@ export class ProcessDataService extends IdentifiableDataService<Process> impleme
   }
 
   /**
-   * Clear the timeout for the given process, if that timeout exists
+   * Clear the timeout for the given id, if that timeout exists
    * @protected
    */
-  protected clearCurrentTimeout(processId: string): void {
-    const timeout = this.activelyBeingPolled.get(processId);
+  protected clearCurrentTimeout(id: string): void {
+    const timeout = this.activelyBeingPolled.get(id);
     if (hasValue(timeout)) {
       clearTimeout(timeout);
     }
+    this.activelyBeingPolled.delete(id);
   }
 
   /**
@@ -229,15 +253,15 @@ export class ProcessDataService extends IdentifiableDataService<Process> impleme
       this.activelyBeingPolled.set(processId, nextTimeout);
     });
 
+    this.subs.set(processId, sub);
+
     // When the process completes create a one off subscription (the `find` completes the
     // observable) that unsubscribes the previous one, removes the processId from the list of
     // processes being polled and clears any running timeouts
     process$.pipe(
       find((processRD: RemoteData<Process>) => ProcessDataService.hasCompletedOrFailed(processRD.payload))
     ).subscribe(() => {
-      this.clearCurrentTimeout(processId);
-      this.activelyBeingPolled.delete(processId);
-      sub.unsubscribe();
+      this.stopAutoRefreshing(processId);
     });
 
     return process$.pipe(

src/app/process-page/detail/process-detail.component.ts

@@ -1,5 +1,5 @@
 import { HttpClient } from '@angular/common/http';
-import { Component, Inject, NgZone, OnInit, PLATFORM_ID } from '@angular/core';
+import { Component, Inject, NgZone, OnInit, PLATFORM_ID, OnDestroy } from '@angular/core';
 import { ActivatedRoute, Router } from '@angular/router';
 import { BehaviorSubject, Observable } from 'rxjs';
 import { finalize, map, switchMap, take, tap, find, startWith } from 'rxjs/operators';
@@ -36,7 +36,7 @@ import { PROCESS_PAGE_FOLLOW_LINKS } from '../process-page.resolver';
 /**
  * A component displaying detailed information about a DSpace Process
  */
-export class ProcessDetailComponent implements OnInit {
+export class ProcessDetailComponent implements OnInit, OnDestroy {
 
   /**
    * The AlertType enumeration
@@ -80,6 +80,8 @@ export class ProcessDetailComponent implements OnInit {
 
   isRefreshing$: Observable<boolean>;
 
+  protected autoRefreshingID: string;
+
   /**
    * Reference to NgbModal
    */
@@ -108,7 +110,8 @@ export class ProcessDetailComponent implements OnInit {
     this.processRD$ = this.route.data.pipe(
       switchMap((data) => {
         if (isPlatformBrowser(this.platformId)) {
-          return this.processService.autoRefreshUntilCompletion(this.route.snapshot.params.id, 5000, ...PROCESS_PAGE_FOLLOW_LINKS);
+          this.autoRefreshingID = this.route.snapshot.params.id;
+          return this.processService.autoRefreshUntilCompletion(this.autoRefreshingID, 5000, ...PROCESS_PAGE_FOLLOW_LINKS);
         } else {
           return [data.process as RemoteData<Process>];
         }
@@ -128,6 +131,15 @@ export class ProcessDetailComponent implements OnInit {
     );
   }
 
+  /**
+   * Make sure the autoRefreshUntilCompletion is cleaned up properly
+   */
+  ngOnDestroy() {
+    if (hasValue(this.autoRefreshingID)) {
+      this.processService.stopAutoRefreshing(this.autoRefreshingID);
+    }
+  }
+
   /**
    * Get the name of a bitstream
    * @param bitstream

src/app/process-page/overview/process-overview.service.ts

@@ -62,12 +62,21 @@ export class ProcessOverviewService {
     }, findListOptions);
 
     if (hasValue(autoRefreshingIntervalInMs) && autoRefreshingIntervalInMs > 0) {
-      return this.processDataService.autoRefreshingSearchBy('byProperty', options, autoRefreshingIntervalInMs);
+      this.processDataService.stopAutoRefreshing(processStatus);
+      return this.processDataService.autoRefreshingSearchBy(processStatus, 'byProperty', options, autoRefreshingIntervalInMs);
     } else {
       return this.processDataService.searchBy('byProperty', options);
     }
   }
 
+  /**
+   * Stop auto-refreshing the process with the given status
+   * @param processStatus the processStatus of the request to stop automatically refreshing
+   */
+  stopAutoRefreshing(processStatus: ProcessStatus) {
+    this.processDataService.stopAutoRefreshing(processStatus);
+  }
+
   /**
    * Map the provided paginationOptions to FindListOptions
    * @param paginationOptions the PaginationComponentOptions to map