Skip to content

Commit

Permalink
fix: correct order of scheme pre-startup tasks
Browse files Browse the repository at this point in the history
  • Loading branch information
nvsukhanov committed Mar 13, 2024
1 parent fb7e2b5 commit ef2a8e1
Showing 1 changed file with 38 additions and 19 deletions.
Original file line number Diff line number Diff line change
@@ -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';
Expand All @@ -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<Observable<unknown>>
): MonoTypeOperatorFunction<unknown> {
if (tasks.length === 0) {
return (source) => source;
}

return (source) => source.pipe(
switchMap(() => forkJoin(tasks)),
last()
);
}

function runTasksSequentially(
tasks: Array<Observable<unknown>>
): MonoTypeOperatorFunction<unknown> {
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),
Expand All @@ -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 })))
Expand Down

0 comments on commit ef2a8e1

Please sign in to comment.