taskbuffer/ts/taskbuffer.classes.taskdebounced.ts

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)}`);
        },
      });
  }
}