diff --git a/modules/store/src/lib/effects/tasks-processing/scheme-pre-run/pre-run-scheme.effect.ts b/modules/store/src/lib/effects/tasks-processing/scheme-pre-run/pre-run-scheme.effect.ts index 50f31696..bb02ba87 100644 --- a/modules/store/src/lib/effects/tasks-processing/scheme-pre-run/pre-run-scheme.effect.ts +++ b/modules/store/src/lib/effects/tasks-processing/scheme-pre-run/pre-run-scheme.effect.ts @@ -1,5 +1,5 @@ import { Actions, concatLatestFrom, createEffect, ofType } from '@ngrx/effects'; -import { catchError, concatWith, filter, forkJoin, last, map, of, switchMap, tap, timeout } from 'rxjs'; +import { MonoTypeOperatorFunction, Observable, catchError, concatWith, filter, forkJoin, last, map, of, switchMap, timeout } from 'rxjs'; import { inject } from '@angular/core'; import { Store } from '@ngrx/store'; import { APP_CONFIG, IAppConfig } from '@app/shared-misc'; @@ -16,6 +16,31 @@ import { HubServoCalibrationFacadeService } from '../../../hub-facades'; import { IWidgetsReadTasksFactory, WIDGET_READ_TASKS_FACTORY } from './i-widgets-read-tasks-factory'; import { createPreRunMotorPositionQueryTasks } from './create-pre-run-motor-position-query-tasks'; +function runTasksInParallel( + tasks: Array> +): MonoTypeOperatorFunction { + if (tasks.length === 0) { + return (source) => source; + } + + return (source) => source.pipe( + switchMap(() => forkJoin(tasks)), + last() + ); +} + +function runTasksSequentially( + tasks: Array> +): MonoTypeOperatorFunction { + if (tasks.length === 0) { + return (source) => source; + } + return (source) => source.pipe( + concatWith(...tasks), + last() + ); +} + export const PRE_RUN_SCHEME_EFFECT = createEffect(( actions: Actions = inject(Actions), hubStorage: HubStorageService = inject(HubStorageService), @@ -30,24 +55,18 @@ export const PRE_RUN_SCHEME_EFFECT = createEffect(( map(([ , scheme ]) => scheme), filter((scheme): scheme is ControlSchemeModel => !!scheme), switchMap((scheme) => { - const combinedTasks = [ - ...createPreRunSetAccelerationProfileTasks(scheme, hubStorage), - ...createPreRunSetDecelerationProfileTasks(scheme, hubStorage), - ...createWidgetReadTasks(scheme, store, widgetReadTaskFactory), - ...createPreRunMotorPositionQueryTasks(scheme, hubStorage, store) - ]; - // TODO: move to Bindings module - const calibrationServoTasks = createPreRunServoCalibrationTasks(scheme, hubCalibrationFacade, store, appConfig); - if (combinedTasks.length + calibrationServoTasks.length === 0) { - return of(CONTROL_SCHEME_ACTIONS.schemeStarted({ name: scheme.name })); - } - - return forkJoin(combinedTasks).pipe( - // We have to start calibration tasks after all other tasks are done to avoid race conditions with position queries. - // Also, somehow running calibration tasks in parallel causes issues with receiving port values. - // For now, we run them sequentially. TODO: investigate and fix the root cause. - calibrationServoTasks.length > 0 ? concatWith(...calibrationServoTasks) : tap(() => void 0), - last(), + return of(null).pipe( + // Calibration tasks are run sequentially bc of some issue with port value reading while calibrating in parralel + // TODO: investigate and fix the issue. Not a priority now. + runTasksSequentially( + createPreRunServoCalibrationTasks(scheme, hubCalibrationFacade, store, appConfig) + ), + runTasksInParallel([ + ...createPreRunMotorPositionQueryTasks(scheme, hubStorage, store), + ...createPreRunSetAccelerationProfileTasks(scheme, hubStorage), + ...createPreRunSetDecelerationProfileTasks(scheme, hubStorage), + ...createWidgetReadTasks(scheme, store, widgetReadTaskFactory), + ]), timeout(appConfig.schemeStartStopTimeoutMs), map(() => CONTROL_SCHEME_ACTIONS.schemeStarted({ name: scheme.name })), catchError((e) => of(CONTROL_SCHEME_ACTIONS.schemeStartFailed({ reason: e })))