40 lines
1.3 KiB
TypeScript
40 lines
1.3 KiB
TypeScript
import * as plugins from './taskbuffer.plugins.js';
|
|
|
|
import { Task, type ITaskFunction } from './taskbuffer.classes.task.js';
|
|
import { logger } from './taskbuffer.logging.js';
|
|
|
|
export class TaskDebounced<T = unknown> extends Task {
|
|
private _debouncedTaskFunction: ITaskFunction;
|
|
private _observableIntake = new plugins.smartrx.ObservableIntake<T>();
|
|
|
|
constructor(optionsArg: {
|
|
name: string;
|
|
taskFunction: ITaskFunction;
|
|
debounceTimeInMillis: number;
|
|
}) {
|
|
super({
|
|
name: optionsArg.name,
|
|
taskFunction: async (x: T) => {
|
|
this._observableIntake.push(x);
|
|
},
|
|
});
|
|
this.taskFunction = optionsArg.taskFunction;
|
|
this._observableIntake.observable
|
|
.pipe(
|
|
plugins.smartrx.rxjs.ops.debounceTime(optionsArg.debounceTimeInMillis),
|
|
)
|
|
.subscribe({
|
|
next: async (x) => {
|
|
try {
|
|
await this.taskFunction(x);
|
|
} catch (err) {
|
|
logger.log('error', `TaskDebounced "${this.name || 'unnamed'}" failed: ${err instanceof Error ? err.message : String(err)}`);
|
|
}
|
|
},
|
|
error: (err) => {
|
|
logger.log('error', `TaskDebounced "${this.name || 'unnamed'}" observable error: ${err instanceof Error ? err.message : String(err)}`);
|
|
},
|
|
});
|
|
}
|
|
}
|