NIFI-14777 Restore capability from NiFi 1.x to view upstream/downstre…

nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-context-menu.service.ts

@@ -49,11 +49,13 @@ import {
     runOnce,
     stopVersionControlRequest,
     terminateThreads,
-    updatePositions
+    updatePositions,
+    viewComponentConnections
 } from '../state/flow/flow.actions';
 import { ComponentType } from '@nifi/shared';
 import {
     ConfirmStopVersionControlRequest,
+    ConnectionDirection,
     MoveComponentRequest,
     OpenChangeVersionDialogRequest,
     OpenLocalChangesDialogRequest
@@ -286,25 +288,23 @@ export class CanvasContextMenu implements ContextMenuDefinitionProvider {
         id: 'upstream-downstream',
         menuItems: [
             {
-                condition: () => {
-                    // TODO - hasUpstream
-                    return false;
+                condition: (selection: d3.Selection<any, any, any, any>) => {
+                    return this.canvasUtils.hasUpstream(selection);
                 },
-                clazz: 'icon',
+                clazz: 'fa fa-long-arrow-up fa-rotate-45',
                 text: 'Upstream',
-                action: () => {
-                    // TODO - showUpstream
+                action: (selection: d3.Selection<any, any, any, any>) => {
+                    this.requestComponentConnections(selection, 'upstream');
                 }
             },
             {
-                condition: () => {
-                    // TODO - hasDownstream
-                    return false;
+                condition: (selection: d3.Selection<any, any, any, any>) => {
+                    return this.canvasUtils.hasDownstream(selection);
                 },
-                clazz: 'icon',
+                clazz: 'fa fa-long-arrow-down fa-rotate-45',
                 text: 'Downstream',
-                action: () => {
-                    // TODO - showDownstream
+                action: (selection: d3.Selection<any, any, any, any>) => {
+                    this.requestComponentConnections(selection, 'downstream');
                 }
             }
         ]
@@ -1434,4 +1434,48 @@ export class CanvasContextMenu implements ContextMenuDefinitionProvider {
             menuItem.action(selection, event);
         }
     }
+
+    /**
+     * Requests the connections attached to the specified component in the specified direction.
+     *
+     * A component's connections are defined in the group that encloses it, which is the group currently
+     * on the canvas. The exception is a port whose connections cross that group's own boundary — the
+     * upstream side of an Input Port and the downstream side of an Output Port — which are defined one
+     * level up and are not rendered alongside the port at all. Those are the two cases that
+     * hasUpstream/hasDownstream gate on the presence of a parent group.
+     */
+    private requestComponentConnections(
+        selection: d3.Selection<any, any, any, any>,
+        direction: ConnectionDirection
+    ): void {
+        const crossesParentBoundary: boolean =
+            (direction === 'upstream' && this.canvasUtils.isInputPort(selection)) ||
+            (direction === 'downstream' && this.canvasUtils.isOutputPort(selection));
+
+        let groupId: string | null = this.canvasUtils.getProcessGroupId();
+        if (crossesParentBoundary) {
+            groupId = this.canvasUtils.getParentProcessGroupId();
+
+            // hasUpstream/hasDownstream do not offer these directions without a parent group
+            if (groupId === null) {
+                return;
+            }
+        }
+
+        const selectionData = selection.datum();
+        this.store.dispatch(
+            viewComponentConnections({
+                request: {
+                    id: selectionData.id,
+                    // funnels have no name, and an unreadable component has no name to read
+                    name: selectionData.permissions.canRead
+                        ? (selectionData.component.name ?? selectionData.id)
+                        : selectionData.id,
+                    type: selectionData.type,
+                    groupId,
+                    direction
+                }
+            })
+        );
+    }
 }

nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-utils.service.spec.ts

@@ -1010,4 +1010,73 @@ describe('CanvasUtils', () => {
             );
         });
     });
+
+    // These resolve a connection endpoint to the component as it is rendered in the group currently
+    // being viewed: a port inside a child group resolves to that group, since that is what the canvas
+    // draws and what the user can select. "View Connections" matches on the result, so every component
+    // type it offers relies on these.
+    describe('connection endpoint resolution', () => {
+        const GROUP_ID = 'group-being-viewed';
+        const CHILD_GROUP_ID = 'child-group';
+        const PARENT_GROUP_ID = 'parent-group';
+
+        function viewing(groupId: string): void {
+            const store = TestBed.inject(MockStore);
+            store.overrideSelector(selectCurrentProcessGroupId, groupId);
+            store.refreshState();
+        }
+
+        function connection(
+            sourceId: string,
+            sourceGroupId: string,
+            destinationId: string,
+            destinationGroupId: string
+        ): any {
+            return { id: 'conn', sourceId, sourceGroupId, destinationId, destinationGroupId };
+        }
+
+        it('resolves a component in the group being viewed to the component itself', () => {
+            viewing(GROUP_ID);
+
+            // a processor wired to a funnel, both drawn in the group being viewed
+            const conn = connection('proc-a', GROUP_ID, 'funnel-a', GROUP_ID);
+
+            expect(service.getConnectionSourceComponentId(conn)).toBe('proc-a');
+            expect(service.getConnectionDestinationComponentId(conn)).toBe('funnel-a');
+        });
+
+        it('resolves a port inside a child group to that child group', () => {
+            viewing(GROUP_ID);
+
+            // a processor in the group being viewed feeding an Input Port of a child group, and an
+            // Output Port of that same child group feeding back in
+            const into = connection('proc-a', GROUP_ID, 'inner-input-port', CHILD_GROUP_ID);
+            const outOf = connection('inner-output-port', CHILD_GROUP_ID, 'proc-a', GROUP_ID);
+
+            expect(service.getConnectionDestinationComponentId(into)).toBe(CHILD_GROUP_ID);
+            expect(service.getConnectionSourceComponentId(outOf)).toBe(CHILD_GROUP_ID);
+        });
+
+        it('resolves a port of the group being viewed to the port when the parent group is searched', () => {
+            // an Input Port's upstream connections are defined in the parent group, so that is the
+            // group whose flow gets searched — but the endpoint is still reported as belonging to the
+            // group being viewed, which is what makes it resolve to the port rather than to the group
+            viewing(GROUP_ID);
+
+            const upstreamOfInputPort = connection('parent-proc', PARENT_GROUP_ID, 'input-port', GROUP_ID);
+            const downstreamOfOutputPort = connection('output-port', GROUP_ID, 'parent-proc', PARENT_GROUP_ID);
+
+            expect(service.getConnectionDestinationComponentId(upstreamOfInputPort)).toBe('input-port');
+            expect(service.getConnectionSourceComponentId(downstreamOfOutputPort)).toBe('output-port');
+        });
+
+        it('resolves both ends of a self loop to the same component', () => {
+            viewing(GROUP_ID);
+
+            const retry = connection('proc-a', GROUP_ID, 'proc-a', GROUP_ID);
+
+            expect(service.getConnectionSourceComponentId(retry)).toBe('proc-a');
+            expect(service.getConnectionDestinationComponentId(retry)).toBe('proc-a');
+        });
+    });
 });

nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.actions.ts

@@ -24,6 +24,7 @@ import {
     ClearBulletinsForGroupResponse,
     ComponentEntity,
     ConfirmStopVersionControlRequest,
+    ComponentConnectionsDialogRequest,
     CreateComponentRequest,
     CreateComponentResponse,
     CreateConnection,
@@ -97,7 +98,8 @@ import {
     TerminateThreadsRequest,
     UpdatePositionsRequest,
     UploadProcessGroupRequest,
-    VersionControlInformationEntity
+    VersionControlInformationEntity,
+    ViewComponentConnectionsRequest
 } from './index';
 import { StatusHistoryRequest } from '../../../../state/status-history';
 import {
@@ -628,6 +630,16 @@ export const replayLastProvenanceEvent = createAction(
     props<{ request: ReplayLastProvenanceEventRequest }>()
 );
 
+export const viewComponentConnections = createAction(
+    `${CANVAS_PREFIX} View Component Connections`,
+    props<{ request: ViewComponentConnectionsRequest }>()
+);
+
+export const openComponentConnectionsDialog = createAction(
+    `${CANVAS_PREFIX} Open Component Connections Dialog`,
+    props<{ request: ComponentConnectionsDialogRequest }>()
+);
+
 export const enableComponent = createAction(
     `${CANVAS_PREFIX} Enable Component`,
     props<{ request: EnableComponentRequest | EnableProcessGroupRequest }>()

nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.effects.ts

@@ -53,6 +53,7 @@ import {
 } from 'rxjs';
 import {
     ComponentEntity,
+    ConnectionEntity,
     CreateConnectionDialogRequest,
     CreateProcessGroupDialogRequest,
     DeleteComponentResponse,
@@ -161,6 +162,7 @@ import { ChangeVersionDialog } from '../../ui/canvas/items/flow/change-version-d
 import { ChangeVersionProgressDialog } from '../../ui/canvas/items/flow/change-version-progress-dialog/change-version-progress-dialog';
 import { LocalChangesDialog } from '../../ui/canvas/items/flow/local-changes-dialog/local-changes-dialog';
 import { ProcessorBacklogDialog } from '../../ui/canvas/items/processor/backlog-dialog/backlog-dialog.component';
+import { ComponentConnectionsDialog } from '../../ui/canvas/component-connections-dialog/component-connections-dialog.component';
 import { ClusterConnectionService } from '../../../../service/cluster-connection.service';
 import { ExtensionTypesService } from '../../../../service/extension-types.service';
 import { ChangeComponentVersionDialog } from '../../../../ui/common/change-component-version-dialog/change-component-version-dialog';
@@ -3159,6 +3161,67 @@ export class FlowEffects {
         { dispatch: false }
     );
 
+    /**
+     * Loads the flow of the group that defines the requested component's connections and retains only
+     * the connections attached to that component in the requested direction. The group is the one on
+     * the canvas for most components, and the parent group for the two port cases whose connections
+     * cross the enclosing group's boundary.
+     *
+     * Matching goes through the canvas' own endpoint resolvers rather than comparing the raw ids. A
+     * connection drawn to a Process Group or Remote Process Group actually terminates at a port inside
+     * it, and getConnectionSourceComponentId/getConnectionDestinationComponentId are what collapse that
+     * port back to the group the user sees and selects. They compare against the group currently on the
+     * canvas, which is the right frame of reference for the parent-group searches too: a connection
+     * into an Input Port carries that port's own group as its destination group, so it resolves to the
+     * port rather than to the group.635
+     *
+     *
+     * Both resolvers read the ids on the connection entity rather than on its component, so a
+     * connection the current user cannot read — the kind most worth reporting — is still matched.
+     */
+    viewComponentConnections$ = createEffect(() =>
+        this.actions$.pipe(
+            ofType(FlowActions.viewComponentConnections),
+            map((action) => action.request),
+            switchMap((request) => {
+                const attachedTo = (connection: ConnectionEntity): boolean =>
+                    request.direction === 'upstream'
+                        ? this.canvasUtils.getConnectionDestinationComponentId(connection) === request.id
+                        : this.canvasUtils.getConnectionSourceComponentId(connection) === request.id;
+
+                return from(this.flowService.getFlow(request.groupId)).pipe(
+                    map((flowEntity: ProcessGroupFlowEntity) =>
+                        FlowActions.openComponentConnectionsDialog({
+                            request: {
+                                componentName: request.name,
+                                componentType: request.type,
+                                groupId: request.groupId,
+                                direction: request.direction,
+                                connections: flowEntity.processGroupFlow.flow.connections.filter(attachedTo)
+                            }
+                        })
+                    ),
+                    catchError((errorResponse: HttpErrorResponse) => of(this.snackBarOrFullScreenError(errorResponse)))
+                );
+            })
+        )
+    );
+
+    openComponentConnectionsDialog$ = createEffect(
+        () =>
+            this.actions$.pipe(
+                ofType(FlowActions.openComponentConnectionsDialog),
+                map((action) => action.request),
+                tap((request) => {
+                    this.dialog.open(ComponentConnectionsDialog, {
+                        ...XL_DIALOG,
+                        data: request
+                    });
+                })
+            ),
+        { dispatch: false }
+    );
+
     showOkDialog$ = createEffect(
         () =>
             this.actions$.pipe(
@@ -4866,7 +4929,7 @@ export class FlowEffects {
                 warnedIds: this.warnedPositionIds
             })
         });
-        const sanitizeConnection = (entity: ComponentEntity): ComponentEntity => ({
+        const sanitizeConnection = (entity: ConnectionEntity): ConnectionEntity => ({
             ...entity,
             position: sanitizePosition(entity.position, {
                 componentId: entity.id,

nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.reducer.ts

@@ -712,23 +712,23 @@ function getComponentCollection(draftState: FlowState, componentType: ComponentT
     return collection;
 }
 
-function processComponentCollection(
-    proposedComponents: ComponentEntity[],
-    currentComponents: ComponentEntity[],
+function processComponentCollection<T extends ComponentEntity>(
+    proposedComponents: T[],
+    currentComponents: T[],
     addedCache: string[],
     removedCache: string[],
     overrideRevisionCheck: boolean
-): ComponentEntity[] {
+): T[] {
     // components in the proposed collection but not the current collection
-    const addedComponents: ComponentEntity[] = proposedComponents.filter((proposedComponent) => {
+    const addedComponents: T[] = proposedComponents.filter((proposedComponent) => {
         return !currentComponents.some((currentComponent) => currentComponent.id === proposedComponent.id);
     });
     // components in the current collection that are no longer in the proposed collection
-    const removedComponents: ComponentEntity[] = currentComponents.filter((currentComponent) => {
+    const removedComponents: T[] = currentComponents.filter((currentComponent) => {
         return !proposedComponents.some((proposedComponent) => proposedComponent.id === currentComponent.id);
     });
     // components that are in both the proposed collection and the current collection
-    const updatedComponents: ComponentEntity[] = currentComponents.filter((currentComponent) => {
+    const updatedComponents: T[] = currentComponents.filter((currentComponent) => {
         return proposedComponents.some((proposedComponents) => proposedComponents.id === currentComponent.id);
     });
 

nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.selectors.ts

@@ -19,6 +19,7 @@ import { flowFeatureKey, FlowState, SelectedComponent } from './index';
 import { createSelector } from '@ngrx/store';
 import { CanvasState, selectCanvasState } from '../index';
 import { ComponentType, selectCurrentRoute } from '@nifi/shared';
+import { BreadcrumbEntity } from '../../../../state/shared';
 import {
     detectOverlappingConnections,
     OverlappingConnectionGroup
@@ -294,3 +295,35 @@ export const selectOverlappingConnections = createSelector(
         return detectOverlappingConnections(connections, processGroupId);
     }
 );
+
+// maps the id of every group whose name is known in the current flow to that group's name: the current
+// process group, its ancestors via the breadcrumb chain, and the child process groups and remote process
+// groups it contains. Remote process groups are included alongside process groups because a connection to
+// one reports the remote process group's own id as the source/destination group id of the remote port it
+// terminates at, so the two kinds of id are looked up the same way.
+export const selectProcessGroupIdToNameMap = createSelector(
+    selectBreadcrumbs,
+    selectProcessGroups,
+    selectRemoteProcessGroups,
+    (breadcrumbs: BreadcrumbEntity, processGroups: any[], remoteProcessGroups: any[]): Map<string, string> => {
+        const idToName = new Map<string, string>();
+
+        // the current process group and each of its ancestors
+        let breadcrumbEntity: BreadcrumbEntity | undefined = breadcrumbs;
+        while (breadcrumbEntity) {
+            if (breadcrumbEntity.permissions.canRead) {
+                idToName.set(breadcrumbEntity.id, breadcrumbEntity.breadcrumb.name);
+            }
+            breadcrumbEntity = breadcrumbEntity.parentBreadcrumb;
+        }
+
+        // any child process groups or remote process groups of the current process group
+        [...(processGroups ?? []), ...(remoteProcessGroups ?? [])].forEach((group) => {
+            if (group.permissions.canRead) {
+                idToName.set(group.id, group.component.name);
+            }
+        });
+
+        return idToName;
+    }
+);