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 632b479dbf8 NIFI-16408 Cancel superseded registry resource requests in
flow registry dialogs (#11738)
632b479dbf8 is described below
commit 632b479dbf8f60dfd2b916a538e36b8410acad26
Author: boguszj <[email protected]>
AuthorDate: Wed Sep 30 18:54:46 2026 +0200
NIFI-16408 Cancel superseded registry resource requests in flow registry
dialogs (#11738)
This closes #11738
---
.../import-from-registry.component.spec.ts | 255 +++++++++++++++-
.../import-from-registry.component.ts | 340 ++++++++++++---------
.../save-version-dialog.component.spec.ts | 162 +++++++++-
.../save-version-dialog.component.ts | 150 +++++----
4 files changed, 694 insertions(+), 213 deletions(-)
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/import-from-registry/import-from-registry.component.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/import-from-registry/import-from-registry.component.spec.ts
index bcc87085c8d..7d5ce5647de 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/import-from-registry/import-from-registry.component.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/import-from-registry/import-from-registry.component.spec.ts
@@ -18,12 +18,11 @@
import { ComponentFixture, TestBed } from '@angular/core/testing';
import { ImportFromRegistry } from './import-from-registry.component';
-import { ImportFromRegistryDialogRequest } from '../../../../../state/flow';
+import { flowFeatureKey, ImportFromRegistryDialogRequest } from
'../../../../../state/flow';
import { MAT_DIALOG_DATA, MatDialogRef } from '@angular/material/dialog';
import { ComponentType } from '@nifi/shared';
import { provideMockStore } from '@ngrx/store/testing';
import { initialState as initialFlowState } from
'../../../../../state/flow/flow.reducer';
-import { flowFeatureKey } from '../../../../../state/flow';
import { canvasFeatureKey } from '../../../../../state';
import { initialState as initialErrorState } from
'../../../../../../../state/error/error.reducer';
import { errorFeatureKey } from '../../../../../../../state/error';
@@ -32,7 +31,15 @@ import { currentUserFeatureKey } from
'../../../../../../../state/current-user';
import { flowConfigurationFeatureKey } from
'../../../../../../../state/flow-configuration';
import { initialState as flowConfigurationInitialState } from
'../../../../../../../state/flow-configuration/flow-configuration.reducer';
import { NoopAnimationsModule } from '@angular/platform-browser/animations';
-import { EMPTY } from 'rxjs';
+import { EMPTY, Observable, Subject } from 'rxjs';
+import {
+ BranchEntity,
+ BucketEntity,
+ RegistryClientEntity,
+ VersionedFlowEntity,
+ VersionedFlowSnapshotMetadataEntity
+} from '../../../../../../../state/shared';
+import { HttpErrorResponse } from '@angular/common/http';
import { ClusterConnectionService } from
'../../../../../../../service/cluster-connection.service';
import { By } from '@angular/platform-browser';
@@ -194,4 +201,246 @@ describe('ImportFromRegistry', () => {
expect(error).toBeFalsy();
expect(versions).toBeTruthy();
});
+
+ describe('with multiple registry clients', () => {
+ const registryAId = 'registry-a';
+ const registryBId = 'registry-b';
+ const bucketAId = 'bucket-a';
+ const bucketBId = 'bucket-b';
+
+ let branchRequests: Map<string, Subject<BranchEntity[]>>;
+ let bucketRequests: Map<string, Subject<BucketEntity[]>>;
+ let flowRequests: Map<string, Subject<VersionedFlowEntity[]>>;
+ let cancelledBucketRequests: string[];
+
+ const createRegistryClient = (id: string, name: string,
supportsBranching = false): RegistryClientEntity => {
+ const template = data.registryClients[0];
+ return {
+ ...template,
+ id,
+ component: {
+ ...template.component,
+ id,
+ name,
+ supportsBranching
+ }
+ };
+ };
+
+ const createBucket = (id: string): BucketEntity => ({
+ id,
+ permissions: { canRead: true, canWrite: true },
+ bucket: { id, created: 0, name: `bucket in ${id}`, description: ''
}
+ });
+
+ const createBranch = (registryId: string): BranchEntity =>
+ ({
+ branch: { name: `branch in ${registryId}` }
+ }) as BranchEntity;
+
+ const createFlow = (bucketId: string): VersionedFlowEntity =>
+ ({
+ versionedFlow: {
+ flowId: `flow in ${bucketId}`,
+ flowName: `flow in ${bucketId}`,
+ description: ''
+ }
+ }) as VersionedFlowEntity;
+
+ /**
+ * Registers a request per key so that the test controls when each one
responds, and records the
+ * keys whose requests were torn down before completing.
+ */
+ const controllableRequest = <T>(requests: Map<string, Subject<T>>,
cancelled?: string[]) => {
+ return (key: string) =>
+ new Observable<T>((subscriber) => {
+ const request = new Subject<T>();
+ requests.set(key, request);
+ const subscription = request.subscribe(subscriber);
+ return () => {
+ if (!request.closed) {
+ cancelled?.push(key);
+ }
+ subscription.unsubscribe();
+ };
+ });
+ };
+
+ const createDialog = (supportsBranching: boolean) => {
+ TestBed.resetTestingModule();
+ TestBed.configureTestingModule({
+ imports: [ImportFromRegistry, NoopAnimationsModule],
+ providers: [
+ {
+ provide: MAT_DIALOG_DATA,
+ useValue: {
+ ...data,
+ registryClients: [
+ createRegistryClient(registryAId, 'A
Registry', supportsBranching),
+ createRegistryClient(registryBId, 'B
Registry', supportsBranching)
+ ]
+ }
+ },
+ provideMockStore({
+ initialState: {
+ [errorFeatureKey]: initialErrorState,
+ [currentUserFeatureKey]: initialCurrentUserState,
+ [flowConfigurationFeatureKey]:
flowConfigurationInitialState,
+ [canvasFeatureKey]: {
+ [flowFeatureKey]: initialFlowState
+ }
+ }
+ }),
+ {
+ provide: ClusterConnectionService,
+ useValue: {
+ isDisconnectionAcknowledged: vi.fn()
+ }
+ },
+ { provide: MatDialogRef, useValue: null }
+ ]
+ });
+ fixture = TestBed.createComponent(ImportFromRegistry);
+ component = fixture.componentInstance;
+ component.getBranches = controllableRequest(branchRequests);
+ component.getBuckets = controllableRequest(bucketRequests,
cancelledBucketRequests);
+ component.getFlows = (registryId: string, bucketId: string) =>
controllableRequest(flowRequests)(bucketId);
+ component.getFlowVersions = () => EMPTY;
+
+ fixture.detectChanges();
+ };
+
+ beforeEach(() => {
+ branchRequests = new Map<string, Subject<BranchEntity[]>>();
+ bucketRequests = new Map<string, Subject<BucketEntity[]>>();
+ flowRequests = new Map<string, Subject<VersionedFlowEntity[]>>();
+ cancelledBucketRequests = [];
+ });
+
+ describe('that do not support branching', () => {
+ beforeEach(() => {
+ createDialog(false);
+ });
+
+ it('should ignore a bucket response for a registry that is no
longer selected', () => {
+ component.registryChanged(registryBId);
+
+
bucketRequests.get(registryAId)?.next([createBucket(registryAId)]);
+
+ expect(component.bucketOptions).toEqual([]);
+
expect(component.importFromRegistryForm.get('bucket')?.value).toBeNull();
+
+
bucketRequests.get(registryBId)?.next([createBucket(registryBId)]);
+
+ expect(component.bucketOptions.map((option) =>
option.value)).toEqual([registryBId]);
+
expect(component.importFromRegistryForm.get('bucket')?.value).toEqual(registryBId);
+ });
+
+ it('should cancel an outstanding bucket request when the selected
registry changes', () => {
+ expect(cancelledBucketRequests).toEqual([]);
+
+ component.registryChanged(registryBId);
+
+ expect(cancelledBucketRequests).toEqual([registryAId]);
+ });
+
+ it('should still load buckets for a later registry selection after
a failed bucket request', () => {
+ bucketRequests.get(registryAId)?.error(new Error('bucket
request failed'));
+
+ expect(component.loadingBuckets()).toBe(false);
+
+ component.registryChanged(registryBId);
+
bucketRequests.get(registryBId)?.next([createBucket(registryBId)]);
+
+ expect(component.bucketOptions.map((option) =>
option.value)).toEqual([registryBId]);
+
expect(component.importFromRegistryForm.get('bucket')?.value).toEqual(registryBId);
+ });
+
+ it('should still load flows for a later bucket selection after a
failed flow request', () => {
+ component.bucketChanged(bucketAId);
+ flowRequests.get(bucketAId)?.error(new Error('flow request
failed'));
+
+ expect(component.loadingFlows()).toBe(false);
+
+ component.bucketChanged(bucketBId);
+ flowRequests.get(bucketBId)?.next([createFlow(bucketBId)]);
+
+ expect(component.flowOptions.map((option) =>
option.value)).toEqual([`flow in ${bucketBId}`]);
+
expect(component.importFromRegistryForm.get('flow')?.value).toEqual(`flow in
${bucketBId}`);
+ });
+ });
+
+ describe('that support branching', () => {
+ beforeEach(() => {
+ createDialog(true);
+ });
+
+ it('should ignore a branch response for a registry that is no
longer selected', () => {
+ component.registryChanged(registryBId);
+
+
branchRequests.get(registryAId)?.next([createBranch(registryAId)]);
+
+ expect(component.branchOptions).toEqual([]);
+
expect(component.importFromRegistryForm.get('branch')?.value).toBeNull();
+ });
+
+ it('should still load branches for a later registry selection
after a failed branch request', () => {
+ branchRequests.get(registryAId)?.error(new Error('branch
request failed'));
+
+ expect(component.loadingBranches()).toBe(false);
+
+ component.registryChanged(registryBId);
+
branchRequests.get(registryBId)?.next([createBranch(registryBId)]);
+
+ expect(component.branchOptions.map((option) =>
option.value)).toEqual([`branch in ${registryBId}`]);
+
expect(component.importFromRegistryForm.get('branch')?.value).toEqual(`branch
in ${registryBId}`);
+ });
+
+ it('should re-enable the bucket control when its request is
cancelled without a replacement', () => {
+ // responding to the branch request auto-selects a branch,
which issues the bucket request
+
branchRequests.get(registryAId)?.next([createBranch(registryAId)]);
+
+ expect(component.loadingBuckets()).toBe(true);
+
expect(component.importFromRegistryForm.get('bucket')?.disabled).toBe(true);
+
+ // changing registry re-issues only the branch level, so
nothing replaces the cancelled
+ // bucket request; the bucket control must not be left disabled
+ component.registryChanged(registryBId);
+
+ expect(component.loadingBuckets()).toBe(false);
+
expect(component.importFromRegistryForm.get('bucket')?.disabled).toBe(false);
+ });
+ });
+ });
+
+ describe('when a versions request is superseded', () => {
+ let versionRequests: Subject<VersionedFlowSnapshotMetadataEntity[]>;
+
+ beforeEach(() => {
+ versionRequests = new
Subject<VersionedFlowSnapshotMetadataEntity[]>();
+ component.getFlowVersions = () => versionRequests;
+ });
+
+ it('should clear the versions loading state when the request is
cancelled without a replacement', () => {
+ component.flowChanged('flow-1');
+ expect(component.loadingVersions()).toBe(true);
+
+ // selecting another bucket clears the flow selection, so no
replacement versions request follows
+ component.bucketChanged('bucket-1');
+
+ expect(component.loadingVersions()).toBe(false);
+ });
+
+ it('should clear a previous versions error once the flow selection is
cleared', () => {
+ component.flowChanged('flow-1');
+ versionRequests.error(new HttpErrorResponse({ status: 409,
statusText: 'Conflict' }));
+
+ expect(component.loadingVersions()).toBe(false);
+ expect(component.loadingVersionsError()).not.toBeNull();
+
+ component.bucketChanged('bucket-1');
+
+ expect(component.loadingVersionsError()).toBeNull();
+ });
+ });
});
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/import-from-registry/import-from-registry.component.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/import-from-registry/import-from-registry.component.ts
index 0661e7cb21d..fac4b9a6a16 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/import-from-registry/import-from-registry.component.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/import-from-registry/import-from-registry.component.ts
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import { Component, Input, OnInit, signal, WritableSignal, inject } from
'@angular/core';
+import { Component, inject, Input, OnInit, signal, WritableSignal } from
'@angular/core';
import { MAT_DIALOG_DATA, MatDialogModule } from '@angular/material/dialog';
import { ImportFromRegistryDialogRequest } from '../../../../../state/flow';
import { Store } from '@ngrx/store';
@@ -38,18 +38,18 @@ import { MatOptionModule } from '@angular/material/core';
import { MatSelectModule } from '@angular/material/select';
import { FormBuilder, FormControl, FormGroup, ReactiveFormsModule, Validators
} from '@angular/forms';
import { MatIconModule } from '@angular/material/icon';
-import { catchError, EMPTY, Observable, of, take } from 'rxjs';
+import { catchError, EMPTY, finalize, map, Observable, of, Subject, switchMap,
take } from 'rxjs';
import { MatCheckboxModule } from '@angular/material/checkbox';
import { MatTableDataSource, MatTableModule } from '@angular/material/table';
import { MatSortModule, Sort } from '@angular/material/sort';
import {
+ CloseOnEscapeDialog,
isDefinedAndNotNull,
- SelectOption,
NiFiCommon,
- TextTip,
+ NifiSpinnerDirective,
NifiTooltipDirective,
- CloseOnEscapeDialog,
- NifiSpinnerDirective
+ SelectOption,
+ TextTip
} from '@nifi/shared';
import { selectTimeOffset } from
'../../../../../../../state/flow-configuration/flow-configuration.selectors';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
@@ -62,6 +62,19 @@ import { ErrorHelper } from
'../../../../../../../service/error-helper.service';
import { HttpErrorResponse } from '@angular/common/http';
import { NgxSkeletonLoaderComponent } from 'ngx-skeleton-loader';
+interface LoadBucketsRequest {
+ registryId: string;
+ branch: string | null;
+}
+
+interface LoadFlowsRequest extends LoadBucketsRequest {
+ bucketId: string;
+}
+
+interface LoadVersionsRequest extends LoadFlowsRequest {
+ flowId: string;
+}
+
@Component({
selector: 'import-from-registry',
imports: [
@@ -139,6 +152,11 @@ export class ImportFromRegistry extends
CloseOnEscapeDialog implements OnInit {
loadingVersions: WritableSignal<boolean> = signal(false);
loadingVersionsError: WritableSignal<string | null> = signal(null);
+ private loadBranchesRequest = new Subject<string | null>();
+ private loadBucketsRequest = new Subject<LoadBucketsRequest | null>();
+ private loadFlowsRequest = new Subject<LoadFlowsRequest | null>();
+ private loadVersionsRequest = new Subject<LoadVersionsRequest | null>();
+
constructor() {
super();
const dialogRequest = this.dialogRequest;
@@ -176,46 +194,199 @@ export class ImportFromRegistry extends
CloseOnEscapeDialog implements OnInit {
flow: new FormControl(null, Validators.required),
keepParameterContexts: new FormControl(true, Validators.required)
});
+
+ this.loadBranchesRequest
+ .pipe(
+ switchMap((registryId: string | null) => {
+ if (!registryId) {
+ return EMPTY;
+ }
+
+ this.setLoading('branch', true);
+ this.branchOptions = [];
+
+ return this.getBranches(registryId).pipe(
+ take(1),
+ map((branches: BranchEntity[]) => ({ registryId,
branches })),
+ catchError(() => EMPTY),
+ finalize(() => this.setLoading('branch', false))
+ );
+ }),
+ takeUntilDestroyed()
+ )
+ .subscribe(({ registryId, branches }) => {
+ branches.forEach((entity: BranchEntity) => {
+ this.branchOptions.push({
+ text: entity.branch.name,
+ value: entity.branch.name
+ });
+ });
+
+ const branch = this.branchOptions[0]?.value;
+ if (branch) {
+
this.importFromRegistryForm.get('branch')?.setValue(branch);
+ this.loadBucketsRequest.next({ registryId, branch });
+ }
+ });
+
+ this.loadBucketsRequest
+ .pipe(
+ switchMap((request: LoadBucketsRequest | null) => {
+ if (!request) {
+ return EMPTY;
+ }
+
+ this.setLoading('bucket', true);
+ this.bucketOptions = [];
+
+ return this.getBuckets(request.registryId,
request.branch).pipe(
+ take(1),
+ map((buckets: BucketEntity[]) => ({ request, buckets
})),
+ catchError(() => EMPTY),
+ finalize(() => this.setLoading('bucket', false))
+ );
+ }),
+ takeUntilDestroyed()
+ )
+ .subscribe(({ request, buckets }) => {
+ buckets.forEach((entity: BucketEntity) => {
+ if (entity.permissions.canRead) {
+ this.bucketOptions.push({
+ text: entity.bucket.name,
+ value: entity.id,
+ description: entity.bucket.description
+ });
+ }
+ });
+
+ const bucketId = this.bucketOptions[0]?.value;
+ if (bucketId) {
+
this.importFromRegistryForm.get('bucket')?.setValue(bucketId);
+ this.loadFlowsRequest.next({ ...request, bucketId });
+ }
+ });
+
+ this.loadFlowsRequest
+ .pipe(
+ switchMap((request: LoadFlowsRequest | null) => {
+ if (!request) {
+ return EMPTY;
+ }
+
+ this.setLoading('flow', true);
+ this.flowOptions = [];
+ this.flowLookup.clear();
+
+ return this.getFlows(request.registryId, request.bucketId,
request.branch).pipe(
+ take(1),
+ map((versionedFlows: VersionedFlowEntity[]) => ({
request, versionedFlows })),
+ catchError(() => EMPTY),
+ finalize(() => this.setLoading('flow', false))
+ );
+ }),
+ takeUntilDestroyed()
+ )
+ .subscribe(({ request, versionedFlows }) => {
+ versionedFlows.forEach((entity: VersionedFlowEntity) => {
+ this.flowLookup.set(entity.versionedFlow.flowId!,
entity.versionedFlow);
+
+ this.flowOptions.push({
+ text: entity.versionedFlow.flowName,
+ value: entity.versionedFlow.flowId!,
+ description: entity.versionedFlow.description
+ });
+ });
+
+ const flowId = this.flowOptions[0]?.value;
+ if (flowId) {
+ this.importFromRegistryForm.get('flow')?.setValue(flowId);
+ this.loadVersionsRequest.next({ ...request, flowId });
+ }
+ });
+
+ this.loadVersionsRequest
+ .pipe(
+ switchMap((request: LoadVersionsRequest | null) => {
+ if (!request) {
+ this.loadingVersionsError.set(null);
+ return EMPTY;
+ }
+
+ this.setLoading('version', true);
+ this.loadingVersionsError.set(null);
+ this.dataSource.data = [];
+ this.selectedFlowVersion = null;
+ this.selectedFlowDescription =
this.flowLookup.get(request.flowId)?.description;
+
+ return this.getFlowVersions(
+ request.registryId,
+ request.bucketId,
+ request.flowId,
+ request.branch
+ ).pipe(
+ take(1),
+ catchError((errorResponse: HttpErrorResponse) => {
+
this.loadingVersionsError.set(this.errorHelper.getErrorString(errorResponse));
+ return EMPTY;
+ }),
+ finalize(() => this.setLoading('version', false))
+ );
+ }),
+ takeUntilDestroyed()
+ )
+ .subscribe((metadataEntities:
VersionedFlowSnapshotMetadataEntity[]) => {
+ if (metadataEntities.length > 0) {
+ const flowVersions = metadataEntities.map(
+ (entity: VersionedFlowSnapshotMetadataEntity) =>
entity.versionedFlowSnapshotMetadata
+ );
+
+ const sortedFlowVersions = this.sortVersions(flowVersions,
this.sort);
+ this.selectedFlowVersion = sortedFlowVersions[0].version;
+
+ this.dataSource.data = sortedFlowVersions;
+ }
+ });
}
ngOnInit(): void {
const selectedRegistryId =
this.importFromRegistryForm.get('registry')?.value;
if (selectedRegistryId) {
- this.supportsBranching =
this.clientBranchingSupportMap.get(selectedRegistryId) || false;
- if (this.supportsBranching) {
- this.loadBranches(selectedRegistryId);
- } else {
- this.loadBuckets(selectedRegistryId);
- }
+ this.registrySelected(selectedRegistryId);
}
}
registryChanged(registryId: string): void {
+ this.registrySelected(registryId);
+ }
+
+ private registrySelected(registryId: string): void {
this.supportsBranching =
this.clientBranchingSupportMap.get(registryId) || false;
this.clearBranches();
if (this.supportsBranching) {
- this.loadBranches(registryId);
+ this.loadBranchesRequest.next(registryId);
} else {
- this.loadBuckets(registryId);
+ this.loadBucketsRequest.next({ registryId, branch: null });
}
}
private clearBranches(): void {
this.branchOptions = [];
this.importFromRegistryForm.get('branch')?.setValue(null);
+ this.loadBranchesRequest.next(null);
this.clearBuckets();
}
branchChanged(branch: string): void {
this.clearBuckets();
const registryId = this.importFromRegistryForm.get('registry')?.value;
- this.loadBuckets(registryId, branch);
+ this.loadBucketsRequest.next({ registryId, branch });
}
private clearBuckets(): void {
this.bucketOptions = [];
this.importFromRegistryForm.get('bucket')?.setValue(null);
+ this.loadBucketsRequest.next(null);
this.clearFlows();
}
@@ -223,7 +394,7 @@ export class ImportFromRegistry extends CloseOnEscapeDialog
implements OnInit {
this.clearFlows();
const registryId = this.importFromRegistryForm.get('registry')?.value;
const branch = this.importFromRegistryForm.get('branch')?.value;
- this.loadFlows(registryId, bucketId, branch);
+ this.loadFlowsRequest.next({ registryId, branch, bucketId });
}
private clearFlows() {
@@ -231,13 +402,16 @@ export class ImportFromRegistry extends
CloseOnEscapeDialog implements OnInit {
this.flowOptions = [];
this.dataSource.data = [];
this.selectedFlowVersion = null;
+ this.loadingVersionsError.set(null);
+ this.loadFlowsRequest.next(null);
+ this.loadVersionsRequest.next(null);
}
flowChanged(flowId: string): void {
const registryId = this.importFromRegistryForm.get('registry')?.value;
const bucketId = this.importFromRegistryForm.get('bucket')?.value;
const branch = this.importFromRegistryForm.get('branch')?.value;
- this.loadVersions(registryId, bucketId, flowId, branch);
+ this.loadVersionsRequest.next({ registryId, branch, bucketId, flowId
});
}
private setLoading(resourceType: string, loading: boolean): void {
@@ -268,138 +442,6 @@ export class ImportFromRegistry extends
CloseOnEscapeDialog implements OnInit {
}
}
- loadBranches(registryId: string): void {
- if (registryId) {
- this.setLoading('branch', true);
- this.branchOptions = [];
-
- this.getBranches(registryId)
- .pipe(
- take(1),
- catchError(() => {
- this.setLoading('branch', false);
- return EMPTY;
- })
- )
- .subscribe((branches: BranchEntity[]) => {
- if (branches.length > 0) {
- branches.forEach((entity: BranchEntity) => {
- this.branchOptions.push({
- text: entity.branch.name,
- value: entity.branch.name
- });
- });
-
- const branchId = this.branchOptions[0].value;
- if (branchId) {
-
this.importFromRegistryForm.get('branch')?.setValue(branchId);
- this.loadBuckets(registryId, branchId);
- }
- }
- this.setLoading('branch', false);
- });
- }
- }
-
- loadBuckets(registryId: string, branch?: string | null): void {
- this.setLoading('bucket', true);
- this.bucketOptions = [];
-
- this.getBuckets(registryId, branch)
- .pipe(
- take(1),
- catchError(() => {
- this.setLoading('bucket', false);
- return EMPTY;
- })
- )
- .subscribe((buckets: BucketEntity[]) => {
- if (buckets.length > 0) {
- buckets.forEach((entity: BucketEntity) => {
- if (entity.permissions.canRead) {
- this.bucketOptions.push({
- text: entity.bucket.name,
- value: entity.id,
- description: entity.bucket.description
- });
- }
- });
-
- const bucketId = this.bucketOptions[0].value;
- if (bucketId) {
-
this.importFromRegistryForm.get('bucket')?.setValue(bucketId);
- this.loadFlows(registryId, bucketId, branch);
- }
- }
- this.setLoading('bucket', false);
- });
- }
-
- loadFlows(registryId: string, bucketId: string, branch?: string | null):
void {
- this.setLoading('flow', true);
- this.flowOptions = [];
- this.flowLookup.clear();
-
- this.getFlows(registryId, bucketId, branch)
- .pipe(
- take(1),
- catchError(() => {
- this.setLoading('flow', false);
- return EMPTY;
- })
- )
- .subscribe((versionedFlows: VersionedFlowEntity[]) => {
- if (versionedFlows.length > 0) {
- versionedFlows.forEach((entity: VersionedFlowEntity) => {
- this.flowLookup.set(entity.versionedFlow.flowId!,
entity.versionedFlow);
-
- this.flowOptions.push({
- text: entity.versionedFlow.flowName,
- value: entity.versionedFlow.flowId!,
- description: entity.versionedFlow.description
- });
- });
-
- const flowId = this.flowOptions[0].value;
- if (flowId) {
-
this.importFromRegistryForm.get('flow')?.setValue(flowId);
- this.loadVersions(registryId, bucketId, flowId,
branch);
- }
- }
- this.setLoading('flow', false);
- });
- }
-
- loadVersions(registryId: string, bucketId: string, flowId: string,
branch?: string | null): void {
- this.setLoading('version', true);
- this.dataSource.data = [];
- this.selectedFlowVersion = null;
- this.selectedFlowDescription =
this.flowLookup.get(flowId)?.description;
-
- this.getFlowVersions(registryId, bucketId, flowId, branch)
- .pipe(
- take(1),
- catchError((errorResponse: HttpErrorResponse) => {
- this.setLoading('version', false);
-
this.loadingVersionsError.set(this.errorHelper.getErrorString(errorResponse));
- return EMPTY;
- })
- )
- .subscribe((metadataEntities:
VersionedFlowSnapshotMetadataEntity[]) => {
- if (metadataEntities.length > 0) {
- const flowVersions = metadataEntities.map(
- (entity: VersionedFlowSnapshotMetadataEntity) =>
entity.versionedFlowSnapshotMetadata
- );
-
- const sortedFlowVersions = this.sortVersions(flowVersions,
this.sort);
- this.selectedFlowVersion = sortedFlowVersions[0].version;
-
- this.dataSource.data = sortedFlowVersions;
- }
- this.setLoading('version', false);
- });
- }
-
formatTimestamp(flowVersion: VersionedFlowSnapshotMetadata) {
// get the current user time to properly convert the server time
const now: Date = new Date();
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/save-version-dialog/save-version-dialog.component.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/save-version-dialog/save-version-dialog.component.spec.ts
index 772b2b19ea9..f124c4ca658 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/save-version-dialog/save-version-dialog.component.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/save-version-dialog/save-version-dialog.component.spec.ts
@@ -28,7 +28,8 @@ import { initialState as initialErrorState } from
'../../../../../../../state/er
import { errorFeatureKey } from '../../../../../../../state/error';
import { initialState as initialCurrentUserState } from
'../../../../../../../state/current-user/current-user.reducer';
import { currentUserFeatureKey } from
'../../../../../../../state/current-user';
-import { EMPTY } from 'rxjs';
+import { EMPTY, Observable, Subject } from 'rxjs';
+import { BranchEntity, BucketEntity, RegistryClientEntity } from
'../../../../../../../state/shared';
import { Signal } from '@angular/core';
import { NoopAnimationsModule } from '@angular/platform-browser/animations';
@@ -150,4 +151,163 @@ describe('SaveVersionDialog', () => {
it('should create', () => {
expect(component).toBeTruthy();
});
+
+ describe('with multiple registry clients', () => {
+ const registryAId = 'registry-a';
+ const registryBId = 'registry-b';
+
+ let bucketRequests: Map<string, Subject<BucketEntity[]>>;
+ let branchRequests: Map<string, Subject<BranchEntity[]>>;
+ let cancelledBucketRequests: string[];
+
+ const createRegistryClient = (id: string, name: string,
supportsBranching = false): RegistryClientEntity => {
+ const template = data.registryClients![0];
+ return {
+ ...template,
+ id,
+ component: {
+ ...template.component,
+ id,
+ name,
+ supportsBranching
+ }
+ };
+ };
+
+ const createBucket = (id: string): BucketEntity => ({
+ id,
+ permissions: { canRead: true, canWrite: true },
+ bucket: { id, created: 0, name: `bucket in ${id}`, description: ''
}
+ });
+
+ const createBranch = (registryId: string): BranchEntity =>
+ ({
+ branch: { name: `branch in ${registryId}` }
+ }) as BranchEntity;
+
+ /**
+ * Registers a request per registry id so that the test controls when
each one responds, and
+ * records the registry ids whose requests were torn down before
completing.
+ */
+ const controllableRequest = <T>(requests: Map<string, Subject<T>>,
cancelled?: string[]) => {
+ return (registryId: string) =>
+ new Observable<T>((subscriber) => {
+ const request = new Subject<T>();
+ requests.set(registryId, request);
+ const subscription = request.subscribe(subscriber);
+ return () => {
+ if (!request.closed) {
+ cancelled?.push(registryId);
+ }
+ subscription.unsubscribe();
+ };
+ });
+ };
+
+ const createDialog = async (supportsBranching: boolean) => {
+ TestBed.resetTestingModule();
+ await TestBed.configureTestingModule({
+ imports: [SaveVersionDialog, MatDialogModule,
NoopAnimationsModule],
+ providers: [
+ {
+ provide: MAT_DIALOG_DATA,
+ useValue: {
+ ...data,
+ registryClients: [
+ createRegistryClient(registryAId, 'A
Registry', supportsBranching),
+ createRegistryClient(registryBId, 'B
Registry', supportsBranching)
+ ]
+ }
+ },
+ provideMockStore({
+ initialState: {
+ [errorFeatureKey]: initialErrorState,
+ [currentUserFeatureKey]: initialCurrentUserState,
+ [canvasFeatureKey]: {
+ [flowFeatureKey]: initialFlowState
+ }
+ }
+ }),
+ { provide: MatDialogRef, useValue: null }
+ ]
+ }).compileComponents();
+
+ fixture = TestBed.createComponent(SaveVersionDialog);
+ component = fixture.componentInstance;
+ component.getBranches = controllableRequest(branchRequests);
+ component.getBuckets = controllableRequest(bucketRequests,
cancelledBucketRequests);
+ component.saving = (() => false) as Signal<boolean>;
+
+ fixture.detectChanges();
+ };
+
+ beforeEach(() => {
+ bucketRequests = new Map<string, Subject<BucketEntity[]>>();
+ branchRequests = new Map<string, Subject<BranchEntity[]>>();
+ cancelledBucketRequests = [];
+ });
+
+ describe('that do not support branching', () => {
+ beforeEach(async () => {
+ await createDialog(false);
+ });
+
+ it('should ignore a bucket response for a registry that is no
longer selected', () => {
+ component.registryChanged(registryBId);
+
+
bucketRequests.get(registryAId)?.next([createBucket(registryAId)]);
+
+ expect(component.bucketOptions).toEqual([]);
+
expect(component.saveVersionForm.get('bucket')?.value).toBeNull();
+
+
bucketRequests.get(registryBId)?.next([createBucket(registryBId)]);
+
+ expect(component.bucketOptions.map((option) =>
option.value)).toEqual([registryBId]);
+
expect(component.saveVersionForm.get('bucket')?.value).toEqual(registryBId);
+ });
+
+ it('should cancel an outstanding bucket request when the selected
registry changes', () => {
+ expect(cancelledBucketRequests).toEqual([]);
+
+ component.registryChanged(registryBId);
+
+ expect(cancelledBucketRequests).toEqual([registryAId]);
+ });
+
+ it('should still load buckets for a later registry selection after
a failed bucket request', () => {
+ bucketRequests.get(registryAId)?.error(new Error('bucket
request failed'));
+
+ component.registryChanged(registryBId);
+
bucketRequests.get(registryBId)?.next([createBucket(registryBId)]);
+
+ expect(component.bucketOptions.map((option) =>
option.value)).toEqual([registryBId]);
+
expect(component.saveVersionForm.get('bucket')?.value).toEqual(registryBId);
+ });
+ });
+
+ describe('that support branching', () => {
+ beforeEach(async () => {
+ await createDialog(true);
+ });
+
+ it('should ignore a branch response for a registry that is no
longer selected', () => {
+ component.registryChanged(registryBId);
+
+
branchRequests.get(registryAId)?.next([createBranch(registryAId)]);
+
+ expect(component.branchOptions).toEqual([]);
+
expect(component.saveVersionForm.get('branch')?.value).toBeNull();
+ });
+
+ it('should still load branches for a later registry selection
after a failed branch request', () => {
+ branchRequests.get(registryAId)?.error(new Error('branch
request failed'));
+
+ component.registryChanged(registryBId);
+
branchRequests.get(registryBId)?.next([createBranch(registryBId)]);
+
+ expect(component.branchOptions.map((option) =>
option.value)).toEqual([`branch in ${registryBId}`]);
+
expect(component.saveVersionForm.get('branch')?.value).toEqual(`branch in
${registryBId}`);
+ });
+ });
+ });
});
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/save-version-dialog/save-version-dialog.component.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/save-version-dialog/save-version-dialog.component.ts
index 4877ccebe6d..46056ec38d7 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/save-version-dialog/save-version-dialog.component.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/ui/canvas/items/flow/save-version-dialog/save-version-dialog.component.ts
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import { Component, EventEmitter, Input, OnInit, Output, Signal, inject } from
'@angular/core';
+import { Component, DestroyRef, EventEmitter, inject, Input, OnInit, Output,
Signal } from '@angular/core';
import {
MAT_DIALOG_DATA,
MatDialogActions,
@@ -27,22 +27,28 @@ import { FormBuilder, FormControl, FormGroup,
ReactiveFormsModule, Validators }
import { MatButton } from '@angular/material/button';
import { MatError, MatFormField, MatLabel } from
'@angular/material/form-field';
import { MatOption, MatSelect } from '@angular/material/select';
-import { Observable, of, take } from 'rxjs';
+import { catchError, EMPTY, map, Observable, of, Subject, switchMap, take }
from 'rxjs';
import { BranchEntity, BucketEntity, RegistryClientEntity } from
'../../../../../../../state/shared';
import { SaveVersionDialogRequest, SaveVersionRequest } from
'../../../../../state/flow';
import { VersionControlInformation } from
'../../../../../../../ui/common/tooltips/version-control-tip/version-control-tip.component';
import {
- TextTip,
+ CloseOnEscapeDialog,
NiFiCommon,
NifiSpinnerDirective,
NifiTooltipDirective,
- CloseOnEscapeDialog,
- SelectOption
+ SelectOption,
+ TextTip
} from '@nifi/shared';
import { MatInput } from '@angular/material/input';
import { ErrorContextKey } from '../../../../../../../state/error';
import { ContextErrorBanner } from
'../../../../../../../ui/common/context-error-banner/context-error-banner.component';
+import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
+
+interface LoadBucketsRequest {
+ registryId: string;
+ branch: string | null;
+}
@Component({
selector: 'save-version-dialog',
@@ -70,6 +76,7 @@ export class SaveVersionDialog extends CloseOnEscapeDialog
implements OnInit {
private dialogRequest = inject<SaveVersionDialogRequest>(MAT_DIALOG_DATA);
private formBuilder = inject(FormBuilder);
private nifiCommon = inject(NiFiCommon);
+ private destroyRef = inject(DestroyRef);
@Input() getBranches: (registryId: string) => Observable<BranchEntity[]> =
() => of([]);
@Input() getBuckets: (registryId: string, branch?: string | null) =>
Observable<BucketEntity[]> = () => of([]);
@@ -87,6 +94,9 @@ export class SaveVersionDialog extends CloseOnEscapeDialog
implements OnInit {
private clientBranchingSupportMap: Map<string, boolean> = new Map<string,
boolean>();
+ private loadBranchesRequest = new Subject<string | null>();
+ private loadBucketsRequest = new Subject<LoadBucketsRequest | null>();
+
constructor() {
super();
const dialogRequest = this.dialogRequest;
@@ -123,6 +133,8 @@ export class SaveVersionDialog extends CloseOnEscapeDialog
implements OnInit {
flowDescription: new FormControl(null),
comments: new FormControl(null)
});
+
+ this.wireRegistryResourceRequests();
} else {
this.saveVersionForm = formBuilder.group({
comments: new FormControl('')
@@ -135,89 +147,107 @@ export class SaveVersionDialog extends
CloseOnEscapeDialog implements OnInit {
const selectedRegistryId: string | null =
this.saveVersionForm.get('registry')?.value;
if (selectedRegistryId) {
- this.supportsBranching =
this.clientBranchingSupportMap.get(selectedRegistryId) || false;
- if (this.supportsBranching) {
- this.loadBranches(selectedRegistryId);
- } else {
- this.loadBuckets(selectedRegistryId);
- }
+ this.registrySelected(selectedRegistryId);
}
}
}
- private clearBranches(): void {
- this.branchOptions = [];
- this.saveVersionForm.get('branch')?.setValue(null);
- this.clearBuckets();
- }
+ private wireRegistryResourceRequests(): void {
+ this.loadBranchesRequest
+ .pipe(
+ switchMap((registryId: string | null) => {
+ if (!registryId) {
+ return EMPTY;
+ }
+ this.branchOptions = [];
+ return this.getBranches(registryId).pipe(
+ take(1),
+ map((branches: BranchEntity[]) => ({ registryId,
branches })),
+ catchError(() => EMPTY)
+ );
+ }),
+ takeUntilDestroyed(this.destroyRef)
+ )
+ .subscribe(({ registryId, branches }) => {
+ branches.forEach((entity: BranchEntity) => {
+ this.branchOptions.push({
+ text: entity.branch.name,
+ value: entity.branch.name
+ });
+ });
- loadBranches(registryId: string): void {
- this.clearBranches();
+ const branch = this.branchOptions[0]?.value;
+ if (branch) {
+ this.saveVersionForm.get('branch')?.setValue(branch);
+ this.loadBucketsRequest.next({ registryId, branch });
+ }
+ });
- this.getBranches(registryId)
- .pipe(take(1))
- .subscribe((branches: BranchEntity[]) => {
- if (branches.length > 0) {
- branches.forEach((entity: BranchEntity) => {
- this.branchOptions.push({
- text: entity.branch.name,
- value: entity.branch.name
+ this.loadBucketsRequest
+ .pipe(
+ switchMap((request: LoadBucketsRequest | null) => {
+ if (!request) {
+ return EMPTY;
+ }
+ this.bucketOptions = [];
+ return this.getBuckets(request.registryId,
request.branch).pipe(
+ take(1),
+ catchError(() => EMPTY)
+ );
+ }),
+ takeUntilDestroyed(this.destroyRef)
+ )
+ .subscribe((buckets: BucketEntity[]) => {
+ buckets.forEach((entity: BucketEntity) => {
+ // only allow buckets to be selectable if the user can
read and write to them
+ if (entity.permissions.canRead &&
entity.permissions.canWrite) {
+ this.bucketOptions.push({
+ text: entity.bucket.name,
+ value: entity.id,
+ description: entity.bucket.description
});
- });
-
- const branchId = this.branchOptions[0].value;
- if (branchId) {
- this.saveVersionForm.get('branch')?.setValue(branchId);
- this.loadBuckets(registryId, branchId);
}
+ });
+
+ const bucketId = this.bucketOptions[0]?.value;
+ if (bucketId) {
+ this.saveVersionForm.get('bucket')?.setValue(bucketId);
}
});
}
+ private clearBranches(): void {
+ this.branchOptions = [];
+ this.saveVersionForm.get('branch')?.setValue(null);
+ this.loadBranchesRequest.next(null);
+ this.clearBuckets();
+ }
+
private clearBuckets(): void {
this.bucketOptions = [];
this.saveVersionForm.get('bucket')?.setValue(null);
+ this.loadBucketsRequest.next(null);
}
- loadBuckets(registryId: string, branch?: string | null): void {
- this.clearBuckets();
-
- this.getBuckets(registryId, branch)
- .pipe(take(1))
- .subscribe((buckets: BucketEntity[]) => {
- if (buckets.length > 0) {
- buckets.forEach((entity: BucketEntity) => {
- // only allow buckets to be selectable if the user can
read and write to them
- if (entity.permissions.canRead &&
entity.permissions.canWrite) {
- this.bucketOptions.push({
- text: entity.bucket.name,
- value: entity.id,
- description: entity.bucket.description
- });
- }
- });
-
- const bucketId = this.bucketOptions[0].value;
- if (bucketId) {
- this.saveVersionForm.get('bucket')?.setValue(bucketId);
- }
- }
- });
+ registryChanged(registryId: string): void {
+ this.registrySelected(registryId);
}
- registryChanged(registryId: string): void {
+ private registrySelected(registryId: string): void {
this.supportsBranching =
this.clientBranchingSupportMap.get(registryId) || false;
+ this.clearBranches();
if (this.supportsBranching) {
- this.loadBranches(registryId);
+ this.loadBranchesRequest.next(registryId);
} else {
- this.loadBuckets(registryId);
+ this.loadBucketsRequest.next({ registryId, branch: null });
}
}
branchChanged(branch: string): void {
const selectedRegistryId: string | null =
this.saveVersionForm.get('registry')?.value;
if (selectedRegistryId) {
- this.loadBuckets(selectedRegistryId, branch);
+ this.clearBuckets();
+ this.loadBucketsRequest.next({ registryId: selectedRegistryId,
branch });
}
}
