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

@@ -50,16 +50,18 @@ import {
     stopSources,
     stopVersionControlRequest,
     terminateThreads,
-    updatePositions
+    updatePositions,
+    viewComponentConnections
 } from '../state/flow/flow.actions';
 import { ComponentType } from '@nifi/shared';
 import {
     ConfirmStopVersionControlRequest,
+    ConnectionDirection,
     MoveComponentRequest,
     OpenChangeVersionDialogRequest,
     OpenLocalChangesDialogRequest
 } from '../state/flow';
-import { UpdateComponentRequest } from '../../../state/shared';
+import { BreadcrumbEntity, UpdateComponentRequest } from '../../../state/shared';
 import {
     ContextMenuDefinition,
     ContextMenuDefinitionProvider,
@@ -287,25 +289,31 @@ export class CanvasContextMenu implements ContextMenuDefinitionProvider {
         id: 'upstream-downstream',
         menuItems: [
             {
-                condition: () => {
-                    // TODO - hasUpstream
-                    return false;
+                condition: (selection: d3.Selection<any, any, any, any>) => {
+                    // an empty selection targets the current process group, whose connections are defined in its
+                    // parent, so there is nothing to report from the root process group since it has no parent
+                    return (
+                        this.canvasUtils.hasUpstream(selection) ||
+                        this.canvasUtils.isNotRootGroupAndEmptySelection(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) ||
+                        this.canvasUtils.isNotRootGroupAndEmptySelection(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');
                 }
             }
         ]
@@ -1465,4 +1473,70 @@ 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 component whose connections cross that group's own boundary and
+     * are defined one level up: the upstream side of an Input Port, the downstream side of an Output
+     * Port, and either side of the current group itself. Those are the cases that
+     * hasUpstream/hasDownstream gate on the presence of a parent group.
+     *
+     * An empty selection means the user did not select a component, which implicitly targets the current
+     * group.
+     */
+    private requestComponentConnections(
+        selection: d3.Selection<any, any, any, any>,
+        direction: ConnectionDirection
+    ): void {
+        const currentProcessGroupTargeted: boolean = this.canvasUtils.emptySelection(selection);
+
+        const crossesParentBoundary: boolean =
+            currentProcessGroupTargeted ||
+            (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 = currentProcessGroupTargeted ? this.currentProcessGroupDatum() : 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
+                }
+            })
+        );
+    }
+
+    /**
+     * Returns the current group shaped like the datum of a component rendered on the canvas, so that it
+     * can be reported the same way a selected component is. The current group draws the canvas itself
+     * rather than a node on it, so there is no selection to read this from.
+     */
+    private currentProcessGroupDatum(): any {
+        const breadcrumb: BreadcrumbEntity | null = this.canvasUtils.getCurrentProcessGroupBreadcrumb();
+
+        return {
+            id: this.canvasUtils.getProcessGroupId(),
+            type: ComponentType.ProcessGroup,
+            permissions: breadcrumb ? breadcrumb.permissions : { canRead: false, canWrite: false },
+            component: { name: breadcrumb?.breadcrumb.name }
+        };
+    }
 }

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/service/canvas-utils.service.ts

@@ -255,6 +255,14 @@ export class CanvasUtils {
         return this.parentProcessGroupId;
     }
 
+    /**
+     * Returns the breadcrumb of the current group, which carries its name and permissions. The current
+     * group is not rendered on its own canvas, so this is the only source for those details.
+     */
+    public getCurrentProcessGroupBreadcrumb(): BreadcrumbEntity | null {
+        return this.breadcrumbs;
+    }
+
     /**
      * Returns the current parameter context id or null if there is no bound parameter context.
      */

nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/component-connections.utils.spec.ts

@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { buildComponentIdToNameMap, collectEndpointGroupIds } from './component-connections.utils';
+import { ComponentEntity, ConnectionEntity, ProcessGroupFlowEntity } from './index';
+
+const DEFINING_GROUP_ID = 'defining-group-id';
+const CHILD_GROUP_ID = 'child-group-id';
+const OTHER_CHILD_GROUP_ID = 'other-child-group-id';
+const REMOTE_GROUP_ID = 'remote-process-group-id';
+
+interface EndpointOptions {
+    sourceGroupId?: string;
+    sourceType?: string;
+    destinationGroupId?: string;
+    destinationType?: string;
+}
+
+function connection(options: EndpointOptions = {}): ConnectionEntity {
+    return {
+        id: 'connection-id',
+        permissions: { canRead: false, canWrite: false },
+        position: { x: 0, y: 0 },
+        revision: { version: 0 },
+        sourceId: 'source-id',
+        sourceGroupId: options.sourceGroupId ?? DEFINING_GROUP_ID,
+        sourceType: options.sourceType ?? 'PROCESSOR',
+        destinationId: 'destination-id',
+        destinationGroupId: options.destinationGroupId ?? DEFINING_GROUP_ID,
+        destinationType: options.destinationType ?? 'PROCESSOR',
+        component: null
+    };
+}
+
+function component(id: string, name: string | undefined, canRead: boolean): ComponentEntity {
+    return {
+        id,
+        permissions: { canRead, canWrite: false },
+        position: { x: 0, y: 0 },
+        revision: { version: 0 },
+        component: { id, name }
+    };
+}
+
+function flowEntity(flow: Partial<ProcessGroupFlowEntity['processGroupFlow']['flow']>): ProcessGroupFlowEntity {
+    return {
+        permissions: { canRead: true, canWrite: true },
+        processGroupFlow: {
+            id: DEFINING_GROUP_ID,
+            uri: '',
+            parentGroupId: null,
+            breadcrumb: {
+                id: DEFINING_GROUP_ID,
+                permissions: { canRead: true, canWrite: true },
+                versionedFlowState: '',
+                breadcrumb: { id: DEFINING_GROUP_ID, name: 'Defining Process Group' }
+            },
+            flow: {
+                processGroups: [],
+                remoteProcessGroups: [],
+                processors: [],
+                inputPorts: [],
+                outputPorts: [],
+                connections: [],
+                labels: [],
+                funnels: [],
+                ...flow
+            }
+        }
+    } as unknown as ProcessGroupFlowEntity;
+}
+
+describe('collectEndpointGroupIds', () => {
+    it('collects nothing when both ends are in the group that defines the connections', () => {
+        expect(collectEndpointGroupIds([connection()], DEFINING_GROUP_ID)).toEqual([]);
+    });
+
+    it('collects the group behind a port an end reaches into', () => {
+        const connections = [
+            connection({ destinationGroupId: CHILD_GROUP_ID, destinationType: 'INPUT_PORT' }),
+            connection({ sourceGroupId: OTHER_CHILD_GROUP_ID, sourceType: 'OUTPUT_PORT' })
+        ];
+
+        expect(collectEndpointGroupIds(connections, DEFINING_GROUP_ID)).toEqual([CHILD_GROUP_ID, OTHER_CHILD_GROUP_ID]);
+    });
+
+    it('collects a group reached by several connections once', () => {
+        const connections = [
+            connection({ destinationGroupId: CHILD_GROUP_ID, destinationType: 'INPUT_PORT' }),
+            connection({ destinationGroupId: CHILD_GROUP_ID, destinationType: 'INPUT_PORT' }),
+            connection({ sourceGroupId: CHILD_GROUP_ID, sourceType: 'OUTPUT_PORT' })
+        ];
+
+        expect(collectEndpointGroupIds(connections, DEFINING_GROUP_ID)).toEqual([CHILD_GROUP_ID]);
+    });
+
+    it('leaves out a Remote Process Group, which has no flow of its own to load', () => {
+        const connections = [
+            connection({ sourceGroupId: REMOTE_GROUP_ID, sourceType: 'REMOTE_OUTPUT_PORT' }),
+            connection({ destinationGroupId: REMOTE_GROUP_ID, destinationType: 'REMOTE_INPUT_PORT' })
+        ];
+
+        expect(collectEndpointGroupIds(connections, DEFINING_GROUP_ID)).toEqual([]);
+    });
+});
+
+describe('buildComponentIdToNameMap', () => {
+    it('names only the components the current user can read', () => {
+        const flow = flowEntity({
+            processors: [
+                component('readable-processor-id', 'Readable Processor', true),
+                component('hidden-processor-id', 'Hidden Processor', false)
+            ]
+        });
+
+        const idToName = buildComponentIdToNameMap([flow]);
+
+        expect(idToName.get('readable-processor-id')).toBe('Readable Processor');
+        expect(idToName.has('hidden-processor-id')).toBeFalsy();
+    });
+
+    it('names components of every group it is given, which is what separates the two ends', () => {
+        const definingFlow = flowEntity({
+            processors: [component('hidden-processor-id', 'Hidden Processor', false)]
+        });
+        const childFlow = flowEntity({
+            inputPorts: [component('input-port-id', 'Input Port A', true)]
+        });
+
+        const idToName = buildComponentIdToNameMap([definingFlow, childFlow]);
+
+        expect(idToName.has('hidden-processor-id')).toBeFalsy();
+        expect(idToName.get('input-port-id')).toBe('Input Port A');
+    });
+
+    it('names every kind of component a connection can be attached to', () => {
+        const flow = flowEntity({
+            processors: [component('processor-id', 'Processor', true)],
+            inputPorts: [component('input-port-id', 'Input Port', true)],
+            outputPorts: [component('output-port-id', 'Output Port', true)],
+            funnels: [component('funnel-id', undefined, true)]
+        });
+
+        const idToName = buildComponentIdToNameMap([flow]);
+
+        expect(idToName.get('processor-id')).toBe('Processor');
+        expect(idToName.get('input-port-id')).toBe('Input Port');
+        expect(idToName.get('output-port-id')).toBe('Output Port');
+        // a funnel has no name of its own, and is still reported as readable
+        expect(idToName.get('funnel-id')).toBe('');
+    });
+
+    it('builds an empty map when no flow could be loaded', () => {
+        expect(buildComponentIdToNameMap([]).size).toBe(0);
+    });
+});