This is an automated email from the ASF dual-hosted git repository.
rfellows pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 2ea99bec1a3 NIFI-15937: Adding support for polling in the connector
canvas. (#11246)
2ea99bec1a3 is described below
commit 2ea99bec1a3b753229a02959e59e5e0c0a138a18
Author: Matt Gilman <[email protected]>
AuthorDate: Wed May 13 14:19:00 2026 -0400
NIFI-15937: Adding support for polling in the connector canvas. (#11246)
---
.../connector-canvas/connector-canvas.actions.ts | 28 ++
.../connector-canvas.effects.spec.ts | 376 ++++++++++++++++++++-
.../connector-canvas/connector-canvas.effects.ts | 105 +++++-
.../connector-canvas.component.spec.ts | 199 +++++++++++
.../connector-canvas/connector-canvas.component.ts | 5 +
5 files changed, 708 insertions(+), 5 deletions(-)
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
index 0a580dbdabb..23ba9fb5f2e 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
@@ -65,6 +65,34 @@ export const loadConnectorFlowFailure = createAction(
export const loadConnectorFlowComplete = createAction('[Connector Canvas] Load
Connector Flow Complete');
+/**
+ * Reload the currently displayed connector flow.
+ *
+ * The polling effect dispatches this action; the corresponding effect
throttles
+ * and resolves the connector and process group identifiers from state before
+ * delegating to {@link loadConnectorFlow}. Keeping the indirection makes the
+ * polling source independent of which connector and process group are
currently
+ * mounted, mirroring the flow-designer's reloadFlow pattern.
+ */
+export const reloadConnectorFlow = createAction('[Connector Canvas] Reload
Connector Flow');
+
+/**
+ * Begin periodic refresh of the connector flow on the currently mounted
canvas.
+ *
+ * The component dispatches this action in ngOnInit; the polling effect drives
a
+ * 30 second interval, gates on document visibility and on the absence of an
+ * in-flight load, and dispatches {@link reloadConnectorFlow}.
+ */
+export const startConnectorCanvasPolling = createAction('[Connector Canvas]
Start Connector Canvas Polling');
+
+/**
+ * Stop the periodic refresh started by {@link startConnectorCanvasPolling}.
+ *
+ * Dispatched by the component in ngOnDestroy (and may be dispatched explicitly
+ * by other effects that need to suppress polling around a destructive
operation).
+ */
+export const stopConnectorCanvasPolling = createAction('[Connector Canvas]
Stop Connector Canvas Polling');
+
export const enterProcessGroup = createAction(
'[Connector Canvas] Enter Process Group',
props<{ request: { id: string } }>()
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
index 04418e70862..797cccb38c4 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
@@ -41,16 +41,23 @@ import {
navigateToProvenanceForComponent,
navigateToQueueListing,
navigateWithoutTransform,
+ reloadConnectorFlow,
selectComponents,
+ startConnectorCanvasPolling,
+ stopConnectorCanvasPolling,
viewComponentConfiguration
} from './connector-canvas.actions';
import { queueEmptied } from
'../../../../state/empty-queue/empty-queue.actions';
import {
+ selectConnectorId,
selectConnectorIdFromRoute,
+ selectLoadingStatus,
selectParentProcessGroupId,
selectProcessGroupId,
selectProcessGroupIdFromRoute
} from './connector-canvas.selectors';
+import { selectDocumentVisibilityState } from
'../../../../state/document-visibility/document-visibility.selectors';
+import { DocumentVisibility } from '../../../../state/document-visibility';
import type { Mock } from 'vitest';
describe('ConnectorCanvasEffects', () => {
@@ -61,6 +68,8 @@ describe('ConnectorCanvasEffects', () => {
parentProcessGroupId?: string | null;
processGroupId?: string | null;
processGroupIdFromRoute?: string | null;
+ documentVisibility?: DocumentVisibility;
+ loadingStatus?: 'pending' | 'loading' | 'success' | 'error';
} = {}
) {
let actions$: Observable<Action>;
@@ -71,6 +80,8 @@ describe('ConnectorCanvasEffects', () => {
const processGroupId = options.processGroupId !== undefined ?
options.processGroupId : 'child-pg';
const processGroupIdFromRoute =
options.processGroupIdFromRoute !== undefined ?
options.processGroupIdFromRoute : processGroupId;
+ const documentVisibility = options.documentVisibility ??
DocumentVisibility.Visible;
+ const loadingStatus = options.loadingStatus ?? 'success';
// Mock services
const mockConnectorService = {
@@ -98,9 +109,15 @@ describe('ConnectorCanvasEffects', () => {
initialState: {},
selectors: [
{ selector: selectConnectorIdFromRoute, value:
connectorId },
+ { selector: selectConnectorId, value: connectorId },
{ selector: selectParentProcessGroupId, value:
parentProcessGroupId },
{ selector: selectProcessGroupId, value:
processGroupId },
- { selector: selectProcessGroupIdFromRoute, value:
processGroupIdFromRoute }
+ { selector: selectProcessGroupIdFromRoute, value:
processGroupIdFromRoute },
+ { selector: selectLoadingStatus, value: loadingStatus
},
+ {
+ selector: selectDocumentVisibilityState,
+ value: { documentVisibility, changedTimestamp: 0 }
+ }
]
}),
{ provide: ConnectorService, useValue: mockConnectorService },
@@ -537,6 +554,19 @@ describe('ConnectorCanvasEffects', () => {
const { mockDialog } = await dispatchView(ComponentType.Funnel);
expect(mockDialog.open).not.toHaveBeenCalled();
});
+
+ it('handles entities without operatePermissions by forcing
operatePermissions.canWrite to false', async () => {
+ const entityWithoutOperatePermissions = {
+ id: 'comp-1',
+ uri: 'https://localhost/nifi-api/processors/comp-1',
+ permissions: { canRead: true, canWrite: true },
+ component: { name: 'My Component' }
+ };
+ const { mockDialog } = await dispatchView(ComponentType.Processor,
entityWithoutOperatePermissions);
+ const [, config] = (mockDialog.open as Mock).mock.calls[0];
+ expect(config.data.entity.operatePermissions).toBeDefined();
+ expect(config.data.entity.operatePermissions.canWrite).toBe(false);
+ });
});
describe('navigateToProvenanceForComponent$', () => {
@@ -567,6 +597,34 @@ describe('ConnectorCanvasEffects', () => {
}
});
});
+
+ it('should use the supplied component type in the back navigation
route for a funnel', async () => {
+ const { effects, actions$, mockRouter } = await setup({
+ connectorId: 'conn-1',
+ processGroupIdFromRoute: 'pg-root'
+ });
+ actions$(
+ of(
+ navigateToProvenanceForComponent({
+ id: 'funnel-1',
+ componentType: ComponentType.Funnel
+ })
+ )
+ );
+
+ await firstValueFrom(effects.navigateToProvenanceForComponent$);
+
+ expect(mockRouter.navigate).toHaveBeenCalledWith(['/provenance'], {
+ queryParams: { componentId: 'funnel-1' },
+ state: {
+ backNavigation: {
+ route: ['/connectors', 'conn-1', 'canvas', 'pg-root',
ComponentType.Funnel, 'funnel-1'],
+ routeBoundary: ['/provenance'],
+ context: 'Funnel'
+ }
+ }
+ });
+ });
});
describe('navigateToQueueListing$', () => {
@@ -681,6 +739,31 @@ describe('ConnectorCanvasEffects', () => {
);
});
+ it('should dispatch loadConnectorFlow when all queues are emptied for
a process group with connector-canvas source', async () => {
+ const { effects, actions$ } = await setup({
+ connectorId: 'conn-1',
+ processGroupIdFromRoute: 'pg-root'
+ });
+ actions$(
+ of(
+ queueEmptied({
+ connectionId: null,
+ processGroupId: 'pg-root',
+ source: 'connector-canvas'
+ })
+ )
+ );
+
+ const result = await
firstValueFrom(effects.refreshAfterQueueEmptied$);
+
+ expect(result).toEqual(
+ loadConnectorFlow({
+ connectorId: 'conn-1',
+ processGroupId: 'pg-root'
+ })
+ );
+ });
+
it('should ignore queueEmptied events from the flow designer', async
() => {
const { effects, actions$ } = await setup({
connectorId: 'conn-1',
@@ -705,5 +788,296 @@ describe('ConnectorCanvasEffects', () => {
expect(emitted).toBeUndefined();
});
+
+ it('should not dispatch when the connectorId route param is missing',
async () => {
+ const { effects, actions$ } = await setup({
+ connectorId: null,
+ processGroupIdFromRoute: 'pg-root'
+ });
+ actions$(
+ of(
+ queueEmptied({
+ connectionId: 'conn-listing-1',
+ processGroupId: null,
+ source: 'connector-canvas'
+ })
+ )
+ );
+
+ let emitted: Action | undefined;
+ effects.refreshAfterQueueEmptied$.subscribe((action) => {
+ emitted = action;
+ });
+
+ await new Promise<void>((resolve) => setTimeout(resolve, 0));
+
+ expect(emitted).toBeUndefined();
+ });
+ });
+
+ describe('reloadConnectorFlow$', () => {
+ it('should dispatch loadConnectorFlow with the connector and process
group ids from state', async () => {
+ const { effects, actions$ } = await setup({
+ connectorId: 'conn-state',
+ processGroupId: 'pg-state'
+ });
+ actions$(of(reloadConnectorFlow()));
+
+ const action = await firstValueFrom(effects.reloadConnectorFlow$);
+ expect(action).toEqual(
+ loadConnectorFlow({
+ connectorId: 'conn-state',
+ processGroupId: 'pg-state'
+ })
+ );
+ });
+
+ it('should not dispatch loadConnectorFlow when the connector id is
empty', async () => {
+ const { effects, actions$ } = await setup({
+ connectorId: '',
+ processGroupId: 'pg-state'
+ });
+ actions$(of(reloadConnectorFlow()));
+
+ let emitted: Action | undefined;
+ effects.reloadConnectorFlow$.subscribe((action) => {
+ emitted = action;
+ });
+
+ await new Promise<void>((resolve) => setTimeout(resolve, 0));
+
+ expect(emitted).toBeUndefined();
+ });
+
+ it('should not dispatch loadConnectorFlow when the process group id is
null', async () => {
+ const { effects, actions$ } = await setup({
+ connectorId: 'conn-state',
+ processGroupId: null
+ });
+ actions$(of(reloadConnectorFlow()));
+
+ let emitted: Action | undefined;
+ effects.reloadConnectorFlow$.subscribe((action) => {
+ emitted = action;
+ });
+
+ await new Promise<void>((resolve) => setTimeout(resolve, 0));
+
+ expect(emitted).toBeUndefined();
+ });
+
+ describe('throttleTime', () => {
+ beforeEach(() => {
+ vi.useFakeTimers();
+ });
+
+ afterEach(() => {
+ vi.useRealTimers();
+ });
+
+ async function flushPromises(): Promise<void> {
+ await Promise.resolve();
+ await Promise.resolve();
+ }
+
+ it('should collapse rapid back-to-back reloadConnectorFlow
dispatches into a single loadConnectorFlow within the throttle window', async
() => {
+ const { effects, actions$ } = await setup({
+ connectorId: 'conn-state',
+ processGroupId: 'pg-state'
+ });
+ const actionSubject = new Subject<Action>();
+ actions$(actionSubject.asObservable());
+
+ const emitted: Action[] = [];
+ const subscription =
effects.reloadConnectorFlow$.subscribe((action) => {
+ emitted.push(action);
+ });
+
+ // Three reloadConnectorFlow dispatches inside the 1s throttle
window:
+ // throttleTime defaults to leading-only, so only the first
should produce
+ // a loadConnectorFlow. The remaining two are dropped.
+ actionSubject.next(reloadConnectorFlow());
+ actionSubject.next(reloadConnectorFlow());
+ actionSubject.next(reloadConnectorFlow());
+
+ await flushPromises();
+
+ expect(emitted).toEqual([
+ loadConnectorFlow({
+ connectorId: 'conn-state',
+ processGroupId: 'pg-state'
+ })
+ ]);
+
+ // After the throttle window expires the next dispatch is
allowed through.
+ await vi.advanceTimersByTimeAsync(1500);
+ actionSubject.next(reloadConnectorFlow());
+ await flushPromises();
+
+ expect(emitted).toEqual([
+ loadConnectorFlow({
+ connectorId: 'conn-state',
+ processGroupId: 'pg-state'
+ }),
+ loadConnectorFlow({
+ connectorId: 'conn-state',
+ processGroupId: 'pg-state'
+ })
+ ]);
+
+ subscription.unsubscribe();
+ });
+ });
+ });
+
+ describe('document visibility wake-up', () => {
+ it('should dispatch reloadConnectorFlow when the document becomes
visible after being hidden longer than the polling interval', async () => {
+ const { store } = await setup();
+ const dispatchSpy = vi.spyOn(store, 'dispatch');
+
+ // Simulate the tab being foregrounded after the polling cadence
elapsed.
+ // The lastReload field starts at 0 in the freshly constructed
effects, so
+ // any changedTimestamp greater than the 30 second threshold
satisfies the
+ // wake-up filter.
+ store.overrideSelector(selectDocumentVisibilityState, {
+ documentVisibility: DocumentVisibility.Visible,
+ changedTimestamp: 31 * 1000
+ });
+ store.refreshState();
+
+ expect(dispatchSpy).toHaveBeenCalledWith(reloadConnectorFlow());
+ });
+
+ it('should not dispatch reloadConnectorFlow when the document becomes
visible within the polling interval', async () => {
+ const { store } = await setup();
+ const dispatchSpy = vi.spyOn(store, 'dispatch');
+
+ store.overrideSelector(selectDocumentVisibilityState, {
+ documentVisibility: DocumentVisibility.Visible,
+ changedTimestamp: 5 * 1000
+ });
+ store.refreshState();
+
+
expect(dispatchSpy).not.toHaveBeenCalledWith(reloadConnectorFlow());
+ });
+
+ it('should not dispatch reloadConnectorFlow when the document
transitions to hidden', async () => {
+ const { store } = await setup();
+ const dispatchSpy = vi.spyOn(store, 'dispatch');
+
+ store.overrideSelector(selectDocumentVisibilityState, {
+ documentVisibility: DocumentVisibility.Hidden,
+ changedTimestamp: 60 * 1000
+ });
+ store.refreshState();
+
+
expect(dispatchSpy).not.toHaveBeenCalledWith(reloadConnectorFlow());
+ });
+ });
+
+ describe('startConnectorCanvasPolling$', () => {
+ beforeEach(() => {
+ vi.useFakeTimers();
+ });
+
+ afterEach(() => {
+ vi.useRealTimers();
+ });
+
+ async function flushPromises(): Promise<void> {
+ await Promise.resolve();
+ await Promise.resolve();
+ }
+
+ it('should dispatch reloadConnectorFlow on each polling tick while the
document is visible', async () => {
+ const { effects, actions$ } = await setup();
+ const actionSubject = new Subject<Action>();
+ actions$(actionSubject.asObservable());
+
+ const emitted: Action[] = [];
+ const subscription =
effects.startConnectorCanvasPolling$.subscribe((action) => {
+ emitted.push(action);
+ });
+
+ actionSubject.next(startConnectorCanvasPolling());
+
+ await vi.advanceTimersByTimeAsync(30000);
+ await flushPromises();
+ await vi.advanceTimersByTimeAsync(30000);
+ await flushPromises();
+
+ subscription.unsubscribe();
+
+ expect(emitted).toEqual([reloadConnectorFlow(),
reloadConnectorFlow()]);
+ });
+
+ it('should suppress reloadConnectorFlow while the document is hidden',
async () => {
+ const { effects, actions$ } = await setup({
+ documentVisibility: DocumentVisibility.Hidden
+ });
+ const actionSubject = new Subject<Action>();
+ actions$(actionSubject.asObservable());
+
+ const emitted: Action[] = [];
+ const subscription =
effects.startConnectorCanvasPolling$.subscribe((action) => {
+ emitted.push(action);
+ });
+
+ actionSubject.next(startConnectorCanvasPolling());
+
+ await vi.advanceTimersByTimeAsync(60000);
+ await flushPromises();
+
+ subscription.unsubscribe();
+
+ expect(emitted).toEqual([]);
+ });
+
+ it('should suppress reloadConnectorFlow while a load is already in
flight', async () => {
+ const { effects, actions$ } = await setup({ loadingStatus:
'loading' });
+ const actionSubject = new Subject<Action>();
+ actions$(actionSubject.asObservable());
+
+ const emitted: Action[] = [];
+ const subscription =
effects.startConnectorCanvasPolling$.subscribe((action) => {
+ emitted.push(action);
+ });
+
+ actionSubject.next(startConnectorCanvasPolling());
+
+ await vi.advanceTimersByTimeAsync(30000);
+ await flushPromises();
+
+ subscription.unsubscribe();
+
+ expect(emitted).toEqual([]);
+ });
+
+ it('should stop polling once stopConnectorCanvasPolling is
dispatched', async () => {
+ const { effects, actions$ } = await setup();
+ const actionSubject = new Subject<Action>();
+ actions$(actionSubject.asObservable());
+
+ const emitted: Action[] = [];
+ const subscription =
effects.startConnectorCanvasPolling$.subscribe((action) => {
+ emitted.push(action);
+ });
+
+ actionSubject.next(startConnectorCanvasPolling());
+
+ await vi.advanceTimersByTimeAsync(30000);
+ await flushPromises();
+ expect(emitted).toEqual([reloadConnectorFlow()]);
+
+ actionSubject.next(stopConnectorCanvasPolling());
+
+ await vi.advanceTimersByTimeAsync(60000);
+ await flushPromises();
+
+ subscription.unsubscribe();
+
+ // No additional emissions after stop.
+ expect(emitted).toEqual([reloadConnectorFlow()]);
+ });
});
});
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
index 718a40d71ad..d636a0190b4 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
@@ -15,14 +15,15 @@
* limitations under the License.
*/
-import { Injectable, inject } from '@angular/core';
+import { DestroyRef, Injectable, inject } from '@angular/core';
+import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { concatLatestFrom } from '@ngrx/operators';
import { Store } from '@ngrx/store';
import { Router } from '@angular/router';
-import { NEVER, Observable, of } from 'rxjs';
-import { catchError, filter, map, switchMap, take, tap } from 'rxjs/operators';
-import { ComponentType, ComponentTypeNamePipe, LARGE_DIALOG, MEDIUM_DIALOG,
XL_DIALOG } from '@nifi/shared';
+import { asyncScheduler, interval, NEVER, Observable, of } from 'rxjs';
+import { catchError, filter, map, switchMap, take, takeUntil, tap,
throttleTime } from 'rxjs/operators';
+import { ComponentType, ComponentTypeNamePipe, LARGE_DIALOG, MEDIUM_DIALOG,
NiFiCommon, XL_DIALOG } from '@nifi/shared';
import { selectCurrentUser } from
'../../../../state/current-user/current-user.selectors';
import { selectPrioritizerTypes } from
'../../../../state/extension-types/extension-types.selectors';
import {
@@ -46,8 +47,10 @@ import { SelectedComponent } from
'./connector-canvas.actions';
import * as EmptyQueueActions from
'../../../../state/empty-queue/empty-queue.actions';
import {
selectBreadcrumbs,
+ selectConnectorId,
selectConnectorIdFromRoute,
selectInputPort,
+ selectLoadingStatus,
selectOutputPort,
selectParentProcessGroupId,
selectProcessGroup,
@@ -56,6 +59,8 @@ import {
selectProcessor,
selectRemoteProcessGroup
} from './connector-canvas.selectors';
+import { selectDocumentVisibilityState } from
'../../../../state/document-visibility/document-visibility.selectors';
+import { DocumentVisibility } from '../../../../state/document-visibility';
@Injectable()
export class ConnectorCanvasEffects {
@@ -66,6 +71,35 @@ export class ConnectorCanvasEffects {
private errorHelper = inject(ErrorHelper);
private componentTypeNamePipe = inject(ComponentTypeNamePipe);
private dialog = inject(MatDialog);
+ private destroyRef = inject(DestroyRef);
+
+ /**
+ * Timestamp of the most recent reload (epoch millis). Used by the document
+ * visibility subscription to debounce "tab refocus" reloads so that
+ * switching tabs in quick succession does not produce a thrash of loads.
+ */
+ private lastReload = 0;
+
+ constructor() {
+ // When the document becomes visible after having been hidden long
enough
+ // for the polling interval to be skipped, trigger a reload so the
canvas
+ // does not display stale bulletins / status while the next interval
tick
+ // is pending. The 30 second floor matches the polling cadence and
keeps
+ // rapid tab switches from forcing a fresh load every time.
+ this.store
+ .select(selectDocumentVisibilityState)
+ .pipe(
+ takeUntilDestroyed(this.destroyRef),
+ filter((documentVisibility) =>
documentVisibility.documentVisibility === DocumentVisibility.Visible),
+ filter(
+ (documentVisibility) =>
+ documentVisibility.changedTimestamp - this.lastReload
> 30 * NiFiCommon.MILLIS_PER_SECOND
+ )
+ )
+ .subscribe(() => {
+
this.store.dispatch(ConnectorCanvasActions.reloadConnectorFlow());
+ });
+ }
loadConnectorFlow$ = createEffect(() =>
this.actions$.pipe(
@@ -128,6 +162,69 @@ export class ConnectorCanvasEffects {
)
);
+ /**
+ * Translate a reload request into a fresh {@link loadConnectorFlow} for
the
+ * connector and process group currently mounted on the canvas.
+ *
+ * The connector and process group ids come from the canvas state (set by
the
+ * most recent successful load) rather than the route so that a reload
fired
+ * mid-navigation does not race the URL change. A 1 second throttle
protects
+ * against rapid back-to-back reload requests, and the effect is a no-op
when
+ * the canvas has not yet completed an initial load.
+ */
+ reloadConnectorFlow$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.reloadConnectorFlow),
+ throttleTime(1000),
+ concatLatestFrom(() => [this.store.select(selectConnectorId),
this.store.select(selectProcessGroupId)]),
+ filter(([, connectorId, processGroupId]) => !!connectorId &&
processGroupId != null),
+ switchMap(([, connectorId, processGroupId]) => {
+ this.lastReload = Date.now();
+
+ return of(
+ ConnectorCanvasActions.loadConnectorFlow({
+ connectorId,
+ processGroupId: processGroupId!
+ })
+ );
+ })
+ )
+ );
+
+ /**
+ * Drive periodic refresh of the connector flow.
+ *
+ * The interval ticks every 30 seconds and is torn down when
+ * {@link stopConnectorCanvasPolling} is dispatched (typically from the
+ * canvas component's ngOnDestroy). Polling is skipped when the document is
+ * hidden so background tabs do not generate load, and it is also skipped
+ * while a load is already in flight to avoid racing user-initiated process
+ * group navigations.
+ */
+ startConnectorCanvasPolling$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.startConnectorCanvasPolling),
+ switchMap(() =>
+ interval(30000, asyncScheduler).pipe(
+
takeUntil(this.actions$.pipe(ofType(ConnectorCanvasActions.stopConnectorCanvasPolling)))
+ )
+ ),
+ concatLatestFrom(() => [
+ this.store.select(selectDocumentVisibilityState),
+ this.store.select(selectLoadingStatus)
+ ]),
+ // Skip polling when the document is hidden or while a flow load is
+ // already in flight. Without the loading guard the polling-driven
+ // reload would race a user-initiated process group navigation: the
+ // load effect uses switchMap, so a polling reload would cancel the
+ // pending navigation load and replace it with a load of the
+ // previously-rendered group, leaving the URL and canvas out of
sync.
+ filter(([, documentVisibility]) =>
documentVisibility.documentVisibility === DocumentVisibility.Visible),
+ filter(([, , loadingStatus]) => loadingStatus !== 'loading'),
+ switchMap(() => of(ConnectorCanvasActions.reloadConnectorFlow()))
+ )
+ );
+
/**
* Select components - updates route with selection
* Routes to /connectors/:id/canvas/:processGroupId/:type/:componentId
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
index d28f81a04e0..d1abe53782b 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
@@ -59,6 +59,8 @@ import {
resetConnectorCanvasState,
selectComponents,
setSkipTransform,
+ startConnectorCanvasPolling,
+ stopConnectorCanvasPolling,
viewComponentConfiguration
} from '../../state/connector-canvas/connector-canvas.actions';
import {
@@ -393,6 +395,14 @@ describe('ConnectorCanvasComponent', () => {
);
expect(loadFlowDispatches).toHaveLength(0);
}));
+
+ it('should dispatch startConnectorCanvasPolling on init', fakeAsync(()
=> {
+ const { fixture, dispatchSpy } = setup();
+ fixture.detectChanges();
+ tick();
+
+
expect(dispatchSpy).toHaveBeenCalledWith(startConnectorCanvasPolling());
+ }));
});
describe('Component destruction', () => {
@@ -415,6 +425,16 @@ describe('ConnectorCanvasComponent', () => {
expect(dispatchSpy).toHaveBeenCalledWith(resetConnectorCanvasEntityState());
});
+
+ it('should dispatch stopConnectorCanvasPolling on destroy', () => {
+ const { fixture, dispatchSpy } = setup();
+ fixture.detectChanges();
+ dispatchSpy.mockClear();
+
+ fixture.destroy();
+
+
expect(dispatchSpy).toHaveBeenCalledWith(stopConnectorCanvasPolling());
+ });
});
describe('Data ready state', () => {
@@ -1793,6 +1813,88 @@ describe('ConnectorCanvasComponent', () => {
})
);
}));
+
+ it('should set canClear to true when status data is missing',
fakeAsync(() => {
+ const { fixture, component, dispatchSpy } = setup();
+ fixture.detectChanges();
+ tick();
+ dispatchSpy.mockClear();
+
+ // No status property: runStatus is undefined (not 'Running')
and activeThreadCount
+ // defaults to 0 via the || 0 fallback, so canClear resolves
to true.
+ const processorEntity = {
+ id: 'proc-1',
+ component: { name: 'Stateless Processor', persistsState:
true }
+ };
+
+ component.viewProcessorStateAction(processorEntity);
+
+ expect(dispatchSpy).toHaveBeenCalledWith(
+ getComponentStateAndOpenDialog({
+ request: {
+ componentName: 'Stateless Processor',
+ componentId: 'proc-1',
+ componentType: ComponentType.Processor,
+ canClear: true,
+ connectorId: DEFAULT_CONNECTOR_ID
+ }
+ })
+ );
+ }));
+
+ it('should set canClear to true when processor is disabled',
fakeAsync(() => {
+ const { fixture, component, dispatchSpy } = setup();
+ fixture.detectChanges();
+ tick();
+ dispatchSpy.mockClear();
+
+ const processorEntity = {
+ id: 'proc-1',
+ component: { name: 'Disabled Processor', persistsState:
true },
+ status: { aggregateSnapshot: { runStatus: 'Disabled',
activeThreadCount: 0 } }
+ };
+
+ component.viewProcessorStateAction(processorEntity);
+
+ expect(dispatchSpy).toHaveBeenCalledWith(
+ getComponentStateAndOpenDialog({
+ request: {
+ componentName: 'Disabled Processor',
+ componentId: 'proc-1',
+ componentType: ComponentType.Processor,
+ canClear: true,
+ connectorId: DEFAULT_CONNECTOR_ID
+ }
+ })
+ );
+ }));
+
+ it('should set canClear to true when processor is invalid',
fakeAsync(() => {
+ const { fixture, component, dispatchSpy } = setup();
+ fixture.detectChanges();
+ tick();
+ dispatchSpy.mockClear();
+
+ const processorEntity = {
+ id: 'proc-1',
+ component: { name: 'Invalid Processor', persistsState:
true },
+ status: { aggregateSnapshot: { runStatus: 'Invalid',
activeThreadCount: 0 } }
+ };
+
+ component.viewProcessorStateAction(processorEntity);
+
+ expect(dispatchSpy).toHaveBeenCalledWith(
+ getComponentStateAndOpenDialog({
+ request: {
+ componentName: 'Invalid Processor',
+ componentId: 'proc-1',
+ componentType: ComponentType.Processor,
+ canClear: true,
+ connectorId: DEFAULT_CONNECTOR_ID
+ }
+ })
+ );
+ }));
});
describe('queue and controller-services actions', () => {
@@ -1910,6 +2012,91 @@ describe('ConnectorCanvasComponent', () => {
cancelConnectorDrain({ connector: operableConnectorEntity
as any })
);
});
+
+ it('canDrain should be true when the entity is operable and
DRAIN_FLOWFILES is allowed', () => {
+ const { component, fixture } = setup();
+ const store = TestBed.inject(MockStore);
+ store.overrideSelector(selectConnectorCanvasEntity,
operableConnectorEntity);
+ store.refreshState();
+ fixture.detectChanges();
+
+ expect(component.canDrain()).toBe(true);
+ });
+
+ it('canDrain should be false when the user cannot operate the
connector', () => {
+ const { component, fixture } = setup();
+ const entityWithoutOperatePermissions = {
+ ...operableConnectorEntity,
+ permissions: { canRead: true, canWrite: false },
+ operatePermissions: { canRead: true, canWrite: false }
+ };
+ const store = TestBed.inject(MockStore);
+ store.overrideSelector(selectConnectorCanvasEntity,
entityWithoutOperatePermissions);
+ store.refreshState();
+ fixture.detectChanges();
+
+ expect(component.canDrain()).toBe(false);
+ });
+
+ it('canDrain should be false when the DRAIN_FLOWFILES action is
not allowed', () => {
+ const { component, fixture } = setup();
+ const entityWithDrainDisallowed = {
+ ...operableConnectorEntity,
+ component: {
+ availableActions: [
+ { name: 'DRAIN_FLOWFILES', allowed: false },
+ { name: 'CANCEL_DRAIN_FLOWFILES', allowed: true }
+ ]
+ }
+ };
+ const store = TestBed.inject(MockStore);
+ store.overrideSelector(selectConnectorCanvasEntity,
entityWithDrainDisallowed);
+ store.refreshState();
+ fixture.detectChanges();
+
+ expect(component.canDrain()).toBe(false);
+ });
+
+ it('canDrain should be true when only operatePermissions.canWrite
is true (permissions.canWrite is false)', () => {
+ const { component, fixture } = setup();
+ const operateOnlyEntity = {
+ ...operableConnectorEntity,
+ permissions: { canRead: true, canWrite: false },
+ operatePermissions: { canRead: true, canWrite: true }
+ };
+ const store = TestBed.inject(MockStore);
+ store.overrideSelector(selectConnectorCanvasEntity,
operateOnlyEntity);
+ store.refreshState();
+ fixture.detectChanges();
+
+ expect(component.canDrain()).toBe(true);
+ });
+
+ it('canCancelDrain should be false when no entity is loaded', ()
=> {
+ const { component } = setup();
+ expect(component.canCancelDrain()).toBe(false);
+ });
+
+ it('canCancelDrain should be true when the entity is operable and
CANCEL_DRAIN_FLOWFILES is allowed', () => {
+ const { component, fixture } = setup();
+ const store = TestBed.inject(MockStore);
+ store.overrideSelector(selectConnectorCanvasEntity,
operableConnectorEntity);
+ store.refreshState();
+ fixture.detectChanges();
+
+ expect(component.canCancelDrain()).toBe(true);
+ });
+
+ it('canCancelDrain should be false while the entity is saving', ()
=> {
+ const { component, fixture } = setup();
+ const store = TestBed.inject(MockStore);
+ store.overrideSelector(selectConnectorCanvasEntity,
operableConnectorEntity);
+ store.overrideSelector(selectConnectorCanvasEntitySaving,
true);
+ store.refreshState();
+ fixture.detectChanges();
+
+ expect(component.canCancelDrain()).toBe(false);
+ });
});
});
@@ -2124,4 +2311,16 @@ describe('ConnectorCanvasComponent', () => {
expect(graphControls.canNavigateToParent()).toBe(true);
}));
});
+
+ describe('returnToConnectorListing', () => {
+ it('should navigate to the connectors listing route', () => {
+ const { component } = setup();
+ const router = TestBed.inject(Router);
+ const navigateSpy = vi.spyOn(router, 'navigate');
+
+ component.returnToConnectorListing();
+
+ expect(navigateSpy).toHaveBeenCalledWith(['/connectors']);
+ });
+ });
});
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.ts
index 84e762ea1ee..1f901190aa6 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.ts
@@ -329,6 +329,10 @@ export class ConnectorCanvasComponent implements OnInit,
OnDestroy {
};
this.store.dispatch(setConfiguration({ configuration: config }));
+ // Begin periodic refresh so bulletins, queue counts, and run status
reflect
+ // server-side changes without requiring the user to manually navigate.
+
this.store.dispatch(ConnectorCanvasActions.startConnectorCanvasPolling());
+
// Subscribe to connector ID and process group ID from route and load
flow data
combineLatest([
this.store.select(ConnectorCanvasSelectors.selectConnectorIdFromRoute),
@@ -394,6 +398,7 @@ export class ConnectorCanvasComponent implements OnInit,
OnDestroy {
}
ngOnDestroy(): void {
+
this.store.dispatch(ConnectorCanvasActions.stopConnectorCanvasPolling());
this.store.dispatch(ConnectorCanvasActions.resetConnectorCanvasState());
this.store.dispatch(ConnectorCanvasEntityActions.resetConnectorCanvasEntityState());
}
