I am trying to make parallel queues for video processing.
However, I've faced problem in doing so. Below is the diagram of what i am trying to achieve.
Flow is:
- User sends
GETrequest to the/processendpoint (actually, GET is only for testing, it rather uses@MessagePatternto receive data from other service) - This request contains
ModelDTOas well assequencewhich is used for internal tracking - Controller imports
private readonly _queueService: QueueServicevia constructor - It then calls
this._queueService.process({ model, sequence }) QueueServiceimports@InjectQueue('video_processor') private readonly _processorQueue: Queuevia constructorQueueServicesimply callsthis._processorQueue.add('process', data);VideoProcessorimportsprivate readonly _videoService: VideoServicevia constructor- Inside
VideoProcessorthere is a method with@Process('process')decorator - Inside this method I am awaiting for the result from the service with
await this._videoService.configure(job.data).process()
And here is the problem:
- Whenever I am trying to run 1 job at a time (sending single request and actually waiting for job to complete) everything works just fine
- If I am queueing two jobs at the same time, for some reason, the
console.log(this._videoData.id)insideVideoServicewill now return the ID of the second model rather than actual ID.
So far I have tried adding scope: Scope.TRANSIENT to almost all services with no luck. Seems like i just can't figure out where this scope should be added.
I am expecting for 10 jobs to be able to run in parallel, however, if I add more than 1 job to the queue, they start mixing in data from the other jobs.

