feat(medusa): Use Bull jobId option to identify duplicates (#3351)

* feat(medusa): Use Bull jobId option to identify duplicates

* Create pink-emus-drum.md
This commit is contained in:
Oliver Windall Juhl authored and GitHub committed 2023-03-01 18:11:30 +01:00
1 parent b458615ed5
commit 4b114cc419
2 files changed
+17 -7

No files matched your search

+5
View File
@@ -0,0 +1,5 @@
---
"@medusajs/medusa": patch
---
feat(medusa): Use Bull `jobId` option to identify duplicates
+12 -7
View File
@@ -1,4 +1,4 @@
import Bull from "bull" import Bull, { JobOptions } from "bull"
import Redis from "ioredis" import Redis from "ioredis"
import { isDefined } from "medusa-core-utils" import { isDefined } from "medusa-core-utils"
import { EntityManager } from "typeorm" import { EntityManager } from "typeorm"
@@ -40,14 +40,16 @@ type SubscriberDescriptor = {
subscriber: Subscriber subscriber: Subscriber
} }
type EmitOptions = { export type EmitOptions = {
delay?: number delay?: number
attempts: number attempts: number
backoff?: { backoff?: {
type: "fixed" | "exponential" type: "fixed" | "exponential"
delay: number delay: number
} }
} } & JobOptions
const COMPLETED_JOB_TTL = 10000
/** /**
* Can keep track of multiple subscribers to different events and run the * Can keep track of multiple subscribers to different events and run the
@@ -225,10 +227,13 @@ export default class EventBusService {
data: T, data: T,
options: Record<string, unknown> & EmitOptions = { attempts: 1 } options: Record<string, unknown> & EmitOptions = { attempts: 1 }
): Promise<StagedJob | void> { ): Promise<StagedJob | void> {
const opts: { removeOnComplete: boolean } & EmitOptions = { const opts: EmitOptions = {
removeOnComplete: true, removeOnComplete: {
attempts: 1, age: COMPLETED_JOB_TTL,
},
...options,
} }
if (typeof options.attempts === "number") { if (typeof options.attempts === "number") {
opts.attempts = options.attempts opts.attempts = options.attempts
if (isDefined(options.backoff)) { if (isDefined(options.backoff)) {
@@ -295,7 +300,7 @@ export default class EventBusService {
this.queue_ this.queue_
.add( .add(
{ eventName: job.event_name, data: job.data }, { eventName: job.event_name, data: job.data },
job.options ?? { removeOnComplete: true } { jobId: job.id, ...job.options }
) )
.then(async () => { .then(async () => {
await stagedJobRepo.remove(job) await stagedJobRepo.remove(job)