Skip to content
This repository was archived by the owner on Jun 18, 2025. It is now read-only.

Commit a50ca76

Browse files
code-xhyuncode-xhyun
authored
fix: update @pulsecron/pulse dependency to version 1.1.9 in package.json (#4)
* fix: update @pulsecron/pulse dependency to version 1.1.9 in package.json * fix: @pulsecron/pulse type
1 parent 8bc1ca9 commit a50ca76

File tree

2 files changed

+65
-122
lines changed

2 files changed

+65
-122
lines changed

‎src/providers/pulse.explorer.ts‎

Lines changed: 37 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,14 @@
1-
import { Injectable, Logger, OnModuleInit } from "@nestjs/common";
2-
import { DiscoveryService, MetadataScanner } from "@nestjs/core";
3-
import { InstanceWrapper } from "@nestjs/core/injector/instance-wrapper";
4-
import { JobAttributes, Processor } from "@pulsecron/pulse";
5-
import { getQueueConfigToken, getQueueToken } from "../utils";
6-
import { PulseMetadataAccessor } from "./pulse-metadata.accessor";
7-
import { PulseOrchestrator } from "./pulse.orchestrator";
1+
import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
2+
import { DiscoveryService, MetadataScanner } from '@nestjs/core';
3+
import { InstanceWrapper } from '@nestjs/core/injector/instance-wrapper';
4+
import { JobAttributes, JobAttributesData, Processor } from '@pulsecron/pulse';
5+
import { getQueueConfigToken, getQueueToken } from '../utils';
6+
import { PulseMetadataAccessor } from './pulse-metadata.accessor';
7+
import { PulseOrchestrator } from './pulse.orchestrator';
88

99
@Injectable()
1010
export class PulseExplorer implements OnModuleInit {
11-
private readonly logger = new Logger("Pulse");
11+
private readonly logger = new Logger('Pulse');
1212

1313
constructor(
1414
private readonly discoveryService: DiscoveryService,
@@ -26,70 +26,48 @@ export class PulseExplorer implements OnModuleInit {
2626
.getProviders()
2727
.filter((wrapper: InstanceWrapper) => {
2828
return this.metadataAccessor.isQueue(
29-
!wrapper.metatype || wrapper.inject
30-
? wrapper?.constructor
31-
: wrapper.metatype
29+
!wrapper.metatype || wrapper.inject ? wrapper?.constructor : wrapper.metatype
3230
);
3331
})
3432
.forEach((wrapper: InstanceWrapper) => {
3533
const { instance, metatype } = wrapper;
3634

37-
const { queueName } = this.metadataAccessor.getQueueMetadata(
38-
instance.constructor || metatype
39-
);
35+
const { queueName } = this.metadataAccessor.getQueueMetadata(instance.constructor || metatype);
4036

4137
const queueToken = getQueueToken(queueName);
4238

4339
const queueConfigToken = getQueueConfigToken(queueName);
4440

4541
this.orchestrator.addQueue(queueName, queueToken, queueConfigToken);
4642

47-
this.metadataScanner.scanFromPrototype(
48-
instance,
49-
Object.getPrototypeOf(instance),
50-
(key: string) => {
51-
const methodRef = instance[key];
52-
53-
if (this.metadataAccessor.isJobProcessor(methodRef)) {
54-
const jobProcessorType =
55-
this.metadataAccessor.getJobProcessorType(methodRef);
56-
57-
const jobOptions =
58-
this.metadataAccessor.getJobProcessorMetadata(methodRef);
59-
60-
const jobProcessor: Processor<JobAttributes> &
61-
Record<"_name", string> = this.wrapFunctionInTryCatchBlocks(
62-
methodRef,
63-
instance
64-
);
65-
66-
this.orchestrator.addJobProcessor(
67-
queueToken,
68-
jobProcessor,
69-
jobOptions,
70-
jobProcessorType,
71-
methodRef.length === 2
72-
);
73-
} else if (this.metadataAccessor.isEventListener(methodRef)) {
74-
const listener = this.wrapFunctionInTryCatchBlocks(
75-
methodRef,
76-
instance
77-
);
78-
79-
const eventName =
80-
this.metadataAccessor.getListenerMetadata(methodRef);
81-
82-
const jobName = this.metadataAccessor.getJobName(methodRef);
83-
84-
return this.orchestrator.addEventListener(
85-
queueToken,
86-
listener,
87-
eventName,
88-
jobName
89-
);
90-
}
43+
this.metadataScanner.scanFromPrototype(instance, Object.getPrototypeOf(instance), (key: string) => {
44+
const methodRef = instance[key];
45+
46+
if (this.metadataAccessor.isJobProcessor(methodRef)) {
47+
const jobProcessorType = this.metadataAccessor.getJobProcessorType(methodRef);
48+
49+
const jobOptions = this.metadataAccessor.getJobProcessorMetadata(methodRef);
50+
51+
const jobProcessor: Processor<JobAttributesData> & Record<'_name', string> =
52+
this.wrapFunctionInTryCatchBlocks(methodRef, instance);
53+
54+
this.orchestrator.addJobProcessor(
55+
queueToken,
56+
jobProcessor,
57+
jobOptions,
58+
jobProcessorType,
59+
methodRef.length === 2
60+
);
61+
} else if (this.metadataAccessor.isEventListener(methodRef)) {
62+
const listener = this.wrapFunctionInTryCatchBlocks(methodRef, instance);
63+
64+
const eventName = this.metadataAccessor.getListenerMetadata(methodRef);
65+
66+
const jobName = this.metadataAccessor.getJobName(methodRef);
67+
68+
return this.orchestrator.addEventListener(queueToken, listener, eventName, jobName);
9169
}
92-
);
70+
});
9371
});
9472
}
9573

‎src/providers/pulse.orchestrator.ts‎

Lines changed: 28 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,21 @@
1-
import {
2-
BeforeApplicationShutdown,
3-
Injectable,
4-
Logger,
5-
OnApplicationBootstrap,
6-
} from "@nestjs/common";
7-
import { ModuleRef } from "@nestjs/core";
1+
import { BeforeApplicationShutdown, Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common';
2+
import { ModuleRef } from '@nestjs/core';
83
import Pulse, {
94
PulseConfig,
105
Job,
116
Processor,
127
JobAttributes,
13-
} from "@pulsecron/pulse";
14-
import { NO_QUEUE_FOUND } from "../pulse.messages";
15-
import {
16-
PulseModuleJobOptions,
17-
NonRepeatableJobOptions,
18-
RepeatableJobOptions,
19-
} from "../decorators";
20-
import { JobProcessorType } from "../enums";
21-
import { PulseQueueConfig } from "../interfaces";
22-
import { DatabaseService } from "./database.service";
8+
PulseOnEventType,
9+
JobAttributesData,
10+
} from '@pulsecron/pulse';
11+
import { NO_QUEUE_FOUND } from '../pulse.messages';
12+
import { PulseModuleJobOptions, NonRepeatableJobOptions, RepeatableJobOptions } from '../decorators';
13+
import { JobProcessorType } from '../enums';
14+
import { PulseQueueConfig } from '../interfaces';
15+
import { DatabaseService } from './database.service';
2316

2417
type JobProcessorConfig = {
25-
handler: Processor<JobAttributes>;
18+
handler: Processor<JobAttributesData>;
2619
type: JobProcessorType;
2720
options: RepeatableJobOptions | NonRepeatableJobOptions;
2821
useCallback: boolean;
@@ -38,17 +31,12 @@ type QueueRegistry = {
3831
};
3932

4033
@Injectable()
41-
export class PulseOrchestrator
42-
implements OnApplicationBootstrap, BeforeApplicationShutdown
43-
{
44-
private readonly logger = new Logger("Pulse");
34+
export class PulseOrchestrator implements OnApplicationBootstrap, BeforeApplicationShutdown {
35+
private readonly logger = new Logger('Pulse');
4536

4637
private readonly queues: Map<string, QueueRegistry> = new Map();
4738

48-
constructor(
49-
private readonly moduleRef: ModuleRef,
50-
private readonly database: DatabaseService
51-
) {}
39+
constructor(private readonly moduleRef: ModuleRef, private readonly database: DatabaseService) {}
5240

5341
async onApplicationBootstrap() {
5442
await this.database.connect();
@@ -60,10 +48,7 @@ export class PulseOrchestrator
6048

6149
this.attachEventListeners(queue, registry);
6250

63-
queue.mongo(
64-
this.database.getConnection(),
65-
config.collection || queueToken
66-
);
51+
queue.mongo(this.database.getConnection(), config.collection || queueToken);
6752

6853
if (config.autoStart) {
6954
await queue.start();
@@ -99,7 +84,7 @@ export class PulseOrchestrator
9984

10085
addJobProcessor(
10186
queueToken: string,
102-
processor: Processor<JobAttributes> & Record<"_name", string>,
87+
processor: Processor<JobAttributesData> & Record<'_name', string>,
10388
options: PulseModuleJobOptions,
10489
type: JobProcessorType,
10590
useCallback: boolean
@@ -114,39 +99,28 @@ export class PulseOrchestrator
11499
});
115100
}
116101

117-
addEventListener(
118-
queueToken: string,
119-
listener: EventListener,
120-
eventName: string,
121-
jobName?: string
122-
) {
102+
addEventListener(queueToken: string, listener: EventListener, eventName: PulseOnEventType, jobName?: string) {
123103
const key = jobName ? `${eventName}:${jobName}` : eventName;
124104

125105
this.queues.get(queueToken)?.listeners.set(key, listener);
126106
}
127107

128108
private attachEventListeners(pulse: Pulse, registry: QueueRegistry) {
129109
registry.listeners.forEach((listener: EventListener, eventName: string) => {
130-
pulse.on(eventName, listener);
110+
pulse.on(eventName as PulseOnEventType, listener);
131111
});
132112
}
133113

134114
private defineJobProcessors(pulse: Pulse, registry: QueueRegistry) {
135-
registry.processors.forEach(
136-
(jobConfig: JobProcessorConfig, jobName: string) => {
137-
const { options, handler, useCallback } = jobConfig;
138-
139-
if (useCallback) {
140-
pulse.define(
141-
jobName,
142-
(job: Job, done?: () => void) => handler(job, done),
143-
options
144-
);
145-
} else {
146-
pulse.define(jobName, handler, options);
147-
}
115+
registry.processors.forEach((jobConfig: JobProcessorConfig, jobName: string) => {
116+
const { options, handler, useCallback } = jobConfig;
117+
118+
if (useCallback) {
119+
pulse.define(jobName, (job: Job, done: () => void = () => {}) => handler(job, done), options);
120+
} else {
121+
pulse.define(jobName, handler, options);
148122
}
149-
);
123+
});
150124
}
151125

152126
private async scheduleJobs(pulse: Pulse, registry: QueueRegistry) {
@@ -156,18 +130,9 @@ export class PulseOrchestrator
156130
const { type, options } = jobConfig;
157131

158132
if (type === JobProcessorType.EVERY) {
159-
await pulse.every(
160-
(options as RepeatableJobOptions).interval,
161-
jobName,
162-
{},
163-
options
164-
);
133+
await pulse.every((options as RepeatableJobOptions).interval, jobName, {}, options);
165134
} else if (type === JobProcessorType.SCHEDULE) {
166-
await pulse.schedule(
167-
(options as NonRepeatableJobOptions).when,
168-
jobName,
169-
{}
170-
);
135+
await pulse.schedule((options as NonRepeatableJobOptions).when, jobName, {});
171136
} else if (type === JobProcessorType.NOW) {
172137
await pulse.now(jobName, {});
173138
}

0 commit comments

Comments
 (0)