Better API proposal & move to BullMQ

[panki]

I'm submitting a better api proposal

I think the value that this library brings to developers may be significantly improved. Please, take a look at the API proposal below:

example.types.ts

interface Task1Data {
  param1: string;
  param2: number;
}

interface Task2Data {
  param1: string;
}

example.queue.ts

@Queue('example', { ...other queue options })
export class ExampleQueue extends AbstractQueue {
  
  // provided by AbstractQueue for full queue control if needed
  private readonly queue: BullQueue; 

  // Job decorator creates a job and adds it to bull queue
  // with data returned from the method
  @Job()
  async task1(param1: string, param2: number): Promise<Task1Data> {
    return {
      param1,
      param2
    }
  }

  // RepeatableJob decorator allows to automatically register the job
  // in the bull queue and acts as a guard to disallow calling method
  @RepeatableJob({
    repeat: { cron: '* * * * *' },
    attempts: 1,
  })
  async task2(): Promise<Task2Data> {
    return { param1: 'param' };
  }
}

example.processor.ts

@Processor(ExampleQueue)
export class ExampleQueueProcessor {
  constructor(
    private readonly service: MyService,
    private readonly queue: ExampleQueue
  ) {}

  // Explicit queue job handler mapped by method name
  // 1) decorator gets metadata from corresponding ExampleQueue.task1 method
  // 2) decorator checks return type of ExampleQueue.task1 and input parameter type
  @Process()
  async task1(data: Task1Data) {
    return this.service.task1(data);
  }

  // Explicit queue job handler, if processor's method name
  // differs from queue method name
  @Process(ExampleQueue.task1)
  async processTask1(data: Task1Data) {
    return this.service.task1(data);
  }

  // Injection of job, if process need to do something with it
  @Process({ concurrency: 1 })
  async task2(data: Task2Data, @InjectJob() job: Job<Task2Data>) {
    await this.service.task2(data);
    // do something with job
    return 'return value';
  }
}

example.module.ts

@Module({
  imports: [
    BullModule.registerQueueAsync({
      imports: [ConfigModule],
      inject: [RedisConfig]
      useFactory: async (redis: RedisConfig): Promise<any> => [
        {
          queue: ExampleQueue,
          redis,
          ...other bull queue options
        },
      ],
    }),
  ],
  providers: [ExampleQueue, ExampleQueueProcessor],
  exports: [ExampleQueue],
})
export class ExampleModule {}

What is the motivation/use case for changing the behavior?

In my opinion, using API like this allows to:

1. Detect most of the configuration errors in compile time
2. Have explicit dependencies
3. Produce more clear code
4. Use types there it possible

I can describe exact responsibilities for each of the decorators: @Queue, @Job, @Processor, @Process, @InjectJob but first I would like to hear your opinion about this in general.

[kamilmysliwiec]

Thank you for sharing your thoughts.

@Queue('example', { ...other queue options })

This decorator looks promising. A way to declaratively define a queue would be nice.

... extends AbstractQueue {
  
  // provided by AbstractQueue for full queue control if needed
  private readonly queue: BullQueue;

Using DI with @InjectQueue seems to be more testable & cleaner in comparison to the proposed approach (with inheritance and abstract classes).

// Job decorator creates a job and adds it to bull queue
  // with data returned from the method
  @Job()
  async task1(param1: string, param2: number): Promise<Task1Data> {
    return {
      param1,
      param2
    }
  }

I'm not sure what would be the potential use-case for statically created one-time jobs that are added to the queue on application bootstrap and how - most importantly - this decorator simplifies this operation (instead of simply calling .add() explicitly).

// RepeatableJob decorator allows to automatically register the job
  // in the bull queue and acts as a guard to disallow calling method
  @RepeatableJob({
    repeat: { cron: '* * * * *' },
    attempts: 1,
  })

This looks great though. I'm guessing having @Job would be nice for the sake of consistency then?

@Process(ExampleQueue.task1)

This is impossible. task1 would have to be defined as a static method which doesn't make any sense.

async task2(data: Task2Data, @InjectJob() job: Job<Task2Data>) 

I'm not a big fan of this decorator either. I think we can just pass 2 arguments by default (job would be optional in this case).

[panki]

hi, @kamilmysliwiec , thank you for your response, I have some thoughts and suggestions based on your feedback, but I'm in the middle of moving out, please don't close this issue, I'll write my comments at the end of next week.

[VictorGaiva]

BullMq, which is the Beta repo for Bull 4.0, is available. I believe that any changes to this API should be built on top of the newer version.

[gperdomor]

@VictorGaiva well i think BullMQ was released under v1 and not as v4... The latest release is 1.6.7 from 6 days ago... https://github.com/taskforcesh/bullmq/releases

[fwoelffel]

@gperdomor I believe BullMQ is actually Bull 4.

@VictorGaiva Good catch. Maybe we should keep this in mind: https://docs.bullmq.io/bull-3.x-migration/compatibility-class

[gperdomor]

@fwoelffel yes, agreed

[VictorGaiva]

@fwoelffel It seems to be. I found the repo directly from main README:

BullMQ 4 Beta

If you want to start using the next major version of Bull you are welcome to the new repo here

[PaulMest]

I might be late to this thread, but I found this BullMQ blog post helpful to talk about how BullMQ is really the same team and just the next gen of bull: https://medium.com/@rockstudillo/beta-release-of-bullmq-4-0-867960aec51.

[brianschardt]

any updates here?

[VictorGaiva]

@brianschardt You can feel free to contribute in any way to the repo. We need someone to kickstart it

[twister21]

The proposed cronjob abstraction is far better than the current workaround.
I have to implement the OnApplicationBootstrap interface myself to enqueue the job after Nest's initialization is completed.

onApplicationBootstrap(): void {
  this.userCleanupQueue.add(null, { repeat: { cron: "0 1 * * *" } });
}

[stavalfi]

bullmq is not production ready. sometimes jobs are not processes. the reproduction is not that easy and i don't have time for it but there are problems. bullmq needs more time.

[main-kun]

@stavalfi Are there issues in bullmq repo for those bugs or any other evidence? I'm preparing a PR for the switch for this repo so i'd love to know more

[stavalfi]

as i remember, there is no such issue. i will try to find a time in the next 2-3 days to reproduce it in small repo.

[vh13294]

Any update on this ?

I am happy to beta test the new version.

[stavalfi]

sorry, don't wait for me. I don't have time to reproduce this. and maybe by now they already fixed the bug I faced.

[codetheweb]

Feel free to check out #720 if you want to test bullmq.

[michael-land]

Any plan move to bullmq?

[Pietrox]

@xiaoyu-tamu #754

There is a pretty fresh mention from @kamilmysliwiec on what may happen.
But honestly, if you use @nestjs/bull + bullmq instead of the bull it should work. I didn't test all the features but "Queue" class seems to work on the most of the features. And that is from last 2 weeks when I spent it on writting a queue feature in Nestjs project.

[kamilmysliwiec]

#720 (comment)