Using Observable with Batches

[lukepolo]

Is this possible? I see that manually processing jobs is a limitation , though this is kinda a mixture,

Currently getting[TimeoutError]: Timeout has occurred

this.worker = new WorkerPro(
      this.queueName,
      async (batch, token) => {
        return new Observable((subscriber) => {
          async function processJob(job: JobPro) {
            await job.getChildrenValues().then((childrenValues) => {
              callback(this.convertToJob(job, token), childrenValues)
                .then((data) => {
                  subscriber.next(data);
                  subscriber.complete();
                })
                .catch(async (error) => {
                  subscriber.error(error);
                });
            });
          }

          if(!workerOptions.batch?.size) {
             return processJob.bind(this)(batch);
          }

          return Promise.all(batch.getBatch().map(processJob.bind(this)));
        });
      },
      workerOptions,
    )
      .on("error", (error) => {
        this.handleError(BullMQErrorTypes.Worker, error);
      })
      .on("completed", async (batch: JobPro, result) => {
        if(!workerOptions.batch?.size) {
          return successCallback(this.convertToJob(batch), result);
        }

        for(const job of batch.getBatch()) {
          await successCallback(this.convertToJob(job), result);
        }
      })
      .on("failed", async (batch: JobPro, error) => {
        if(!workerOptions.batch?.size) {
          return failedCallback(this.convertToJob(batch), error);
        }

        for(const job of batch.getBatch()) {
          await failedCallback(this.convertToJob(job), error);
        }
      });

[lukepolo]

ok I found its my TTL , i setup a TTL for all my jobs, but the wrapper didn't have one , working on how to add it to to the batch

[lukepolo]

Ok after fixing the TTL it seems now if i have a batch larger than 1 it fails to secure the lock

QueueWorker <error> Unable to process job (HelloJob): Error: Missing lock for job 2538,2539,2540,2541,2542,2543,2544,2545,2546,2547. storeResult
    at QueueEventsPro.<anonymous> (/Users/luke/code/work/framework/qx-microservice-template/dist/app.js:6494:64)
    at process.processTicksAndRejections (node:internal/process/task_queues:95:5) +243ms

[manast]

I think we do not have proper tests for this case, so we need to do that in order to guarantee the correct behaviour.

[lukepolo]

Gotcha for now if we enable batches for the worker ill ignore the TTL