rfellows commented on code in PR #11900:
URL: https://github.com/apache/nifi/pull/11900#discussion_r4517666067
##########
nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts:
##########
@@ -1890,6 +1968,438 @@ export class ConnectorCanvasEffects {
)
);
+ createComponentRequest$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createComponentRequest),
+ map((action) => action.request),
+ concatLatestFrom(() => this.store.select(selectProcessGroupId)),
+ switchMap(([request, processGroupId]) => {
+ // Downstream create effects drop requests without a process
group, so clear dragging here.
+ if (processGroupId == null) {
+ return of(ConnectorCanvasActions.setDragging({ dragging:
false }));
+ }
+ switch (request.type) {
+ case ComponentType.Processor:
+ return
of(ConnectorCanvasActions.openNewProcessorDialog({ request: { request } }));
+ case ComponentType.InputPort:
+ case ComponentType.OutputPort:
+ return of(ConnectorCanvasActions.openNewPortDialog({
request }));
+ case ComponentType.ProcessGroup:
+ return of(
+ ConnectorCanvasActions.openNewProcessGroupDialog({
+ request: { request, parameterContexts: [] }
+ })
+ );
+ case ComponentType.RemoteProcessGroup:
+ return
of(ConnectorCanvasActions.openNewRemoteProcessGroupDialog({ request }));
+ case ComponentType.Funnel:
+ return of(ConnectorCanvasActions.createFunnel({
request }));
+ case ComponentType.Label:
+ return of(ConnectorCanvasActions.createLabel({ request
}));
+ default:
+ // No dialog opens to clear dragging through
afterClosed().
+ return from([
+ ConnectorCanvasActions.setDragging({ dragging:
false }),
+ ErrorActions.addBannerError({
+ errorContext: {
+ errors: [`Unsupported component type:
${request.type}`],
+ context: ErrorContextKey.CONNECTOR_CANVAS
+ }
+ })
+ ]);
+ }
+ })
+ )
+ );
+
+ openNewProcessorDialog$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.openNewProcessorDialog),
+ concatLatestFrom(() =>
this.store.select(selectProcessGroupId)),
+ filter(([, processGroupId]) => processGroupId != null),
+ tap(([{ request }, processGroupId]) => {
+ const dialogRef = this.dialog.open(CreateProcessor, {
+ ...LARGE_DIALOG,
+ data: request
+ });
+ const instance = dialogRef.componentInstance;
+ instance.saving$ = this.store.select(selectSaving);
+ instance.processorTypes$ =
this.store.select(selectProcessorTypes);
+ instance.processorTypesLoadingStatus$ =
this.store.select(selectExtensionTypesLoadingStatus);
+
+ const afterClosed$ = dialogRef.afterClosed().pipe(take(1));
+
instance.createProcessor.pipe(takeUntil(afterClosed$)).subscribe((createRequest)
=> {
+ this.store.dispatch(
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...createRequest, type:
ComponentType.Processor },
+ processGroupId: processGroupId!,
+ // The processor dialog has no error banner of
its own.
+ errorContextKey:
ErrorContextKey.CONNECTOR_CANVAS
+ })
+ );
+ });
+ afterClosed$.subscribe(() =>
+
this.store.dispatch(ConnectorCanvasActions.setDragging({ dragging: false }))
+ );
+ })
+ ),
+ { dispatch: false }
+ );
+
+ openNewPortDialog$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.openNewPortDialog),
+ concatLatestFrom(() => [
+ this.store.select(selectProcessGroupId),
+ this.store.select(selectParentProcessGroupId)
+ ]),
+ filter(([, processGroupId]) => processGroupId != null),
+ tap(([{ request }, processGroupId, parentProcessGroupId]) => {
+ const dialogRef = this.dialog.open(CreatePort, {
+ ...SMALL_DIALOG,
+ data: request
+ });
+ const instance = dialogRef.componentInstance;
+ instance.saving$ = this.store.select(selectSaving);
+ instance.isRootProcessGroup = parentProcessGroupId == null;
+
+ const afterClosed$ = dialogRef.afterClosed().pipe(take(1));
+
instance.createPort.pipe(takeUntil(afterClosed$)).subscribe((createRequest) => {
+ const type =
+ createRequest.type === ComponentType.OutputPort
+ ? ComponentType.OutputPort
+ : ComponentType.InputPort;
+ this.store.dispatch(
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...createRequest, type },
+ processGroupId: processGroupId!,
+ errorContextKey: ErrorContextKey.PORT
+ })
+ );
+ });
+ afterClosed$.subscribe(() =>
+
this.store.dispatch(ConnectorCanvasActions.setDragging({ dragging: false }))
+ );
+ })
+ ),
+ { dispatch: false }
+ );
+
+ openNewProcessGroupDialog$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.openNewProcessGroupDialog),
+ concatLatestFrom(() =>
this.store.select(selectProcessGroupId)),
+ filter(([, processGroupId]) => processGroupId != null),
+ tap(([{ request }, processGroupId]) => {
+ const dialogRef = this.dialog.open(CreateProcessGroup, {
+ ...MEDIUM_DIALOG,
+ data: request
+ });
+ const instance = dialogRef.componentInstance;
+ instance.parameterContexts = request.parameterContexts;
+ instance.saving$ = this.store.select(selectSaving);
+ instance.currentUser$ =
this.store.select(selectCurrentUser);
+ // Connector flows bind the connector's own parameter
context.
+ instance.supportsParameters = false;
+
+ const afterClosed$ = dialogRef.afterClosed().pipe(take(1));
+
instance.createProcessGroup.pipe(takeUntil(afterClosed$)).subscribe((createRequest)
=> {
+ this.store.dispatch(
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...createRequest, type:
ComponentType.ProcessGroup },
+ processGroupId: processGroupId!,
+ errorContextKey: ErrorContextKey.PROCESS_GROUP
+ })
+ );
+ });
+
instance.uploadProcessGroup.pipe(takeUntil(afterClosed$)).subscribe((uploadRequest)
=> {
+ this.store.dispatch(
+ ConnectorCanvasActions.uploadProcessGroup({
+ request: uploadRequest,
+ processGroupId: processGroupId!
+ })
+ );
+ });
+ afterClosed$.subscribe(() =>
+
this.store.dispatch(ConnectorCanvasActions.setDragging({ dragging: false }))
+ );
+ })
+ ),
+ { dispatch: false }
+ );
+
+ openNewRemoteProcessGroupDialog$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.openNewRemoteProcessGroupDialog),
+ concatLatestFrom(() =>
this.store.select(selectProcessGroupId)),
+ filter(([, processGroupId]) => processGroupId != null),
+ tap(([{ request }, processGroupId]) => {
+ const dialogRef =
this.dialog.open(CreateRemoteProcessGroup, {
+ ...LARGE_DIALOG,
+ data: request
+ });
+ const instance = dialogRef.componentInstance;
+ instance.saving$ = this.store.select(selectSaving);
+
+ const afterClosed$ = dialogRef.afterClosed().pipe(take(1));
+
instance.createRemoteProcessGroup.pipe(takeUntil(afterClosed$)).subscribe((createRequest)
=> {
+ this.store.dispatch(
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...createRequest, type:
ComponentType.RemoteProcessGroup },
+ processGroupId: processGroupId!,
+ errorContextKey:
ErrorContextKey.REMOTE_PROCESS_GROUP
+ })
+ );
+ });
+ afterClosed$.subscribe(() =>
+
this.store.dispatch(ConnectorCanvasActions.setDragging({ dragging: false }))
+ );
+ })
+ ),
+ { dispatch: false }
+ );
+
+ createFunnel$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createFunnel),
+ concatLatestFrom(() => this.store.select(selectProcessGroupId)),
+ filter(([, processGroupId]) => processGroupId != null),
+ map(([{ request }, processGroupId]) =>
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...request, type: ComponentType.Funnel },
+ processGroupId: processGroupId!,
+ errorContextKey: ErrorContextKey.CONNECTOR_CANVAS
+ })
+ )
+ )
+ );
+
+ createLabel$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createLabel),
+ concatLatestFrom(() => [this.store.select(selectProcessGroupId),
this.store.select(selectMaxLabelZIndex)]),
+ filter(([, processGroupId]) => processGroupId != null),
+ map(([{ request }, processGroupId, maxZIndex]) =>
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...request, type: ComponentType.Label, zIndex:
maxZIndex + 1 },
+ processGroupId: processGroupId!,
+ errorContextKey: ErrorContextKey.CONNECTOR_CANVAS
+ })
+ )
+ )
+ );
+
+ /**
+ * exhaustMap drops a second submit while the first is in flight so a
double click
+ * cannot create two components.
+ */
+ createCanvasComponent$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createCanvasComponent),
+ exhaustMap(({ request, processGroupId, errorContextKey }) =>
+ this.connectorEditService.createComponent(processGroupId,
request).pipe(
+ map((response) =>
ConnectorCanvasActions.createCanvasComponentSuccess({ response })),
+ catchError((error: HttpErrorResponse) =>
+ from(
+ this.mutationFailure(
+ error,
+ (message) =>
+
ConnectorCanvasActions.createCanvasComponentFailure({
+ error: message,
+ errorContextKey
+ }),
+ errorContextKey
+ )
+ )
+ )
+ )
+ )
+ )
+ );
+
+ createCanvasComponentSuccess$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createCanvasComponentSuccess),
+ tap(() => this.dialog.closeAll())
Review Comment:
The flow designer selects a newly created component
(`createComponentComplete$`), and this PR already selects new connections.
Processors, ports, funnels, labels and process groups created here aren't
selected, though.
Would it make sense to dispatch `selectComponents` for `response.payload.id`
/ `response.type` alongside `closeAll()`, for parity?
##########
nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas-entity/connector-canvas-entity.effects.ts:
##########
@@ -81,6 +84,30 @@ export class ConnectorCanvasEntityEffects {
)
);
+ /**
+ * Keeps the canvas edit feature in line with the connector's
troubleshooting state.
+ *
+ * Every flow load refreshes the connector entity, so the canvas follows
the state
+ * the server reports. The connector canvas initializes the feature to
read-only, so
+ * a configuration is dispatched only when the feature diverges from the
entity.
+ */
+ troubleshootingTransitions$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(loadConnectorEntitySuccess),
+ concatLatestFrom(() => [
+ this.store.select(selectConnectorCanvasEntity),
+ this.store.select(selectCanvasFeatures)
+ ]),
+ filter(([, connectorEntity]) => connectorEntity !== null),
+ map(([, connectorEntity, features]) => ({
+ canEdit: connectorEntity!.component.state ===
'TROUBLESHOOTING',
Review Comment:
`canEdit` here is set from `state === 'TROUBLESHOOTING'` alone. The palette,
the Delete context-menu item and the Delete key all gate on
`ConnectorCanvasContextMenu.canEdit`
(`connector-canvas-context-menu.service.ts:78`), which also requires
`canvasPermissions()?.canWrite`.
This canvas feature flag is what enables component drag and the
connection-drawing handles. So a user who can read but not write the flow can
still start a drag or draw a connection during Troubleshooting. The server
rejects it, but the user sees an error banner and the component snaps back.
Could this use the same condition as the context menu? It would also be
worth adding a spec for "in Troubleshooting but not writable".
##########
nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.reducer.spec.ts:
##########
@@ -959,4 +997,254 @@ describe('connectorCanvasReducer', () => {
expect(result.processors[0].revision).toBe(response.revision);
});
});
+
+ describe('component create and delete', () => {
+ function loadSnapshot(
+ overrides: Partial<Parameters<typeof
ConnectorCanvasActions.loadConnectorFlowSuccess>[0]> = {}
+ ) {
+ return ConnectorCanvasActions.loadConnectorFlowSuccess({
+ connectorId: 'connector-1',
+ processGroupId: 'pg-current',
+ processGroupRevision: null,
+ parentProcessGroupId: 'pg-parent',
+ breadcrumb: null,
+ permissions: null,
+ labels: [],
+ funnels: [],
+ inputPorts: [],
+ outputPorts: [],
+ remoteProcessGroups: [],
+ processGroups: [],
+ processors: [],
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache: [] },
+ connectedStateChanged: false,
+ ...overrides
+ });
+ }
+
+ it('stores the flow-level permissions from the loaded process group',
() => {
+ const permissions = { canRead: true, canWrite: true };
+
+ const result = connectorCanvasReducer(initialConnectorCanvasState,
loadSnapshot({ permissions }));
+
+ expect(result.permissions).toEqual(permissions);
+ });
+
+ it('clears saving when a delete fails', () => {
+ const result = connectorCanvasReducer(
+ createPopulatedState({ saving: true }),
+ ConnectorCanvasActions.deleteSelectedComponentsFailure({
error: 'boom' })
+ );
+
+ expect(result.saving).toBe(false);
+ });
+
+ it('tracks palette dragging', () => {
+ const result = connectorCanvasReducer(
+ initialConnectorCanvasState,
+ ConnectorCanvasActions.setDragging({ dragging: true })
+ );
+
+ expect(result.dragging).toBe(true);
+ });
+
+ it.each([
+ [ComponentType.Processor, 'processors',
makeProcessor('new-component')],
+ [ComponentType.InputPort, 'inputPorts', makePort('new-component')],
+ [ComponentType.OutputPort, 'outputPorts',
makePort('new-component', { portType: 'OUTPUT_PORT' })],
+ [ComponentType.ProcessGroup, 'processGroups',
makeProcessGroup('new-component')],
+ [ComponentType.RemoteProcessGroup, 'remoteProcessGroups',
makeRemoteProcessGroup('new-component')],
+ [ComponentType.Funnel, 'funnels', makeFunnel('new-component')],
+ [ComponentType.Label, 'labels', makeLabel('new-component')]
+ ] as const)('adds a created %s to %s and the added cache', (type,
collection, payload) => {
+ const state = createPopulatedState({ saving: true, dragging: true
});
+
+ const result = connectorCanvasReducer(
+ state,
+ ConnectorCanvasActions.createCanvasComponentSuccess({
+ response: { type, payload } as Parameters<
+ typeof
ConnectorCanvasActions.createCanvasComponentSuccess
+ >[0]['response']
+ })
+ );
+
+ expect(result[collection]).toContainEqual(payload);
+ expect(result[collection]).toHaveLength(state[collection].length +
1);
+ expect(result.addedCache).toEqual(['new-component']);
+ expect(result.saving).toBe(false);
+ expect(result.dragging).toBe(false);
+ });
+
+ it('keeps a created component when an older snapshot omits it', () => {
+ const created = makeFunnel('funnel-new');
+ const afterCreate = connectorCanvasReducer(
+ createPopulatedState(),
+ ConnectorCanvasActions.createCanvasComponentSuccess({
+ response: { type: ComponentType.Funnel, payload: created }
+ })
+ );
+
+ const result = connectorCanvasReducer(afterCreate, loadSnapshot({
funnels: [makeFunnel('funnel-1')] }));
+
+ expect(result.funnels.map((funnel) =>
funnel.id)).toEqual(['funnel-1', 'funnel-new']);
+ expect(result.addedCache).toEqual(['funnel-new']);
+ });
+
+ it('clears saving and dragging when a create fails', () => {
+ const result = connectorCanvasReducer(
+ createPopulatedState({ saving: true, dragging: true }),
+ ConnectorCanvasActions.createCanvasComponentFailure({
+ error: 'boom',
+ errorContextKey: ErrorContextKey.PROCESSOR
+ })
+ );
+
+ expect(result.saving).toBe(false);
+ expect(result.dragging).toBe(false);
+ });
+
+ it('sets saving while a delete is in flight', () => {
+ const result = connectorCanvasReducer(
+ createPopulatedState(),
+ ConnectorCanvasActions.deleteSelectedComponents({
+ request: [{ id: 'proc-1', type: ComponentType.Processor,
revision: { version: 0 } }]
+ })
+ );
+
+ expect(result.saving).toBe(true);
+ });
+
+ it('removes deleted components and their outgoing connections', () => {
+ const state = createPopulatedState({
+ saving: true,
+ connections: [
+ makeConnection('outgoing', { sourceId: 'proc-1',
destinationId: 'funnel-1' }),
+ makeConnection('self-loop', { sourceId: 'proc-1',
destinationId: 'proc-1' }),
+ makeConnection('unrelated', { sourceId: 'proc-2',
destinationId: 'funnel-1' })
+ ]
+ });
+
+ const result = connectorCanvasReducer(
+ state,
+ ConnectorCanvasActions.deleteSelectedComponentsSuccess({
+ ids: [{ id: 'proc-1', type: ComponentType.Processor }]
+ })
+ );
+
+ expect(result.processors.map((processor) =>
processor.id)).toEqual(['proc-2']);
+ expect(result.connections.map((connection) =>
connection.id)).toEqual(['unrelated']);
+ expect(result.removedCache).toEqual(['proc-1', 'outgoing',
'self-loop']);
+ expect(result.saving).toBe(false);
+ });
+
+ it('removes connections leaving a deleted remote process group', () =>
{
+ const state = createPopulatedState({
+ remoteProcessGroups: [makeRemoteProcessGroup('rpg-1')],
+ connections: [
+ makeConnection('from-rpg', {
+ sourceId: 'remote-output-port-1',
+ sourceGroupId: 'rpg-1',
+ sourceType: 'REMOTE_OUTPUT_PORT',
+ destinationId: 'funnel-1'
+ }),
+ makeConnection('unrelated', { sourceId: 'proc-2',
destinationId: 'funnel-1' })
+ ]
+ });
+
+ const result = connectorCanvasReducer(
+ state,
+ ConnectorCanvasActions.deleteSelectedComponentsSuccess({
+ ids: [{ id: 'rpg-1', type:
ComponentType.RemoteProcessGroup }]
+ })
+ );
+
+ expect(result.remoteProcessGroups).toEqual([]);
+ expect(result.connections.map((connection) =>
connection.id)).toEqual(['unrelated']);
+ expect(result.removedCache).toEqual(['rpg-1', 'from-rpg']);
+ });
+
+ it('does not restore a deleted component or its connections from an
older snapshot', () => {
+ const outgoing = makeConnection('outgoing', { sourceId: 'proc-1',
destinationId: 'funnel-1' });
+ const afterDelete = connectorCanvasReducer(
+ createPopulatedState({ connections: [outgoing] }),
+ ConnectorCanvasActions.deleteSelectedComponentsSuccess({
+ ids: [{ id: 'proc-1', type: ComponentType.Processor }]
+ })
+ );
+
+ const result = connectorCanvasReducer(
+ afterDelete,
+ loadSnapshot({
+ processors: [makeProcessor('proc-1'),
makeProcessor('proc-2')],
+ connections: [outgoing]
+ })
+ );
+
+ expect(result.processors.map((processor) =>
processor.id)).toEqual(['proc-2']);
+ expect(result.connections).toEqual([]);
+ });
+
+ it('does not restore a connection leaving a deleted remote process
group from an older snapshot', () => {
Review Comment:
The reducer correctly maps a connection's source to the child process group
when `sourceGroupId !== processGroupId`. These specs cover the remote process
group case, but not deleting a child process group whose output ports have
outgoing connections. Could you add a case for that?
##########
nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts:
##########
@@ -1890,6 +1968,438 @@ export class ConnectorCanvasEffects {
)
);
+ createComponentRequest$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createComponentRequest),
+ map((action) => action.request),
+ concatLatestFrom(() => this.store.select(selectProcessGroupId)),
+ switchMap(([request, processGroupId]) => {
+ // Downstream create effects drop requests without a process
group, so clear dragging here.
+ if (processGroupId == null) {
+ return of(ConnectorCanvasActions.setDragging({ dragging:
false }));
+ }
+ switch (request.type) {
+ case ComponentType.Processor:
+ return
of(ConnectorCanvasActions.openNewProcessorDialog({ request: { request } }));
+ case ComponentType.InputPort:
+ case ComponentType.OutputPort:
+ return of(ConnectorCanvasActions.openNewPortDialog({
request }));
+ case ComponentType.ProcessGroup:
+ return of(
+ ConnectorCanvasActions.openNewProcessGroupDialog({
+ request: { request, parameterContexts: [] }
+ })
+ );
+ case ComponentType.RemoteProcessGroup:
+ return
of(ConnectorCanvasActions.openNewRemoteProcessGroupDialog({ request }));
+ case ComponentType.Funnel:
+ return of(ConnectorCanvasActions.createFunnel({
request }));
+ case ComponentType.Label:
+ return of(ConnectorCanvasActions.createLabel({ request
}));
+ default:
+ // No dialog opens to clear dragging through
afterClosed().
+ return from([
+ ConnectorCanvasActions.setDragging({ dragging:
false }),
+ ErrorActions.addBannerError({
+ errorContext: {
+ errors: [`Unsupported component type:
${request.type}`],
+ context: ErrorContextKey.CONNECTOR_CANVAS
+ }
+ })
+ ]);
+ }
+ })
+ )
+ );
+
+ openNewProcessorDialog$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.openNewProcessorDialog),
+ concatLatestFrom(() =>
this.store.select(selectProcessGroupId)),
+ filter(([, processGroupId]) => processGroupId != null),
+ tap(([{ request }, processGroupId]) => {
+ const dialogRef = this.dialog.open(CreateProcessor, {
+ ...LARGE_DIALOG,
+ data: request
+ });
+ const instance = dialogRef.componentInstance;
+ instance.saving$ = this.store.select(selectSaving);
+ instance.processorTypes$ =
this.store.select(selectProcessorTypes);
+ instance.processorTypesLoadingStatus$ =
this.store.select(selectExtensionTypesLoadingStatus);
+
+ const afterClosed$ = dialogRef.afterClosed().pipe(take(1));
+
instance.createProcessor.pipe(takeUntil(afterClosed$)).subscribe((createRequest)
=> {
+ this.store.dispatch(
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...createRequest, type:
ComponentType.Processor },
+ processGroupId: processGroupId!,
+ // The processor dialog has no error banner of
its own.
+ errorContextKey:
ErrorContextKey.CONNECTOR_CANVAS
+ })
+ );
+ });
+ afterClosed$.subscribe(() =>
+
this.store.dispatch(ConnectorCanvasActions.setDragging({ dragging: false }))
+ );
+ })
+ ),
+ { dispatch: false }
+ );
+
+ openNewPortDialog$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.openNewPortDialog),
+ concatLatestFrom(() => [
+ this.store.select(selectProcessGroupId),
+ this.store.select(selectParentProcessGroupId)
+ ]),
+ filter(([, processGroupId]) => processGroupId != null),
+ tap(([{ request }, processGroupId, parentProcessGroupId]) => {
+ const dialogRef = this.dialog.open(CreatePort, {
+ ...SMALL_DIALOG,
+ data: request
+ });
+ const instance = dialogRef.componentInstance;
+ instance.saving$ = this.store.select(selectSaving);
+ instance.isRootProcessGroup = parentProcessGroupId == null;
+
+ const afterClosed$ = dialogRef.afterClosed().pipe(take(1));
+
instance.createPort.pipe(takeUntil(afterClosed$)).subscribe((createRequest) => {
+ const type =
+ createRequest.type === ComponentType.OutputPort
+ ? ComponentType.OutputPort
+ : ComponentType.InputPort;
+ this.store.dispatch(
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...createRequest, type },
+ processGroupId: processGroupId!,
+ errorContextKey: ErrorContextKey.PORT
+ })
+ );
+ });
+ afterClosed$.subscribe(() =>
+
this.store.dispatch(ConnectorCanvasActions.setDragging({ dragging: false }))
+ );
+ })
+ ),
+ { dispatch: false }
+ );
+
+ openNewProcessGroupDialog$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.openNewProcessGroupDialog),
+ concatLatestFrom(() =>
this.store.select(selectProcessGroupId)),
+ filter(([, processGroupId]) => processGroupId != null),
+ tap(([{ request }, processGroupId]) => {
+ const dialogRef = this.dialog.open(CreateProcessGroup, {
+ ...MEDIUM_DIALOG,
+ data: request
+ });
+ const instance = dialogRef.componentInstance;
+ instance.parameterContexts = request.parameterContexts;
+ instance.saving$ = this.store.select(selectSaving);
+ instance.currentUser$ =
this.store.select(selectCurrentUser);
+ // Connector flows bind the connector's own parameter
context.
+ instance.supportsParameters = false;
+
+ const afterClosed$ = dialogRef.afterClosed().pipe(take(1));
+
instance.createProcessGroup.pipe(takeUntil(afterClosed$)).subscribe((createRequest)
=> {
+ this.store.dispatch(
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...createRequest, type:
ComponentType.ProcessGroup },
+ processGroupId: processGroupId!,
+ errorContextKey: ErrorContextKey.PROCESS_GROUP
+ })
+ );
+ });
+
instance.uploadProcessGroup.pipe(takeUntil(afterClosed$)).subscribe((uploadRequest)
=> {
+ this.store.dispatch(
+ ConnectorCanvasActions.uploadProcessGroup({
+ request: uploadRequest,
+ processGroupId: processGroupId!
+ })
+ );
+ });
+ afterClosed$.subscribe(() =>
+
this.store.dispatch(ConnectorCanvasActions.setDragging({ dragging: false }))
+ );
+ })
+ ),
+ { dispatch: false }
+ );
+
+ openNewRemoteProcessGroupDialog$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.openNewRemoteProcessGroupDialog),
+ concatLatestFrom(() =>
this.store.select(selectProcessGroupId)),
+ filter(([, processGroupId]) => processGroupId != null),
+ tap(([{ request }, processGroupId]) => {
+ const dialogRef =
this.dialog.open(CreateRemoteProcessGroup, {
+ ...LARGE_DIALOG,
+ data: request
+ });
+ const instance = dialogRef.componentInstance;
+ instance.saving$ = this.store.select(selectSaving);
+
+ const afterClosed$ = dialogRef.afterClosed().pipe(take(1));
+
instance.createRemoteProcessGroup.pipe(takeUntil(afterClosed$)).subscribe((createRequest)
=> {
+ this.store.dispatch(
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...createRequest, type:
ComponentType.RemoteProcessGroup },
+ processGroupId: processGroupId!,
+ errorContextKey:
ErrorContextKey.REMOTE_PROCESS_GROUP
+ })
+ );
+ });
+ afterClosed$.subscribe(() =>
+
this.store.dispatch(ConnectorCanvasActions.setDragging({ dragging: false }))
+ );
+ })
+ ),
+ { dispatch: false }
+ );
+
+ createFunnel$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createFunnel),
+ concatLatestFrom(() => this.store.select(selectProcessGroupId)),
+ filter(([, processGroupId]) => processGroupId != null),
+ map(([{ request }, processGroupId]) =>
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...request, type: ComponentType.Funnel },
+ processGroupId: processGroupId!,
+ errorContextKey: ErrorContextKey.CONNECTOR_CANVAS
+ })
+ )
+ )
+ );
+
+ createLabel$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createLabel),
+ concatLatestFrom(() => [this.store.select(selectProcessGroupId),
this.store.select(selectMaxLabelZIndex)]),
+ filter(([, processGroupId]) => processGroupId != null),
+ map(([{ request }, processGroupId, maxZIndex]) =>
+ ConnectorCanvasActions.createCanvasComponent({
+ request: { ...request, type: ComponentType.Label, zIndex:
maxZIndex + 1 },
+ processGroupId: processGroupId!,
+ errorContextKey: ErrorContextKey.CONNECTOR_CANVAS
+ })
+ )
+ )
+ );
+
+ /**
+ * exhaustMap drops a second submit while the first is in flight so a
double click
+ * cannot create two components.
+ */
+ createCanvasComponent$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createCanvasComponent),
+ exhaustMap(({ request, processGroupId, errorContextKey }) =>
+ this.connectorEditService.createComponent(processGroupId,
request).pipe(
+ map((response) =>
ConnectorCanvasActions.createCanvasComponentSuccess({ response })),
+ catchError((error: HttpErrorResponse) =>
+ from(
+ this.mutationFailure(
+ error,
+ (message) =>
+
ConnectorCanvasActions.createCanvasComponentFailure({
+ error: message,
+ errorContextKey
+ }),
+ errorContextKey
+ )
+ )
+ )
+ )
+ )
+ )
+ );
+
+ createCanvasComponentSuccess$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createCanvasComponentSuccess),
+ tap(() => this.dialog.closeAll())
+ ),
+ { dispatch: false }
+ );
+
+ /**
+ * Dialogs with their own error banner stay open so the user can retry.
Any other
+ * failure is shown on the canvas banner, which an open dialog would cover.
+ */
+ createCanvasComponentFailure$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.createCanvasComponentFailure),
+ filter(({ errorContextKey }) => errorContextKey ===
ErrorContextKey.CONNECTOR_CANVAS),
+ tap(() => this.dialog.closeAll())
+ ),
+ { dispatch: false }
+ );
+
+ uploadProcessGroup$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.uploadProcessGroup),
+ exhaustMap(({ request, processGroupId }) =>
+ this.connectorEditService.uploadProcessGroup(processGroupId,
request).pipe(
+ switchMap((payload) => {
+ const response: ConnectorCanvasCreatedComponent = {
+ type: ComponentType.ProcessGroup,
+ payload
+ };
+ return from([
+
ConnectorCanvasActions.createCanvasComponentSuccess({ response }),
+ ConnectorCanvasActions.uploadProcessGroupSuccess({
response })
+ ]);
+ }),
+ catchError((error: HttpErrorResponse) =>
+ from(
+ this.mutationFailure(
+ error,
+ (message) =>
ConnectorCanvasActions.uploadProcessGroupFailure({ error: message }),
+ ErrorContextKey.PROCESS_GROUP
+ )
+ )
+ )
+ )
+ )
+ )
+ );
+
+ /**
+ * A single component is deleted directly; a multi-selection goes through
a snippet.
+ * Deleting a single connection reloads its endpoints so their validity
updates.
+ * A failure reloads the flow so the canvas reflects what the server kept.
+ */
+ deleteSelectedComponents$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.deleteSelectedComponents),
+ concatLatestFrom(() => [
+ this.store.select(selectConnectorId),
+ this.store.select(selectProcessGroupId),
+ this.store.select(selectConnections)
+ ]),
+ filter(
+ ([{ request }, connectorId, processGroupId]) =>
+ request.length > 0 && connectorId !== '' && processGroupId
!== null
+ ),
+ mergeMap(([{ request }, connectorId, processGroupId, connections])
=> {
Review Comment:
Two small things on this effect:
- With `mergeMap`, pressing Delete twice quickly sends the same revisions
twice. The second request gets a 409, which shows a banner and reloads the
flow. `exhaustMap` would make this consistent with `createCanvasComponent$`.
- If the `filter` above drops a request, the reducer has already set
`saving: true` and nothing clears it. In practice `canEdit` means the flow is
always loaded by then, so this is defensive only.
##########
nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/service/connector-edit.service.ts:
##########
@@ -51,11 +78,238 @@ export class ConnectorEditService {
);
}
+ /**
+ * Create a component of the request's type and return it tagged with that
type.
+ */
+ createComponent(
+ processGroupId: string,
+ request: ConnectorCanvasCreateRequest
+ ): Observable<ConnectorCanvasCreatedComponent> {
+ switch (request.type) {
+ case ComponentType.Processor:
+ return this.createProcessor(processGroupId, request).pipe(
+ map((payload) => ({ type: request.type, payload }))
+ );
+ case ComponentType.InputPort:
+ case ComponentType.OutputPort:
+ return this.createPort(processGroupId, request).pipe(
+ map((payload) => ({ type: request.type, payload }))
+ );
+ case ComponentType.ProcessGroup:
+ return this.createProcessGroup(processGroupId, request).pipe(
+ map((payload) => ({ type: request.type, payload }))
+ );
+ case ComponentType.RemoteProcessGroup:
+ return this.createRemoteProcessGroup(processGroupId,
request).pipe(
+ map((payload) => ({ type: request.type, payload }))
+ );
+ case ComponentType.Funnel:
+ return this.createFunnel(processGroupId, request).pipe(
+ map((payload) => ({ type: request.type, payload }))
+ );
+ case ComponentType.Label:
+ return this.createLabel(processGroupId, request).pipe(
+ map((payload) => ({ type: request.type, payload }))
+ );
+ }
+ }
+
+ createProcessor(processGroupId: string, request: CreateProcessorRequest):
Observable<ProcessorEntity> {
+ return defer(() =>
+ this.httpClient.post<ProcessorEntity>(
+
`${ConnectorEditService.API}/process-groups/${safeApiPath(processGroupId)}/processors`,
Review Comment:
The new create, get and delete calls in this service go through
`safeApiPath`, but two raw segments remain earlier in the file (outside this
diff):
- `updateComponent` (line 71): `${path}/${request.id}`
- `createConnection` (line 75):
`process-groups/${processGroupId}/connections`
The PR description says every dynamic path segment in create and delete
requests is validated, and `createConnection` is a create. Could both be
wrapped the same way as these new methods (`safeApiPath(...)` inside `defer`)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
