fix(medusa): add support for retrying failed event bus jobs (#2566)
Closes CORE-770
This commit is contained in:
1 parent
5ea4b728e7
commit
8069ed5e99
2 files changed
+22
-2
No files matched your search
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@medusajs/medusa": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
fix(medusa): add support for retrying failed event bus jobs
|
||||||
@@ -16,6 +16,15 @@ type InjectedDependencies = {
|
|||||||
|
|
||||||
type Subscriber<T = unknown> = (data: T, eventName: string) => Promise<void>
|
type Subscriber<T = unknown> = (data: T, eventName: string) => Promise<void>
|
||||||
|
|
||||||
|
type EmitOptions = {
|
||||||
|
delay?: number
|
||||||
|
attempts?: number
|
||||||
|
backoff?: {
|
||||||
|
type: "fixed" | "exponential"
|
||||||
|
delay: number
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Can keep track of multiple subscribers to different events and run the
|
* Can keep track of multiple subscribers to different events and run the
|
||||||
* subscribers when events happen. Events will run asynchronously.
|
* subscribers when events happen. Events will run asynchronously.
|
||||||
@@ -179,7 +188,7 @@ export default class EventBusService {
|
|||||||
async emit<T>(
|
async emit<T>(
|
||||||
eventName: string,
|
eventName: string,
|
||||||
data: T,
|
data: T,
|
||||||
options: { delay?: number } = {}
|
options: EmitOptions = {}
|
||||||
): Promise<StagedJob | void> {
|
): Promise<StagedJob | void> {
|
||||||
if (this.transactionManager_) {
|
if (this.transactionManager_) {
|
||||||
const stagedJobRepository = this.transactionManager_.getCustomRepository(
|
const stagedJobRepository = this.transactionManager_.getCustomRepository(
|
||||||
@@ -192,9 +201,15 @@ export default class EventBusService {
|
|||||||
})
|
})
|
||||||
return await stagedJobRepository.save(stagedJobInstance)
|
return await stagedJobRepository.save(stagedJobInstance)
|
||||||
} else {
|
} else {
|
||||||
const opts: { removeOnComplete: boolean; delay?: number } = {
|
const opts: { removeOnComplete: boolean } & EmitOptions = {
|
||||||
removeOnComplete: true,
|
removeOnComplete: true,
|
||||||
}
|
}
|
||||||
|
if (typeof options.attempts === "number") {
|
||||||
|
opts.attempts = options.attempts
|
||||||
|
if (typeof options.backoff !== "undefined") {
|
||||||
|
opts.backoff = options.backoff
|
||||||
|
}
|
||||||
|
}
|
||||||
if (typeof options.delay === "number") {
|
if (typeof options.delay === "number") {
|
||||||
opts.delay = options.delay
|
opts.delay = options.delay
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user