Browse Source

Extend queue service

pull/3725/head
Thomas Kaul 12 months ago
parent
commit
9d65030e06
  1. 3
      apps/api/src/app/admin/queue/queue.module.ts
  2. 38
      apps/api/src/app/admin/queue/queue.service.ts
  3. 2
      apps/client/src/app/components/admin-jobs/admin-jobs.html

3
apps/api/src/app/admin/queue/queue.module.ts

@ -1,4 +1,5 @@
import { DataGatheringModule } from '@ghostfolio/api/services/data-gathering/data-gathering.module';
import { PortfolioSnapshotQueueModule } from '@ghostfolio/api/services/portfolio-snapshot/portfolio-snapshot.module';
import { Module } from '@nestjs/common';
@ -7,7 +8,7 @@ import { QueueService } from './queue.service';
@Module({
controllers: [QueueController],
imports: [DataGatheringModule],
imports: [DataGatheringModule, PortfolioSnapshotQueueModule],
providers: [QueueService]
})
export class QueueModule {}

38
apps/api/src/app/admin/queue/queue.service.ts

@ -1,5 +1,6 @@
import {
DATA_GATHERING_QUEUE,
PORTFOLIO_SNAPSHOT_QUEUE,
QUEUE_JOB_STATUS_LIST
} from '@ghostfolio/common/config';
import { AdminJobs } from '@ghostfolio/common/interfaces';
@ -12,11 +13,19 @@ import { JobStatus, Queue } from 'bull';
export class QueueService {
public constructor(
@InjectQueue(DATA_GATHERING_QUEUE)
private readonly dataGatheringQueue: Queue
private readonly dataGatheringQueue: Queue,
@InjectQueue(PORTFOLIO_SNAPSHOT_QUEUE)
private readonly portfolioSnapshotQueue: Queue
) {}
public async deleteJob(aId: string) {
return (await this.dataGatheringQueue.getJob(aId))?.remove();
let job = await this.dataGatheringQueue.getJob(aId);
if (!job) {
job = await this.portfolioSnapshotQueue.getJob(aId);
}
return job?.remove();
}
public async deleteJobs({
@ -25,15 +34,21 @@ export class QueueService {
status?: JobStatus[];
}) {
for (const statusItem of status) {
await this.dataGatheringQueue.clean(
300,
statusItem === 'waiting' ? 'wait' : statusItem
);
const queueStatus = statusItem === 'waiting' ? 'wait' : statusItem;
await this.dataGatheringQueue.clean(300, queueStatus);
await this.portfolioSnapshotQueue.clean(300, queueStatus);
}
}
public async executeJob(aId: string) {
return (await this.dataGatheringQueue.getJob(aId))?.promote();
let job = await this.dataGatheringQueue.getJob(aId);
if (!job) {
job = await this.portfolioSnapshotQueue.getJob(aId);
}
return job?.promote();
}
public async getJobs({
@ -43,10 +58,15 @@ export class QueueService {
limit?: number;
status?: JobStatus[];
}): Promise<AdminJobs> {
const jobs = await this.dataGatheringQueue.getJobs(status);
const [dataGatheringJobs, portfolioSnapshotJobs] = await Promise.all([
this.dataGatheringQueue.getJobs(status),
this.portfolioSnapshotQueue.getJobs(status)
]);
const allJobs = [...dataGatheringJobs, ...portfolioSnapshotJobs];
const jobsWithState = await Promise.all(
jobs
allJobs
.filter((job) => {
return job;
})

2
apps/client/src/app/components/admin-jobs/admin-jobs.html

@ -35,6 +35,8 @@
<ng-container i18n>Asset Profile</ng-container>
} @else if (element.name === 'GATHER_HISTORICAL_MARKET_DATA') {
<ng-container i18n>Historical Market Data</ng-container>
} @else if (element.name === 'PORTFOLIO') {
<ng-container i18n>Portfolio Snapshot</ng-container>
}
</td>
</ng-container>

Loading…
Cancel
Save