asyavee/medusa-store
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

refactor(medusa): Refactor and fix EventBusService (#1377)

Browse Source
This commit is contained in:
Adrien de Peretti
2022-06-09 17:18:22 +01:00
committed by GitHub
parent 78bd61abe1
commit 5172b21d09
8 changed files with 376 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/medusa/src/services/__tests__/event-bus.js
+6 -6
View File
@@ -32,7 +32,7 @@ describe("EventBusService", () => {
})
afterAll(async () => {
await await eventBus.stopEnqueuer()
await eventBus.stopEnqueuer()
})
it("creates bull queue", () => {
@@ -64,7 +64,7 @@ describe("EventBusService", () => {
})
it("added the subscriber to the queue", () => {
expect(eventBus.observers_["eventName"].length).toEqual(1)
expect(eventBus.observers_.get("eventName").length).toEqual(1)
})
})
@@ -138,7 +138,7 @@ describe("EventBusService", () => {
manager: MockManager,
stagedJobRepository,
logger: loggerMock,
})
}, {})
eventBus.subscribe("eventName", () => Promise.resolve("hi"))
result = await eventBus.worker_({
data: { eventName: "eventName", data: {} },
@@ -191,13 +191,13 @@ describe("EventBusService", () => {
it("calls logger warn on rejections", () => {
expect(loggerMock.warn).toHaveBeenCalledTimes(3)
expect(loggerMock.warn).toHaveBeenCalledWith(
"An error occured while processing eventName: fail1"
"An error occurred while processing eventName: fail1"
)
expect(loggerMock.warn).toHaveBeenCalledWith(
"An error occured while processing eventName: fail2"
"An error occurred while processing eventName: fail2"
)
expect(loggerMock.warn).toHaveBeenCalledWith(
"An error occured while processing eventName: fail3"
"An error occurred while processing eventName: fail3"
)
})
packages/medusa/src/services/event-bus.js
-293
View File
@@ -1,293 +0,0 @@
import Bull from "bull"
import Redis from "ioredis"
/**
* Can keep track of multiple subscribers to different events and run the
* subscribers when events happen. Events will run asynchronously.
* @class
*/
class EventBusService {
constructor(
{ manager, logger, stagedJobRepository, redisClient, redisSubscriber },
config,
singleton = true
) {
const opts = {
createClient: (type) => {
switch (type) {
case "client":
return redisClient
case "subscriber":
return redisSubscriber
default:
if (config.projectConfig.redis_url) {
return new Redis(config.projectConfig.redis_url)
}
return redisClient
}
},
}
this.config_ = config
/** @private {EntityManager} */
this.manager_ = manager
/** @private {logger} */
this.logger_ = logger
this.stagedJobRepository_ = stagedJobRepository
if (singleton) {
/** @private {object} */
this.observers_ = {}
/** @private {BullQueue} */
this.queue_ = new Bull(`${this.constructor.name}:queue`, opts)
/** @private {object} to handle cron jobs */
this.cronHandlers_ = {}
this.redisClient_ = redisClient
this.redisSubscriber_ = redisSubscriber
/** @private {BullQueue} used for cron jobs */
this.cronQueue_ = new Bull(`cron-jobs:queue`, opts)
// Register our worker to handle emit calls
this.queue_.process(this.worker_)
// Register cron worker
this.cronQueue_.process(this.cronWorker_)
if (process.env.NODE_ENV !== "test") {
this.startEnqueuer()
}
}
}
withTransaction(transactionManager) {
if (!transactionManager) {
return this
}
const cloned = new EventBusService(
{
manager: transactionManager,
stagedJobRepository: this.stagedJobRepository_,
logger: this.logger_,
redisClient: this.redisClient_,
redisSubscriber: this.redisSubscriber_,
},
this.config_,
false
)
cloned.transactionManager_ = transactionManager
cloned.queue_ = this.queue_
return cloned
}
/**
* Adds a function to a list of event subscribers.
* @param {string} event - the event that the subscriber will listen for.
* @param {func} subscriber - the function to be called when a certain event
* happens. Subscribers must return a Promise.
*/
subscribe(event, subscriber) {
if (typeof subscriber !== "function") {
throw new Error("Subscriber must be a function")
}
if (this.observers_[event]) {
this.observers_[event].push(subscriber)
} else {
this.observers_[event] = [subscriber]
}
}
/**
* Adds a function to a list of event subscribers.
* @param {string} event - the event that the subscriber will listen for.
* @param {func} subscriber - the function to be called when a certain event
* happens. Subscribers must return a Promise.
*/
unsubscribe(event, subscriber) {
if (typeof subscriber !== "function") {
throw new Error("Subscriber must be a function")
}
if (this.observers_[event]) {
const index = this.observers_[event].indexOf(subscriber)
if (index !== -1) {
this.observers_[event].splice(index, 1)
}
}
}
/**
* Adds a function to a list of event subscribers.
* @param {string} event - the event that the subscriber will listen for.
* @param {func} subscriber - the function to be called when a certain event
* happens. Subscribers must return a Promise.
*/
registerCronHandler_(event, subscriber) {
if (typeof subscriber !== "function") {
throw new Error("Handler must be a function")
}
if (this.observers_[event]) {
this.cronHandlers_[event].push(subscriber)
} else {
this.cronHandlers_[event] = [subscriber]
}
}
/**
* Calls all subscribers when an event occurs.
* @param {string} eventName - the name of the event to be process.
* @param {?any} data - the data to send to the subscriber.
* @param {?any} options - options to add the job with
* @return {BullJob} - the job from our queue
*/
async emit(eventName, data, options = {}) {
if (this.transactionManager_) {
const stagedJobRepository = this.transactionManager_.getCustomRepository(
this.stagedJobRepository_
)
const created = await stagedJobRepository.create({
event_name: eventName,
data,
})
return stagedJobRepository.save(created)
} else {
const opts = { removeOnComplete: true }
if (typeof options.delay === "number") {
opts.delay = options.delay
}
this.queue_.add({ eventName, data }, opts)
}
}
async sleep(ms) {
return new Promise((resolve) => {
setTimeout(resolve, ms)
})
}
async startEnqueuer() {
this.enRun_ = true
this.enqueue_ = this.enqueuer_()
}
async stopEnqueuer() {
this.enRun_ = false
await this.enqueue_
}
async enqueuer_() {
while (this.enRun_) {
const listConfig = {
relations: [],
skip: 0,
take: 1000,
}
const sjRepo = this.manager_.getCustomRepository(
this.stagedJobRepository_
)
const jobs = await sjRepo.find({}, listConfig)
await Promise.all(
jobs.map((job) => {
this.queue_
.add(
{ eventName: job.event_name, data: job.data },
{ removeOnComplete: true }
)
.then(async () => {
await sjRepo.remove(job)
})
})
)
await this.sleep(3000)
}
}
/**
* Handles incoming jobs.
* @param {Object} job The job object
* @return {Promise} resolves to the results of the subscriber calls.
*/
worker_ = (job) => {
const { eventName, data } = job.data
const eventObservers = this.observers_[eventName] || []
const wildcardObservers = this.observers_["*"] || []
const observers = eventObservers.concat(wildcardObservers)
this.logger_.info(
`Processing ${eventName} which has ${eventObservers.length} subscribers`
)
return Promise.all(
observers.map((subscriber) => {
return subscriber(data, eventName).catch((err) => {
this.logger_.warn(
`An error occured while processing ${eventName}: ${err}`
)
console.log(err)
return err
})
})
)
}
/**
* Handles incoming jobs.
* @param {Object} job The job object
* @return {Promise} resolves to the results of the subscriber calls.
*/
cronWorker_ = (job) => {
const { eventName, data } = job.data
const observers = this.cronHandlers_[eventName] || []
this.logger_.info(`Processing cron job: ${eventName}`)
return Promise.all(
observers.map((subscriber) => {
return subscriber(data, eventName).catch((err) => {
this.logger_.warn(
`An error occured while processing ${eventName}: ${err}`
)
return err
})
})
)
}
/**
* Registers a cron job.
* @param {string} eventName - the name of the event
* @param {object} data - the data to be sent with the event
* @param {string} cron - the cron pattern
* @param {function} handler - the handler to call on each cron job
* @return {void}
*/
createCronJob(eventName, data, cron, handler) {
this.logger_.info(`Registering ${eventName}`)
this.registerCronHandler_(eventName, handler)
return this.cronQueue_.add(
{
eventName,
data,
},
{ repeat: { cron } }
)
}
}
export default EventBusService
packages/medusa/src/services/event-bus.ts
+324
View File
@@ -0,0 +1,324 @@
import Bull from "bull"
import Redis from "ioredis"
import { EntityManager } from "typeorm"
import { ConfigModule, Logger } from "../types/global"
import { StagedJobRepository } from "../repositories/staged-job"
import { StagedJob } from "../models"
import { sleep } from "../utils/sleep"
type InjectedDependencies = {
manager: EntityManager
logger: Logger
stagedJobRepository: typeof StagedJobRepository
redisClient: Redis
redisSubscriber: Redis
}
type Subscriber<T = unknown> = (data: T, eventName: string) => Promise<void>
/**
* Can keep track of multiple subscribers to different events and run the
* subscribers when events happen. Events will run asynchronously.
*/
export default class EventBusService {
protected readonly config_: ConfigModule
protected readonly manager_: EntityManager
protected readonly logger_: Logger
protected readonly stagedJobRepository_: typeof StagedJobRepository
protected readonly observers_: Map<string | symbol, Subscriber[]>
protected readonly cronHandlers_: Map<string | symbol, Subscriber[]>
protected readonly redisClient_: Redis
protected readonly redisSubscriber_: Redis
protected readonly cronQueue_: Bull
protected queue_: Bull
protected shouldEnqueuerRun: boolean
protected transactionManager_: EntityManager | undefined
protected enqueue_: Promise<void>
constructor(
{
manager,
logger,
stagedJobRepository,
redisClient,
redisSubscriber,
}: InjectedDependencies,
config: ConfigModule,
singleton = true
) {
const opts = {
createClient: (type: string): Redis => {
switch (type) {
case "client":
return redisClient
case "subscriber":
return redisSubscriber
default:
if (config.projectConfig.redis_url) {
return new Redis(config.projectConfig.redis_url)
}
return redisClient
}
},
}
this.config_ = config
this.manager_ = manager
this.logger_ = logger
this.stagedJobRepository_ = stagedJobRepository
if (singleton) {
this.observers_ = new Map()
this.queue_ = new Bull(`${this.constructor.name}:queue`, opts)
this.cronHandlers_ = new Map()
this.redisClient_ = redisClient
this.redisSubscriber_ = redisSubscriber
this.cronQueue_ = new Bull(`cron-jobs:queue`, opts)
// Register our worker to handle emit calls
this.queue_.process(this.worker_)
// Register cron worker
this.cronQueue_.process(this.cronWorker_)
if (process.env.NODE_ENV !== "test") {
this.startEnqueuer()
}
}
}
withTransaction(transactionManager): this | EventBusService {
if (!transactionManager) {
return this
}
const cloned = new EventBusService(
{
manager: transactionManager,
stagedJobRepository: this.stagedJobRepository_,
logger: this.logger_,
redisClient: this.redisClient_,
redisSubscriber: this.redisSubscriber_,
},
this.config_,
false
)
cloned.transactionManager_ = transactionManager
cloned.queue_ = this.queue_
return cloned
}
/**
* Adds a function to a list of event subscribers.
* @param event - the event that the subscriber will listen for.
* @param subscriber - the function to be called when a certain event
* happens. Subscribers must return a Promise.
* @return this
*/
subscribe(event: string | symbol, subscriber: Subscriber): this {
if (typeof subscriber !== "function") {
throw new Error("Subscriber must be a function")
}
const observers = this.observers_.get(event) ?? []
this.observers_.set(event, [...observers, subscriber])
return this
}
/**
* Adds a function to a list of event subscribers.
* @param event - the event that the subscriber will listen for.
* @param subscriber - the function to be called when a certain event
* happens. Subscribers must return a Promise.
* @return this
*/
unsubscribe(event: string | symbol, subscriber: Subscriber): this {
if (typeof subscriber !== "function") {
throw new Error("Subscriber must be a function")
}
if (this.observers_.get(event)?.length) {
const index = this.observers_.get(event)?.indexOf(subscriber)
if (index !== -1) {
this.observers_.get(event)?.splice(index as number, 1)
}
}
return this
}
/**
* Adds a function to a list of event subscribers.
* @param event - the event that the subscriber will listen for.
* @param subscriber - the function to be called when a certain event
* happens. Subscribers must return a Promise.
* @return this
*/
protected registerCronHandler_(
event: string | symbol,
subscriber: Subscriber
): this {
if (typeof subscriber !== "function") {
throw new Error("Handler must be a function")
}
const cronHandlers = this.cronHandlers_.get(event) ?? []
this.cronHandlers_.set(event, [...cronHandlers, subscriber])
return this
}
/**
* Calls all subscribers when an event occurs.
* @param {string} eventName - the name of the event to be process.
* @param data - the data to send to the subscriber.
* @param options - options to add the job with
* @return the job from our queue
*/
async emit<T>(
eventName: string,
data: T,
options: { delay?: number } = {}
): Promise<StagedJob | void> {
if (this.transactionManager_) {
const stagedJobRepository = this.transactionManager_.getCustomRepository(
this.stagedJobRepository_
)
const stagedJobInstance = stagedJobRepository.create({
event_name: eventName,
data,
})
return await stagedJobRepository.save(stagedJobInstance)
} else {
const opts: { removeOnComplete: boolean; delay?: number } = {
removeOnComplete: true,
}
if (typeof options.delay === "number") {
opts.delay = options.delay
}
this.queue_.add({ eventName, data }, opts)
}
}
startEnqueuer(): void {
this.shouldEnqueuerRun = true
this.enqueue_ = this.enqueuer_()
}
async stopEnqueuer(): Promise<void> {
this.shouldEnqueuerRun = false
await this.enqueue_
}
async enqueuer_(): Promise<void> {
while (this.shouldEnqueuerRun) {
const listConfig = {
relations: [],
skip: 0,
take: 1000,
}
const stagedJobRepo = this.manager_.getCustomRepository(
this.stagedJobRepository_
)
const jobs = await stagedJobRepo.find(listConfig)
await Promise.all(
jobs.map((job) => {
this.queue_
.add(
{ eventName: job.event_name, data: job.data },
{ removeOnComplete: true }
)
.then(async () => {
await stagedJobRepo.remove(job)
})
})
)
await sleep(3000)
}
}
/**
* Handles incoming jobs.
* @param job The job object
* @return resolves to the results of the subscriber calls.
*/
worker_ = async <T>(job: {
data: { eventName: string; data: T }
}): Promise<unknown[]> => {
const { eventName, data } = job.data
const eventObservers = this.observers_.get(eventName) || []
const wildcardObservers = this.observers_.get("*") || []
const observers = eventObservers.concat(wildcardObservers)
this.logger_.info(
`Processing ${eventName} which has ${eventObservers.length} subscribers`
)
return await Promise.all(
observers.map((subscriber) => {
return subscriber(data, eventName).catch((err) => {
this.logger_.warn(
`An error occurred while processing ${eventName}: ${err}`
)
console.error(err)
return err
})
})
)
}
/**
* Handles incoming jobs.
* @param job The job object
* @return resolves to the results of the subscriber calls.
*/
cronWorker_ = async <T>(job: {
data: { eventName: string; data: T }
}): Promise<unknown[]> => {
const { eventName, data } = job.data
const observers = this.cronHandlers_.get(eventName) || []
this.logger_.info(`Processing cron job: ${eventName}`)
return await Promise.all(
observers.map((subscriber) => {
return subscriber(data, eventName).catch((err) => {
this.logger_.warn(
`An error occured while processing ${eventName}: ${err}`
)
return err
})
})
)
}
/**
* Registers a cron job.
* @param eventName - the name of the event
* @param data - the data to be sent with the event
* @param cron - the cron pattern
* @param handler - the handler to call on each cron job
* @return void
*/
createCronJob<T>(
eventName: string,
data: T,
cron: string,
handler: Subscriber
): void {
this.logger_.info(`Registering ${eventName}`)
this.registerCronHandler_(eventName, handler)
return this.cronQueue_.add(
{
eventName,
data,
},
{ repeat: { cron } }
)
}
}
packages/medusa/src/services/product-variant.ts
+2 -2
View File
@@ -17,8 +17,8 @@ import {
FindWithRelationsOptions,
ProductVariantRepository,
} from "../repositories/product-variant"
import EventBusService from "../services/event-bus"
import RegionService from "../services/region"
import EventBusService from "./event-bus"
import RegionService from "./region"
import { FindConfig } from "../types/common"
import {
CreateProductVariantInput,
packages/medusa/src/types/global.ts
+2
View File
@@ -23,6 +23,8 @@ export type MedusaContainer = AwilixContainer & {
export type Logger = _Logger & {
progress: (activityId: string, msg: string) => void
info: (msg: string) => void
warn: (msg: string) => void
}
export type ConfigModule = {
packages/medusa/src/utils/sleep.ts
+5
View File
@@ -0,0 +1,5 @@
export async function sleep(ms: number) {
return new Promise((resolve) => {
setTimeout(resolve, ms)
})
}