NIFI-13068: (#8668)

- Adding support to terminate threads for Processors that are no longer running.
- Updating import for concatLatestFrom.

This closes #8668

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/access-policies/state/access-policy/access-policy.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { NiFiState } from '../../../../state';
 import { Store } from '@ngrx/store';
 import { Router } from '@angular/router';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/bulletins/state/bulletin-board/bulletin-board.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { Store } from '@ngrx/store';
 import { NiFiState } from '../../../../state';
 import * as BulletinBoardActions from './bulletin-board.actions';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/counters/state/counter-listing/counter-listing.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { NiFiState } from '../../../../state';
 import { Store } from '@ngrx/store';
 import * as CounterListingActions from './counter-listing.actions';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-configuration-history/state/flow-configuration-history-listing/flow-configuration-history-listing.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { Store } from '@ngrx/store';
 import { NiFiState } from '../../../../state';
 import { ErrorHelper } from '../../../../service/error-helper.service';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/service/canvas-context-menu.service.ts

@@ -56,7 +56,8 @@ import {
     stopCurrentProcessGroup,
     stopVersionControlRequest,
     copy,
-    paste
+    paste,
+    terminateThreads
 } from '../state/flow/flow.actions';
 import { ComponentType } from '../../../state/shared';
 import {
@@ -639,14 +640,21 @@ export class CanvasContextMenu implements ContextMenuDefinitionProvider {
                 }
             },
             {
-                condition: (selection: any) => {
-                    // TODO - canTerminate
-                    return false;
+                condition: (selection: d3.Selection<any, any, any, any>) => {
+                    return this.canvasUtils.canTerminate(selection);
                 },
                 clazz: 'fa fa-hourglass-end',
                 text: 'Terminate',
-                action: () => {
-                    // TODO - terminate
+                action: (selection: d3.Selection<any, any, any, any>) => {
+                    const d: any = selection.datum();
+                    this.store.dispatch(
+                        terminateThreads({
+                            request: {
+                                id: d.id,
+                                uri: d.uri
+                            }
+                        })
+                    );
                 }
             },
             {

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/service/canvas-utils.service.ts

@@ -1611,6 +1611,30 @@ export class CanvasUtils {
         return selectionSize === writableSize;
     }
 
+    /**
+     * Determines whether the selection represents a Processor that is no longer running but
+     * still has active threads.
+     *
+     * @param selection
+     */
+    public canTerminate(selection: d3.Selection<any, any, any, any>): boolean {
+        if (selection.size() !== 1) {
+            return false;
+        }
+
+        if (!this.canOperate(selection)) {
+            return false;
+        }
+
+        if (this.isProcessor(selection)) {
+            const selectionData = selection.datum();
+            const aggregateSnapshot = selectionData.status.aggregateSnapshot;
+            return aggregateSnapshot.runStatus !== 'Running' && aggregateSnapshot.activeThreadCount > 0;
+        }
+
+        return false;
+    }
+
     /**
      * Determines whether the current selection supports starting flow versioning.
      *

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/service/flow.service.ts

@@ -40,6 +40,7 @@ import {
     StopComponentRequest,
     StopProcessGroupRequest,
     StopVersionControlRequest,
+    TerminateThreadsRequest,
     UpdateComponentRequest,
     UploadProcessGroupRequest,
     VersionControlInformationEntity
@@ -287,6 +288,10 @@ export class FlowService implements PropertyDescriptorRetriever {
         return this.httpClient.put(`${this.nifiCommon.stripProtocol(request.uri)}/run-status`, stopRequest);
     }
 
+    terminateThreads(request: TerminateThreadsRequest): Observable<any> {
+        return this.httpClient.delete(`${this.nifiCommon.stripProtocol(request.uri)}/threads`);
+    }
+
     startProcessGroup(request: StartProcessGroupRequest): Observable<any> {
         const startRequest: ProcessGroupRunStatusRequest = {
             id: request.id,

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/state/controller-services/controller-services.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as ControllerServicesActions from './controller-services.actions';
 import { catchError, combineLatest, from, map, of, switchMap, take, takeUntil, tap } from 'rxjs';
 import { MatDialog } from '@angular/material/dialog';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/state/flow/flow.actions.ts

@@ -88,6 +88,7 @@ import {
     StopProcessGroupResponse,
     StopVersionControlRequest,
     StopVersionControlResponse,
+    TerminateThreadsRequest,
     UpdateComponentFailure,
     UpdateComponentRequest,
     UpdateComponentResponse,
@@ -616,6 +617,11 @@ export const stopComponents = createAction(
     props<{ request: StopComponentsRequest }>()
 );
 
+export const terminateThreads = createAction(
+    `${CANVAS_PREFIX} Terminate Threads`,
+    props<{ request: TerminateThreadsRequest }>()
+);
+
 export const stopComponentSuccess = createAction(
     `${CANVAS_PREFIX} Stop Component Success`,
     props<{ response: StopComponentResponse }>()

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/state/flow/flow.effects.ts

@@ -17,7 +17,8 @@
 
 import { Injectable } from '@angular/core';
 import { FlowService } from '../../service/flow.service';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as FlowActions from './flow.actions';
 import * as StatusHistoryActions from '../../../../state/status-history/status-history.actions';
 import * as ErrorActions from '../../../../state/error/error.actions';
@@ -2582,6 +2583,36 @@ export class FlowEffects {
         )
     );
 
+    /**
+     * Terminates threads for the processor in the request.
+     */
+    terminateThreads$ = createEffect(() =>
+        this.actions$.pipe(
+            ofType(FlowActions.terminateThreads),
+            map((action) => action.request),
+            switchMap((request) =>
+                from(this.flowService.terminateThreads(request)).pipe(
+                    map((response) =>
+                        FlowActions.updateComponentSuccess({
+                            response: {
+                                id: request.id,
+                                type: ComponentType.Processor,
+                                response
+                            }
+                        })
+                    ),
+                    catchError((errorResponse: HttpErrorResponse) =>
+                        of(
+                            FlowActions.flowSnackbarError({
+                                error: errorResponse.error
+                            })
+                        )
+                    )
+                )
+            )
+        )
+    );
+
     /**
      * If the component stopped was the current process group, reload the flow
      */

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/state/flow/index.ts

@@ -724,6 +724,11 @@ export interface StopComponentsRequest {
     components: StopComponentRequest[];
 }
 
+export interface TerminateThreadsRequest {
+    id: string;
+    uri: string;
+}
+
 export interface LoadChildProcessGroupRequest {
     id: string;
 }

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/state/manage-remote-ports/manage-remote-ports.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as ManageRemotePortsActions from './manage-remote-ports.actions';
 import { catchError, from, map, of, switchMap, tap } from 'rxjs';
 import { MatDialog } from '@angular/material/dialog';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/state/parameter/parameter.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as ParameterActions from './parameter.actions';
 import { Store } from '@ngrx/store';
 import { CanvasState } from '../index';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/state/queue/queue.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as QueueActions from './queue.actions';
 import { Store } from '@ngrx/store';
 import { asyncScheduler, catchError, filter, from, interval, map, of, switchMap, take, takeUntil, tap } from 'rxjs';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/state/transform/transform.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as TransformActions from './transform.actions';
 import { map, tap } from 'rxjs';
 import { Storage } from '../../../../service/storage.service';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/ui/canvas/canvas.component.ts

@@ -64,7 +64,7 @@ import { initialState } from '../../state/flow/flow.reducer';
 import { CanvasContextMenu } from '../../service/canvas-context-menu.service';
 import { getStatusHistoryAndOpenDialog } from '../../../../state/status-history/status-history.actions';
 import { loadFlowConfiguration } from '../../../../state/flow-configuration/flow-configuration.actions';
-import { concatLatestFrom } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { selectUrl } from '../../../../state/router/router.selectors';
 import { Storage } from '../../../../service/storage.service';
 import {

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/flow-designer/ui/manage-remote-ports/manage-remote-ports.component.ts

@@ -46,7 +46,7 @@ import { NiFiCommon } from '../../../../service/nifi-common.service';
 import { MatTableDataSource } from '@angular/material/table';
 import { Sort } from '@angular/material/sort';
 import { TextTip } from '../../../../ui/common/tooltips/text-tip/text-tip.component';
-import { concatLatestFrom } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { loadFlowConfiguration } from '../../../../state/flow-configuration/flow-configuration.actions';
 import {
     selectFlowConfiguration,

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/parameter-contexts/state/parameter-context-listing/parameter-context-listing.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as ParameterContextListingActions from './parameter-context-listing.actions';
 import * as ErrorActions from '../../../../state/error/error.actions';
 import {

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/provenance/state/lineage/lineage.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as LineageActions from './lineage.actions';
 import { asyncScheduler, catchError, filter, from, interval, map, of, switchMap, takeUntil, tap } from 'rxjs';
 import { MatDialog } from '@angular/material/dialog';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/provenance/state/provenance-event-listing/provenance-event-listing.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as ProvenanceEventListingActions from './provenance-event-listing.actions';
 import { asyncScheduler, catchError, filter, from, interval, map, of, switchMap, take, takeUntil, tap } from 'rxjs';
 import { MatDialog } from '@angular/material/dialog';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/queue/state/queue-listing/queue-listing.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as QueueListingActions from './queue-listing.actions';
 import { Store } from '@ngrx/store';
 import { CanvasState } from '../../../flow-designer/state';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/settings/state/flow-analysis-rules/flow-analysis-rules.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as FlowAnalysisRuleActions from './flow-analysis-rules.actions';
 import { catchError, from, map, of, switchMap, take, takeUntil, tap } from 'rxjs';
 import { MatDialog } from '@angular/material/dialog';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/settings/state/general/general.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as GeneralActions from './general.actions';
 import * as ErrorActions from '../../../../state/error/error.actions';
 import { catchError, from, map, of, switchMap, tap } from 'rxjs';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/settings/state/management-controller-services/management-controller-services.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as ManagementControllerServicesActions from './management-controller-services.actions';
 import * as ErrorActions from '../../../../state/error/error.actions';
 import { catchError, from, map, of, switchMap, take, takeUntil, tap } from 'rxjs';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/settings/state/parameter-providers/parameter-providers.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { Store } from '@ngrx/store';
 import { NiFiState } from '../../../../state';
 import { Client } from '../../../../service/client.service';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/settings/state/registry-clients/registry-clients.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as RegistryClientsActions from './registry-clients.actions';
 import { catchError, from, map, of, switchMap, take, takeUntil, tap } from 'rxjs';
 import { MatDialog } from '@angular/material/dialog';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/settings/state/reporting-tasks/reporting-tasks.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as ReportingTaskActions from './reporting-tasks.actions';
 import { catchError, from, map, of, switchMap, take, takeUntil, tap } from 'rxjs';
 import { MatDialog } from '@angular/material/dialog';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/summary/state/component-cluster-status/component-cluster-status.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { NiFiState } from '../../../../state';
 import { Store } from '@ngrx/store';
 import { ErrorHelper } from '../../../../service/error-helper.service';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/summary/state/summary-listing/summary-listing.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { Store } from '@ngrx/store';
 import { NiFiState } from '../../../../state';
 import { ProcessGroupStatusService } from '../../service/process-group-status.service';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/pages/users/state/user-listing/user-listing.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { NiFiState } from '../../../../state';
 import { Store } from '@ngrx/store';
 import { Router } from '@angular/router';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/state/cluster-summary/cluster-summary.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as ClusterSummaryActions from './cluster-summary.actions';
 import { asyncScheduler, catchError, delay, filter, from, interval, map, of, switchMap, takeUntil, tap } from 'rxjs';
 import { ClusterService } from '../../service/cluster.service';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/state/component-state/component-state.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import { Store } from '@ngrx/store';
 import { NiFiState } from '../index';
 import * as ComponentStateActions from './component-state.actions';

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-frontend/src/main/nifi/src/app/state/contoller-service-state/controller-service-state.effects.ts

@@ -16,7 +16,8 @@
  */
 
 import { Injectable } from '@angular/core';
-import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects';
+import { Actions, createEffect, ofType } from '@ngrx/effects';
+import { concatLatestFrom } from '@ngrx/operators';
 import * as ControllerServiceActions from './controller-service-state.actions';
 import { asyncScheduler, catchError, filter, from, interval, map, of, switchMap, takeUntil, tap } from 'rxjs';
 import { Store } from '@ngrx/store';