feat: added ability to register workers manually

packages/bullmq/e2e/module.e2e-spec.ts

@@ -11,6 +11,7 @@ import {
  QueueEventsHost,
  QueueEventsListener,
  WorkerHost,
+ BullRegistrator,
 } from '../lib';
 
 jest.setTimeout(10000);
@@ -677,6 +678,148 @@ describe('BullModule', () => {
  });
  });
 
+ describe('manual registration', () => {
+ const queueName = 'a_queue';
+
+ it('should manually register workers - forRoot', (done) => {
+ const processorCalledSpy = jest.fn();
+ const queueCompletedEventSpy = jest.fn();
+ const workerCompletedEventSpy = jest.fn();
+
+ @QueueEventsListener(queueName)
+ class EventsListener extends QueueEventsHost {
+ @OnQueueEvent('completed')
+ onCompleted() {
+ queueCompletedEventSpy();
+ }
+ }
+
+ @Processor(queueName)
+ class TestProcessor extends WorkerHost {
+ async process(job: Job<any, any, string>): Promise<any> {
+ await new Promise((resolve) => setTimeout(resolve, 1500));
+ processorCalledSpy();
+ }
+
+ @OnWorkerEvent('completed')
+ onCompleted() {
+ workerCompletedEventSpy();
+ }
+ }
+
+ Test.createTestingModule({
+ imports: [
+ BullModule.forRoot({
+ connection: {
+ host: '0.0.0.0',
+ port: 6380,
+ },
+ extraOptions: {
+ manualRegistration: true,
+ },
+ }),
+ BullModule.registerQueue({
+ name: queueName,
+ }),
+ ],
+ providers: [EventsListener, TestProcessor],
+ })
+ .compile()
+ .then(async (testingModule) => {
+ await testingModule.init();
+
+ expect(() => testingModule.get(TestProcessor).worker).toThrow(
+ '"Worker" has not yet been initialized. Make sure to interact with worker instances after the "onModuleInit" lifecycle hook is triggered for example, in the "onApplicationBootstrap" hook, or if "manualRegistration" is set to true make sure to call "BullRegistrator.register()"',
+ );
+ expect(() => testingModule.get(EventsListener).queueEvents).toThrow(
+ '"QueueEvents" class has not yet been initialized. Make sure to interact with queue events instances after the "onModuleInit" lifecycle hook is triggered, for example, in the "onApplicationBootstrap" hook, or if "manualRegistration" is set to true make sure to call "BullRegistrator.register()"',
+ );
+
+ const bullRegistrator = testingModule.get(BullRegistrator);
+ bullRegistrator.register();
+
+ const processorWorker = testingModule.get(TestProcessor).worker;
+ const queueEvents = testingModule.get(EventsListener).queueEvents;
+
+ expect(processorWorker).toBeDefined();
+ expect(queueEvents).toBeDefined();
+
+ await testingModule.close();
+ done();
+ });
+ });
+
+ it('should manually register workers - forRootAsync', (done) => {
+ const processorCalledSpy = jest.fn();
+ const queueCompletedEventSpy = jest.fn();
+ const workerCompletedEventSpy = jest.fn();
+
+ @QueueEventsListener(queueName)
+ class EventsListener extends QueueEventsHost {
+ @OnQueueEvent('completed')
+ onCompleted() {
+ queueCompletedEventSpy();
+ }
+ }
+
+ @Processor(queueName)
+ class TestProcessor extends WorkerHost {
+ async process(job: Job<any, any, string>): Promise<any> {
+ await new Promise((resolve) => setTimeout(resolve, 1500));
+ processorCalledSpy();
+ }
+
+ @OnWorkerEvent('completed')
+ onCompleted() {
+ workerCompletedEventSpy();
+ }
+ }
+
+ Test.createTestingModule({
+ imports: [
+ BullModule.forRootAsync({
+ useFactory: () => ({
+ connection: {
+ host: '0.0.0.0',
+ port: 6380,
+ },
+ }),
+ extraOptions: {
+ manualRegistration: true,
+ },
+ }),
+ BullModule.registerQueue({
+ name: queueName,
+ }),
+ ],
+ providers: [EventsListener, TestProcessor],
+ })
+ .compile()
+ .then(async (testingModule) => {
+ await testingModule.init();
+
+ expect(() => testingModule.get(TestProcessor).worker).toThrow(
+ '"Worker" has not yet been initialized. Make sure to interact with worker instances after the "onModuleInit" lifecycle hook is triggered for example, in the "onApplicationBootstrap" hook, or if "manualRegistration" is set to true make sure to call "BullRegistrator.register()"',
+ );
+ expect(() => testingModule.get(EventsListener).queueEvents).toThrow(
+ '"QueueEvents" class has not yet been initialized. Make sure to interact with queue events instances after the "onModuleInit" lifecycle hook is triggered, for example, in the "onApplicationBootstrap" hook, or if "manualRegistration" is set to true make sure to call "BullRegistrator.register()"',
+ );
+
+ const bullRegistrator = testingModule.get(BullRegistrator);
+ bullRegistrator.register();
+
+ const processorWorker = testingModule.get(TestProcessor).worker;
+ const queueEvents = testingModule.get(EventsListener).queueEvents;
+
+ expect(processorWorker).toBeDefined();
+ expect(queueEvents).toBeDefined();
+
+ await testingModule.close();
+ done();
+ });
+ });
+ });
+
  describe('handles all kind of valid processors providers', () => {
  @Processor('test_processor_registering')
  class MyProcessorA extends WorkerHost {

packages/bullmq/lib/bull.explorer.ts

@@ -30,7 +30,7 @@ import { NestQueueOptions } from './interfaces/queue-options.interface';
 import { NestWorkerOptions } from './interfaces/worker-options.interface';
 
 @Injectable()
-export class BullExplorer implements OnModuleInit {
+export class BullExplorer {
  private static _workerClass: Type = Worker;
  private readonly logger = new Logger('BullModule');
  private readonly injector = new Injector();
@@ -46,7 +46,7 @@ export class BullExplorer implements OnModuleInit {
  private readonly metadataScanner: MetadataScanner,
  ) {}
 
- onModuleInit() {
+ register() {
  this.registerWorkers();
  this.registerQueueEventListeners();
  }