fix(): Workflow engine redis worker instance in worker mode (#14099)
* fix(): Workflow engine redis worker instance in worker mode * Create wicked-tips-buy.md
This commit is contained in:
1 parent
f67bfb9f92
commit
6c3ec528f1
2 files changed
+32
-27
No files matched your search
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@medusajs/workflow-engine-redis": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
fix(): Workflow engine redis worker instance in worker mode
|
||||||
@@ -145,37 +145,37 @@ export class RedisDistributedTransactionStorage
|
|||||||
// Remove all repeatable jobs from the old queue since now we have a queue dedicated to scheduled jobs
|
// Remove all repeatable jobs from the old queue since now we have a queue dedicated to scheduled jobs
|
||||||
await this.removeAllRepeatableJobs(this.queue)
|
await this.removeAllRepeatableJobs(this.queue)
|
||||||
|
|
||||||
this.worker = new Worker(
|
if (this.#isWorkerMode) {
|
||||||
this.queueName,
|
this.worker = new Worker(
|
||||||
async (job) => {
|
this.queueName,
|
||||||
this.logger_.debug(
|
async (job) => {
|
||||||
`executing job ${job.name} from queue ${
|
this.logger_.debug(
|
||||||
this.queueName
|
`executing job ${job.name} from queue ${
|
||||||
} with the following data: ${JSON.stringify(job.data)}`
|
this.queueName
|
||||||
)
|
} with the following data: ${JSON.stringify(job.data)}`
|
||||||
if (allowedJobs.includes(job.name as JobType)) {
|
)
|
||||||
try {
|
if (allowedJobs.includes(job.name as JobType)) {
|
||||||
await this.executeTransaction(
|
try {
|
||||||
job.data.workflowId,
|
await this.executeTransaction(
|
||||||
job.data.transactionId,
|
job.data.workflowId,
|
||||||
job.data.transactionMetadata
|
job.data.transactionId,
|
||||||
)
|
job.data.transactionMetadata
|
||||||
} catch (error) {
|
)
|
||||||
if (!SkipExecutionError.isSkipExecutionError(error)) {
|
} catch (error) {
|
||||||
throw error
|
if (!SkipExecutionError.isSkipExecutionError(error)) {
|
||||||
|
throw error
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
if (job.name === JobType.SCHEDULE) {
|
if (job.name === JobType.SCHEDULE) {
|
||||||
// Remove repeatable job from the old queue since now we have a queue dedicated to scheduled jobs
|
// Remove repeatable job from the old queue since now we have a queue dedicated to scheduled jobs
|
||||||
await this.remove(job.data.jobId)
|
await this.remove(job.data.jobId)
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
workerOptions
|
workerOptions
|
||||||
)
|
)
|
||||||
|
|
||||||
if (this.#isWorkerMode) {
|
|
||||||
this.jobWorker = new Worker(
|
this.jobWorker = new Worker(
|
||||||
this.jobQueueName,
|
this.jobQueueName,
|
||||||
async (job) => {
|
async (job) => {
|
||||||
|
|||||||
Reference in new issue
Block a user