This is an automated email from the ASF dual-hosted git repository. JHoelli pushed a commit to branch 1357-harmonize-resource-creation-workflow-for-pipelines in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit dd5dfc5f84b8d6632994c5c0995564879b151e35 Author: Jacqueline Höllig <[email protected]> AuthorDate: Fri Jul 10 08:56:49 2026 +0000 deleted unnecssary component --- .../pipeline-assembly.component.ts | 59 +++ .../save-pipeline-settings.component.html | 177 -------- .../save-pipeline-settings.component.scss | 21 - .../save-pipeline-settings.component.ts | 181 -------- .../save-pipeline/save-pipeline.component.html | 117 ------ .../save-pipeline/save-pipeline.component.scss | 35 -- .../save-pipeline/save-pipeline.component.ts | 465 --------------------- 7 files changed, 59 insertions(+), 996 deletions(-) diff --git a/ui/src/app/editor/components/pipeline-assembly/pipeline-assembly.component.ts b/ui/src/app/editor/components/pipeline-assembly/pipeline-assembly.component.ts index d8b2e3e8b9..db0d6fb7b2 100644 --- a/ui/src/app/editor/components/pipeline-assembly/pipeline-assembly.component.ts +++ b/ui/src/app/editor/components/pipeline-assembly/pipeline-assembly.component.ts @@ -48,6 +48,7 @@ import { EditorService } from '../../services/editor.service'; import { LinkageData, Message, + MeasurementUpdateInfo, Pipeline, PipelineCanvasMetadata, PipelineCanvasMetadataService, @@ -64,6 +65,7 @@ import { JsplumbService } from '../../services/jsplumb.service'; import { TranslateService } from '@ngx-translate/core'; import { FlexDirective } from '@ngbracket/ngx-layout/flex'; import { PipelineOperationsService } from '../../../pipelines/services/pipeline-operations.service'; +import { SavePipelineUpdateMigrationComponent } from '../../dialog/save-pipeline/save-pipeline-update-migration/save-pipeline-update-migration.component'; @Component({ selector: 'sp-pipeline-assembly', @@ -285,6 +287,10 @@ export class PipelineAssemblyComponent implements AfterViewInit, OnDestroy { pipeline.createdAt = this.originalPipeline.createdAt; pipeline.createdByUser = this.originalPipeline.createdByUser; + if (!(await this.confirmPipelineUpdateIfRequired(pipeline))) { + return; + } + await this.savePipelineResource(pipeline, startPipelineAfterStorage); await this.savePendingManagePipelineChanges(); this.editorService.makePipelineAssemblyEmpty(true); @@ -410,6 +416,59 @@ export class PipelineAssemblyComponent implements AfterViewInit, OnDestroy { ]; } + private async confirmPipelineUpdateIfRequired( + pipeline: Pipeline, + ): Promise<boolean> { + if (!this.originalPipeline || !this.hasDataLakeSink(pipeline)) { + return true; + } + + const measurementUpdateInfos = await firstValueFrom( + this.pipelineService.performPipelineMigrationPreflight(pipeline), + ); + + if (measurementUpdateInfos.length === 0) { + return true; + } + + return this.openPipelineUpdateMigrationDialog(measurementUpdateInfos); + } + + private hasDataLakeSink(pipeline: Pipeline): boolean { + return pipeline.actions.some( + action => + action.appId === + 'org.apache.streampipes.sinks.internal.jvm.datalake', + ); + } + + private async openPipelineUpdateMigrationDialog( + measurementUpdateInfos: MeasurementUpdateInfo[], + ): Promise<boolean> { + const dialogRef = this.dialogService.open( + SavePipelineUpdateMigrationComponent, + { + panelType: PanelType.SLIDE_IN_PANEL, + title: this.translateService.instant('Pipeline update review'), + width: '50vw', + data: { + measurementUpdateInfos, + }, + }, + ); + + const startUpdateSubscription = + dialogRef.componentInstance.instance.startUpdateEmitter.subscribe( + () => { + dialogRef.close(true); + }, + ); + + const shouldContinue = await firstValueFrom(dialogRef.afterClosed()); + startUpdateSubscription.unsubscribe(); + return !!shouldContinue; + } + togglePreview(): void { this.previewModeActive = !this.previewModeActive; this.drawingAreaComponent.togglePipelineElementLivePreview(); diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.html b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.html deleted file mode 100644 index 2fa9e2902a..0000000000 --- a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.html +++ /dev/null @@ -1,177 +0,0 @@ -<!-- -~ 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. -~ ---> - -<div fxLayout="column"> - <form [formGroup]="submitPipelineForm"> - <div fxFlex="100" fxLayout="column"> - <sp-form-field [level]="3" [label]="'Pipeline Name' | translate"> - <mat-form-field fxFlex color="accent"> - <input - [formControlName]="'pipelineName'" - data-cy="sp-editor-pipeline-name" - matInput - name="pipelineName" - (blur)="triggerTutorial()" - /> - @if ( - submitPipelineForm - .get('pipelineName') - .hasError('required') || - submitPipelineForm - .get('pipelineName') - .hasError('whiteSpaceOnly') - ) { - <mat-error> - {{ 'Pipeline name is required.' | translate }} - </mat-error> - } - @if ( - submitPipelineForm - .get('pipelineName') - .hasError('leadingOrTrailingWhitespace') - ) { - <mat-error> - {{ - 'Pipeline name cannot start or end with a space.' - | translate - }} - </mat-error> - } - @if ( - submitPipelineForm - .get('pipelineName') - .hasError('minlength') || - submitPipelineForm - .get('pipelineName') - .hasError('maxlength') - ) { - <mat-error> - {{ - 'Pipeline name must have between 3 and 50 characters.' - | translate - }} - </mat-error> - } - @if ( - submitPipelineForm - .get('pipelineName') - .hasError('invalidName') - ) { - <mat-error> - {{ - 'Pipeline name can only contain letters, numbers, dashes (-), and underscores (_).' - | translate - }} - </mat-error> - } - </mat-form-field> - </sp-form-field> - <sp-form-field [level]="3" [label]="'Description' | translate"> - <mat-form-field fxFlex color="accent"> - <input [formControlName]="'pipelineDescription'" matInput /> - <mat-error - >{{ - 'Pipeline description must not have more than 80 characters.' - | translate - }} - </mat-error> - </mat-form-field> - </sp-form-field> - </div> - </form> - @if (storageOptions.updateModeActive) { - <div id="overwriteCheckbox" class="checkbox"> - <mat-radio-group - [(ngModel)]="storageOptions.updateMode" - fxLayout="column" - color="accent" - class="pipeline-radio-group" - > - <mat-radio-button - [value]="'update'" - style="padding-left: 0" - data-cy="pipeline-update-mode-update" - > - {{ 'Update pipeline' | translate }} - </mat-radio-button> - <mat-radio-button - [value]="'clone'" - class="mb-10" - data-cy="pipeline-update-mode-clone" - > - {{ 'Create new pipeline' | translate }} - </mat-radio-button> - </mat-radio-group> - </div> - } - <sp-split-section [level]="3" [title]="'Options' | translate"> - <mat-checkbox - [(ngModel)]="storageOptions.startPipelineAfterStorage" - color="accent" - data-cy="sp-editor-checkbox-start-immediately" - > - {{ 'Start pipeline immediately' | translate }} - </mat-checkbox> - <mat-checkbox - [(ngModel)]="storageOptions.navigateToPipelineOverview" - color="accent" - data-cy="sp-editor-checkbox-navigate-to-overview" - > - {{ 'Navigate to pipeline overview afterwards' | translate }} - </mat-checkbox> - @if (isAssetAdmin) { - <mat-checkbox - [(ngModel)]="addToAssets" - color="accent" - data-cy="sp-show-pipeline-asset-checkbox" - > - {{ 'Add Pipeline to Assets' | translate }} - </mat-checkbox> - @if (addToAssets) { - <div class="mt-10"> - <sp-asset-link-configuration - [isEdit]="storageOptions.updateMode === 'update'" - [itemId]="pipeline._id" - (selectedAssetsChange)="onSelectedAssetsChange($event)" - (deselectedAssetsChange)=" - onDeselectedAssetsChange($event) - " - (originalAssetsEmitter)=" - onOriginalAssetsEmitted($event) - " - > - </sp-asset-link-configuration> - </div> - } - } - </sp-split-section> - <div class="mt-10"> - <mat-expansion-panel class="mat-elevation-z0 border-1"> - <mat-expansion-panel-header>{{ - 'Show pipeline configuration as code' | translate - }}</mat-expansion-panel-header> - @if (compactPipeline) { - <sp-configuration-code-panel - [configuration]="compactPipeline" - maxHeight="none" - > - </sp-configuration-code-panel> - } - </mat-expansion-panel> - </div> -</div> diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.scss b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.scss deleted file mode 100644 index e22ed79985..0000000000 --- a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.scss +++ /dev/null @@ -1,21 +0,0 @@ -/* - * 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. - * - */ - -.border-1 { - border: 1px solid var(--color-bg-2); -} diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.ts b/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.ts deleted file mode 100644 index 2b82cb626c..0000000000 --- a/ui/src/app/editor/dialog/save-pipeline/save-pipeline-settings/save-pipeline-settings.component.ts +++ /dev/null @@ -1,181 +0,0 @@ -/* - * 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 { - Component, - EventEmitter, - inject, - Input, - OnInit, - Output, -} from '@angular/core'; -import { ShepherdService } from '../../../../services/tour/shepherd.service'; -import { - FormsModule, - ReactiveFormsModule, - UntypedFormControl, - UntypedFormGroup, - Validators, -} from '@angular/forms'; -import { - CompactPipeline, - Pipeline, - PipelineService, - SpAssetTreeNode, - UserInfo, -} from '@streampipes/platform-services'; -import { PipelineStorageOptions } from '../../../model/editor.model'; -import { ValidateName } from '../../../../core-ui/static-properties/input.validator'; -import { - AssetLinkConfigurationComponent, - CurrentUserService, - FormFieldComponent, - SplitSectionComponent, -} from '@streampipes/shared-ui'; -import { UserRole } from '../../../../core/auth/user-role.enum'; -import { FlexDirective, LayoutDirective } from '@ngbracket/ngx-layout/flex'; -import { MatError, MatFormField } from '@angular/material/form-field'; -import { MatInput } from '@angular/material/input'; -import { MatRadioButton, MatRadioGroup } from '@angular/material/radio'; -import { MatCheckbox } from '@angular/material/checkbox'; -import { - MatExpansionPanel, - MatExpansionPanelHeader, -} from '@angular/material/expansion'; -import { ConfigurationCodePanelComponent } from '../../../../core-ui/configuration-code-panel/configuration-code-panel.component'; -import { TranslatePipe } from '@ngx-translate/core'; - -@Component({ - selector: 'sp-save-pipeline-settings', - templateUrl: './save-pipeline-settings.component.html', - styleUrls: ['./save-pipeline-settings.component.scss'], - imports: [ - LayoutDirective, - FormsModule, - ReactiveFormsModule, - FlexDirective, - FormFieldComponent, - MatFormField, - MatInput, - MatError, - MatRadioGroup, - MatRadioButton, - SplitSectionComponent, - MatCheckbox, - AssetLinkConfigurationComponent, - MatExpansionPanel, - MatExpansionPanelHeader, - ConfigurationCodePanelComponent, - TranslatePipe, - ], -}) -export class SavePipelineSettingsComponent implements OnInit { - private readonly currentUserService = inject(CurrentUserService); - - @Input() - submitPipelineForm: UntypedFormGroup = new UntypedFormGroup({}); - - @Input() - pipeline: Pipeline; - - @Input() - storageOptions: PipelineStorageOptions; - - @Input() - currentPipelineName: string; - - private shepherdService = inject(ShepherdService); - private pipelineService = inject(PipelineService); - - compactPipeline: CompactPipeline; - currentUser: UserInfo; - isAssetAdmin = false; - - addToAssets: boolean = false; - @Input() - selectedAssets: SpAssetTreeNode[]; - @Input() - deselectedAssets: SpAssetTreeNode[]; - @Input() - originalAssets: SpAssetTreeNode[]; - - @Output() selectedAssetsChange = new EventEmitter<SpAssetTreeNode[]>(); - @Output() deselectedAssetsChange = new EventEmitter<SpAssetTreeNode[]>(); - @Output() originalAssetsChange = new EventEmitter<SpAssetTreeNode[]>(); - - ngOnInit() { - this.currentUser = this.currentUserService.getCurrentUser(); - this.isAssetAdmin = this.currentUserService.hasRole( - UserRole.ROLE_ASSET_ADMIN, - ); - this.submitPipelineForm.addControl( - 'pipelineName', - new UntypedFormControl(this.pipeline.name, [ - Validators.required, - Validators.minLength(3), - Validators.maxLength(50), - ValidateName(), - ]), - ); - this.submitPipelineForm.addControl( - 'pipelineDescription', - new UntypedFormControl(this.pipeline.description, [ - Validators.maxLength(80), - ]), - ); - - this.submitPipelineForm.controls['pipelineName'].valueChanges.subscribe( - value => { - this.pipeline.name = value; - }, - ); - - this.submitPipelineForm.controls[ - 'pipelineDescription' - ].valueChanges.subscribe(value => { - this.pipeline.description = value; - }); - this.pipelineService - .convertToCompactPipeline(this.pipeline) - .subscribe(p => (this.compactPipeline = p)); - if (this.storageOptions.updateModeActive) { - this.addToAssets = true; - } - } - - onSelectedAssetsChange(updatedAssets: SpAssetTreeNode[]): void { - this.selectedAssets = updatedAssets; - this.selectedAssetsChange.emit(this.selectedAssets); - } - - onDeselectedAssetsChange(updatedAssets: SpAssetTreeNode[]): void { - this.deselectedAssets = updatedAssets; - this.deselectedAssetsChange.emit(this.deselectedAssets); - } - - onOriginalAssetsEmitted(updatedAssets: SpAssetTreeNode[]): void { - this.originalAssets = updatedAssets; - this.originalAssetsChange.emit(this.originalAssets); - } - - triggerTutorial() { - if (this.shepherdService.isTourActive()) { - this.shepherdService.trigger('save-pipeline-dialog'); - } - } -} diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.html b/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.html deleted file mode 100644 index 51004cc737..0000000000 --- a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.html +++ /dev/null @@ -1,117 +0,0 @@ -<!-- -~ 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. -~ ---> - -<div class="sp-dialog-container"> - <div class="sp-dialog-content padding-20"> - <div fxFlex="100" fxLayout="column"> - @if ( - !operationCompleted && - !operationProgress && - !pipelineUpdatePreflight - ) { - <sp-save-pipeline-settings - [currentPipelineName]="pipeline.name" - [submitPipelineForm]="submitPipelineForm" - [pipeline]="pipeline" - [storageOptions]="storageOptions" - [(selectedAssets)]="selectedAssets" - [(deselectedAssets)]="deselectedAssets" - [(originalAssets)]="originalAssets" - > - </sp-save-pipeline-settings> - } - - @if ( - pipelineUpdatePreflight && - !operationCompleted && - !operationProgress - ) { - <sp-save-pipeline-update-migration - [measurementUpdateInfos]="measurementUpdateInfos" - (startUpdateEmitter)="savePipeline(true)" - > - </sp-save-pipeline-update-migration> - } - - @if (operationProgress || operationCompleted) { - <sp-multi-step-status-indicator - [statusIndicators]="statusIndicators" - > - </sp-multi-step-status-indicator> - } - - @if (finalPipelineOperationStatus) { - <div class="mt-10"> - <mat-divider></mat-divider> - <sp-pipeline-started-status - class="mt-10" - [forceStopDisabled]="true" - [action]="pipelineAction" - [pipelineOperationStatus]="finalPipelineOperationStatus" - > - </sp-pipeline-started-status> - </div> - } - </div> - </div> - <mat-divider></mat-divider> - <div class="sp-dialog-actions" fxLayoutGap="10px"> - @if (operationCompleted) { - <button mat-flat-button (click)="hide(false)"> - Create another pipeline - </button> - } - @if (operationCompleted) { - <button - mat-flat-button - class="mat-basic" - data-cy="sp-navigate-to-pipeline-overview" - (click)="navigateToPipelineOverview()" - > - Open pipeline overview - </button> - } - @if ( - !operationCompleted && !operationSuccess && !pipelineUpdatePreflight - ) { - <button - [disabled]=" - !submitPipelineForm.valid || - operationProgress || - operationCompleted - " - mat-flat-button - color="accent" - (click)="savePipeline()" - data-cy="sp-editor-apply" - > - {{ 'Apply' | translate }} - </button> - } - @if (!operationProgress && (!operationCompleted || !operationSuccess)) { - <button - mat-flat-button - class="mat-basic" - (click)="hide(true)" - data-cy="sp-editor-cancel" - > - {{ 'Cancel' | translate }} - </button> - } - </div> -</div> diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.scss b/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.scss deleted file mode 100644 index cdfe259c98..0000000000 --- a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.scss +++ /dev/null @@ -1,35 +0,0 @@ -/* - * 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. - * - */ - -.customize-section { - display: flex; - flex: 1 1 auto; - padding: 20px; -} - -.padding-20 { - padding: 20px; -} - -.mb-10 { - margin-bottom: 10px; -} - -::ng-deep .pipeline-radio-group .mat-radio-label { - padding: 0; -} diff --git a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.ts b/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.ts deleted file mode 100644 index 1cb1fa125c..0000000000 --- a/ui/src/app/editor/dialog/save-pipeline/save-pipeline.component.ts +++ /dev/null @@ -1,465 +0,0 @@ -/* - * 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 { Component, inject, Input, OnInit } from '@angular/core'; -import { AssetSaveService, DialogRef } from '@streampipes/shared-ui'; -import { - DatalakeRestService, - DataSinkInvocation, - LinkageData, - Message, - Pipeline, - PipelineCanvasMetadata, - PipelineCanvasMetadataService, - PipelineOperationStatus, - MeasurementUpdateInfo, - PipelineService, - SpAssetTreeNode, -} from '@streampipes/platform-services'; -import { EditorService } from '../../services/editor.service'; -import { ShepherdService } from '../../../services/tour/shepherd.service'; -import { UntypedFormGroup } from '@angular/forms'; -import { Router } from '@angular/router'; -import { - InvocablePipelineElementUnion, - PipelineStorageOptions, -} from '../../model/editor.model'; -import { IdGeneratorService } from '../../../core-services/id-generator/id-generator.service'; -import { firstValueFrom, lastValueFrom, Observable, of, tap } from 'rxjs'; -import { filter, switchMap } from 'rxjs/operators'; -import { - Status, - StatusIndicator, -} from '../../../core-ui/multi-step-status-indicator/multi-step-status-indicator.model'; -import { PipelineAction } from '../../../pipelines/model/pipeline-model'; -import { TranslatePipe, TranslateService } from '@ngx-translate/core'; -import { - FlexDirective, - LayoutDirective, - LayoutGapDirective, -} from '@ngbracket/ngx-layout/flex'; -import { SavePipelineSettingsComponent } from './save-pipeline-settings/save-pipeline-settings.component'; -import { MultiStepStatusIndicatorComponent } from '../../../core-ui/multi-step-status-indicator/multi-step-status-indicator.component'; -import { MatDivider } from '@angular/material/divider'; -import { PipelineStartedStatusComponent } from '../../../core-ui/pipeline/pipeline-started-status/pipeline-started-status.component'; -import { MatButton } from '@angular/material/button'; -import { SavePipelineUpdateMigrationComponent } from './save-pipeline-update-migration/save-pipeline-update-migration.component'; - -@Component({ - selector: 'sp-save-pipeline', - templateUrl: './save-pipeline.component.html', - styleUrls: ['./save-pipeline.component.scss'], - imports: [ - FlexDirective, - LayoutDirective, - SavePipelineSettingsComponent, - MultiStepStatusIndicatorComponent, - MatDivider, - PipelineStartedStatusComponent, - LayoutGapDirective, - MatButton, - TranslatePipe, - SavePipelineUpdateMigrationComponent, - ], -}) -export class SavePipelineComponent implements OnInit { - private editorService = inject(EditorService); - private dialogRef = inject(DialogRef<SavePipelineComponent>); - private idGeneratorService = inject(IdGeneratorService); - private pipelineService = inject(PipelineService); - private router = inject(Router); - private shepherdService = inject(ShepherdService); - private pipelineCanvasService = inject(PipelineCanvasMetadataService); - private assetSaveService = inject(AssetSaveService); - private dataLakeService = inject(DatalakeRestService); - private translateService = inject(TranslateService); - - @Input() - pipeline: Pipeline; - - @Input() - originalPipeline: Pipeline; - - selectedAssets: SpAssetTreeNode[]; - deselectedAssets: SpAssetTreeNode[]; - originalAssets: SpAssetTreeNode[]; - - @Input() - pipelineCanvasMetadata: PipelineCanvasMetadata; - - operationProgress = false; - operationCompleted = false; - operationSuccess = false; - - errorMessage = ''; - pipelineId: string; - - storageOptions: PipelineStorageOptions = { - updateMode: 'update', - startPipelineAfterStorage: true, - navigateToPipelineOverview: true, - updateModeActive: false, - }; - - submitPipelineForm: UntypedFormGroup = new UntypedFormGroup({}); - statusIndicators: StatusIndicator[] = []; - finalPipelineOperationStatus: PipelineOperationStatus; - pipelineAction: PipelineAction; - pipelineUpdatePreflight = false; - measurementUpdateInfos: MeasurementUpdateInfo[] = []; - - ngOnInit() { - this.storageOptions.updateModeActive = - this.originalPipeline !== undefined; - if (this.storageOptions.updateModeActive) { - this.pipeline._id = this.originalPipeline._id; - this.pipeline.name = this.originalPipeline.name; - this.pipeline.description = this.originalPipeline.description; - this.pipeline.running = this.originalPipeline.running; - this.pipeline.createdAt = this.originalPipeline.createdAt; - this.pipeline.createdByUser = this.originalPipeline.createdByUser; - } - - if (this.shepherdService.isTourActive()) { - this.shepherdService.trigger('enter-pipeline-name'); - } - } - - performStorageOperations( - stopPipeline$: Observable<null | PipelineOperationStatus>, - savePipeline$: Observable<Message>, - ) { - // if pipeline is running and update mode: stop pipeline - // if update mode: update pipeline, if not update mode or update mode clone: save pipeline - // if update mode and not clone: update canvas, else store new canvas - // if should start: start pipeline - stopPipeline$ - .pipe( - tap(() => - this.addStatusIndicator( - this.translateService.instant('Saving pipeline'), - Status.PROGRESS, - ), - ), - switchMap(() => savePipeline$), - tap(message => { - this.operationSuccess = message.success; - if (!message.success) { - this.handleStorageError(); - } - this.modifyStatusIndicator(Status.SUCCESS); - this.pipelineId = message.notifications[1].description; - }), - // only continue if pipeline was saved - filter(message => message.success), - tap(() => - this.addStatusIndicator( - this.translateService.instant('Saving metadata'), - Status.PROGRESS, - ), - ), - switchMap(() => - this.getPipelineCanvasMetadata$(this.pipelineId), - ), - tap(() => this.modifyStatusIndicator(Status.SUCCESS)), - switchMap(() => this.getStartPipeline$()), - ) - .subscribe({ - next: message => { - this.onSuccess(message); - // Add Asset as soon as pipelineId is known - this.addToAsset(); - }, - error: msg => { - this.onFailure(msg); - }, - }); - } - - clonePipeline(): void { - this.pipeline._id = undefined; - this.pipeline._rev = undefined; - this.pipeline.running = false; - this.pipeline.actions.forEach(element => this.updateId(element)); - this.pipeline.sepas.forEach(element => this.updateId(element)); - this.pipelineCanvasMetadata._id = undefined; - this.pipelineCanvasMetadata._rev = undefined; - } - - savePipeline(skipPreflight = false) { - if (this.shouldPerformUpdatePreflight(skipPreflight)) { - this.performUpdatePreflight(); - return; - } - - this.pipelineUpdatePreflight = false; - let stopPipeline$: Observable<null | PipelineOperationStatus> = - of(null); - let savePipeline$: Observable<Message> = - this.pipelineService.storePipeline(this.pipeline); - this.operationProgress = true; - if (this.storageOptions.updateModeActive) { - if (this.storageOptions.updateMode === 'clone') { - this.clonePipeline(); - } else { - if (this.pipeline.running) { - stopPipeline$ = this.getStopPipeline$(); - } - savePipeline$ = this.pipelineService.updatePipeline( - this.pipeline, - ); - } - } - - this.performStorageOperations(stopPipeline$, savePipeline$); - } - - shouldPerformUpdatePreflight(skipPreflight: boolean): boolean { - return ( - !skipPreflight && - this.storageOptions.updateModeActive && - this.storageOptions.updateMode !== 'clone' && - this.hasDataLakeSink() - ); - } - - hasDataLakeSink(): boolean { - return this.pipeline.actions.some( - action => - action.appId === - 'org.apache.streampipes.sinks.internal.jvm.datalake', - ); - } - - performUpdatePreflight(): void { - this.operationProgress = true; - this.addStatusIndicator( - this.translateService.instant('Checking pipeline update'), - Status.PROGRESS, - ); - this.pipelineService - .performPipelineMigrationPreflight(this.pipeline) - .subscribe({ - next: updateInfos => { - if (updateInfos.length === 0) { - this.modifyStatusIndicator(Status.SUCCESS); - this.savePipeline(true); - } else { - this.measurementUpdateInfos = updateInfos; - this.pipelineUpdatePreflight = true; - this.operationProgress = false; - this.statusIndicators = []; - } - }, - error: msg => { - this.onFailure(msg); - }, - }); - } - - updateId(entity: InvocablePipelineElementUnion) { - const lastIdIndex = entity.elementId.lastIndexOf(':'); - entity.elementId = - entity.elementId.substring(0, lastIdIndex + 1) + - this.idGeneratorService.generate(5); - } - - getStopPipeline$(): Observable<PipelineOperationStatus> { - return of(null).pipe( - tap(() => - this.addStatusIndicator( - this.translateService.instant('Stopping pipeline'), - Status.PROGRESS, - ), - ), - switchMap(() => - this.pipelineService.stopPipeline(this.originalPipeline._id), - ), - tap(msg => { - this.operationSuccess = msg.success; - if (!msg.success) { - this.handlePipelineOperationError(msg, PipelineAction.Stop); - } else { - this.modifyStatusIndicator(Status.SUCCESS); - } - }), - filter(status => status.success), - ); - } - - getStartPipeline$(): Observable<null | PipelineOperationStatus> { - if (this.storageOptions.startPipelineAfterStorage) { - return of(null).pipe( - tap(() => - this.addStatusIndicator( - this.translateService.instant('Starting pipeline'), - Status.PROGRESS, - ), - ), - switchMap(() => - this.pipelineService.startPipeline(this.pipelineId), - ), - tap(msg => { - if (!msg.success) { - this.handlePipelineOperationError( - msg, - PipelineAction.Start, - ); - } else { - this.modifyStatusIndicator( - msg.success ? Status.SUCCESS : Status.FAILURE, - ); - } - }), - ); - } else { - return of(null); - } - } - - getPipelineCanvasMetadata$(pipelineId: string): Observable<object> { - this.pipelineCanvasMetadata.pipelineId = pipelineId; - return this.pipelineCanvasService.updatePipelineCanvasMetadata( - pipelineId, - this.pipelineCanvasMetadata, - ); - } - - addStatusIndicator(message: string, status: Status) { - this.statusIndicators.push({ message, status }); - } - - modifyStatusIndicator(status: Status) { - // modify status of the last indicator - this.statusIndicators[this.statusIndicators.length - 1].status = status; - } - - handleStorageError(): void { - this.onFailure(); - } - - handlePipelineOperationError( - status: PipelineOperationStatus, - pipelineAction: PipelineAction, - ) { - this.onFailure(); - this.showPipelineOperationStatus(status, pipelineAction); - } - - onFailure(_msg?: any) { - this.operationCompleted = true; - this.operationSuccess = false; - if (this.statusIndicators.length > 0) { - this.modifyStatusIndicator(Status.FAILURE); - } - } - - showPipelineOperationStatus( - status: PipelineOperationStatus, - pipelineAction: PipelineAction, - ) { - this.finalPipelineOperationStatus = status; - this.pipelineAction = pipelineAction; - } - - onSuccess(status?: PipelineOperationStatus) { - this.operationProgress = false; - this.operationCompleted = true; - if (status) { - this.showPipelineOperationStatus(status, PipelineAction.Start); - } - this.editorService.makePipelineAssemblyEmpty(true); - this.editorService.removePipelineFromCache().subscribe(); - if (this.shepherdService.isTourActive()) { - this.shepherdService.hideCurrentStep(); - } - if (this.storageOptions.navigateToPipelineOverview && status?.success) { - this.navigateToPipelineOverview(); - } - } - - navigateToPipelineOverview(): void { - this.hide(true); - this.router.navigate(['pipelines']); - } - - hide(skipReload: boolean) { - let reloadConfig = undefined; - if (!skipReload) { - reloadConfig = this.operationSuccess - ? { reload: true, pipelineId: this.pipelineId } - : undefined; - } - this.dialogRef.close(reloadConfig); - } - - async addToAsset(): Promise<void> { - let linkageData: LinkageData[] = []; - linkageData = await this.addPipelineLinkageData(linkageData); - - await this.saveAssets(linkageData); - } - private async addPipelineLinkageData( - linkageData: LinkageData[], - ): Promise<LinkageData[]> { - const pipeline = await firstValueFrom( - this.pipelineService.getPipelineById(this.pipelineId), - ); - - linkageData.push({ - type: 'pipeline', - id: this.pipelineId, - name: pipeline.name, - }); - - const serviceList: DataSinkInvocation[] = - pipeline.actions as DataSinkInvocation[]; - const dataSinkServices: DataSinkInvocation[] = serviceList.filter( - action => action.serviceTagPrefix === 'DATA_SINK', - ); - - for (const service of dataSinkServices) { - const staticProperty = service.staticProperties.find( - prop => prop.internalName === 'db_measurement', - ); - - const measureFromPipeline = (staticProperty as { value: string }) - .value; - - const measure = await lastValueFrom( - this.dataLakeService.getMeasurementByName(measureFromPipeline), - ); - - linkageData.push({ - type: 'measurement', - id: measure.elementId, - name: measureFromPipeline, - }); - } - return linkageData; - } - - private async saveAssets(linkageData: LinkageData[]): Promise<void> { - await this.assetSaveService.saveSelectedAssets( - this.selectedAssets, - linkageData, - this.deselectedAssets, - this.originalAssets, - ); - } -}
