Worker & Job Queue

import { bootstrap, JobQueueService } from '@vendure/core';import { config } from './vendure-config';bootstrap(config)    .then(app => app.get(JobQueueService).start())    .catch(err => {        console.log(err);        process.exit(1);    });

import { Injectable, OnApplicationBootstrap } from '@nestjs/common';import { ProcessContext } from '@vendure/core';@Injectable()export class MyService implements OnApplicationBootstrap {    constructor(private processContext: ProcessContext) {}    onApplicationBootstrap() {        if (this.processContext.isServer) {            // code which will only execute when running in            // the server process        }    }}

import gql from 'graphql-tag';export const adminApiExtensions = gql`  extend type Mutation {    addVideoToProduct(productId: ID! videoUrl: String!): Job!  }`;

import { Args, Mutation, Resolver } from '@nestjs/graphql';import { Allow, Ctx, RequestContext, Permission, RequestContext } from '@vendure/core'import { ProductVideoService } from '../services/product-video.service';@Resolver()export class ProductVideoResolver {    constructor(private productVideoService: ProductVideoService) {}    @Mutation()    @Allow(Permission.UpdateProduct)    addVideoToProduct(@Ctx() ctx: RequestContext, @Args() args: { productId: ID; videoUrl: string; }) {        return this.productVideoService.transcodeForProduct(            args.productId,            args.videoUrl,        );    }}

import { Injectable, OnModuleInit } from '@nestjs/common';import { JobQueue, JobQueueService, ID, Product, TransactionalConnection } from '@vendure/core';import { transcode } from 'third-party-video-sdk';@Injectable()class ProductVideoService implements OnModuleInit {    private jobQueue: JobQueue<{ productId: ID; videoUrl: string; }>;    constructor(private jobQueueService: JobQueueService,                private connection: TransactionalConnection) {    }    async onModuleInit() {        this.jobQueue = await this.jobQueueService.createQueue({            name: 'transcode-video',            process: async job => {                // Inside the `process` function we define how each job                // in the queue will be processed.                // In this case we call out to some imaginary 3rd-party video                // transcoding API, which performs the work and then                // returns a new URL of the transcoded video, which we can then                // associate with the Product via the customFields.                const result = await transcode(job.data.videoUrl);                await this.connection.getRepository(Product).save({                    id: job.data.productId,                    customFields: {                        videoUrl: result.url,                    },                });                // The value returned from the `process` function is stored as the "result"                // field of the job (for those JobQueueStrategies that support recording of results).                //                // Any error thrown from this function will cause the job to fail.                return result;            },        });    }    transcodeForProduct(productId: ID, videoUrl: string) {        // Add a new job to the queue and immediately return the        // job itself.        return this.jobQueue.add({productId, videoUrl}, {retries: 2});    }}

JobQueue<{ productId: ID; videoUrl: string; }>

productVideoService.transcodeForProduct(id, url);

import gql from 'graphql-tag';import { PluginCommonModule, VendurePlugin } from '@vendure/core';import { ProductVideoService } from './services/product-video.service';import { ProductVideoResolver } from './api/product-video.resolver';import { adminApiExtensions } from './api/api-extensions';@VendurePlugin({    imports: [PluginCommonModule],    providers: [ProductVideoService],    adminApiExtensions: {        schema: adminApiExtensions,        resolvers: [ProductVideoResolver]    },    configuration: config => {        config.customFields.Product.push({            name: 'videoUrl',            type: 'string',        });        return config;    }})export class ProductVideoPlugin {}

import { Injectable, OnModuleInit } from '@nestjs/common';import { JobQueue, JobQueueService, Product, TransactionalConnection,    SerializedRequestContext, RequestContext } from '@vendure/core';@Injectable()class ProductExportService implements OnModuleInit {    private jobQueue: JobQueue<{ ctx: SerializedRequestContext; }>; // [!code highlight]    constructor(private jobQueueService: JobQueueService,                private connection: TransactionalConnection) {    }    async onModuleInit() {        this.jobQueue = await this.jobQueueService.createQueue({            name: 'export-products',            process: async job => {                const ctx = RequestContext.deserialize(job.data.ctx); // [!code highlight]                const allProducts = await this.connection.getRepository(ctx, Product).find();                // ... logic to export the product omitted for brevity            },        });    }    exportAllProducts(ctx: RequestContext) {        return this.jobQueue.add({ ctx: ctx.serialize() }); // [!code highlight]    }}

exportAllProducts(ctx: RequestContext) {    return this.jobQueue.add({        ctx: ctx.serialize(),        hostname: ctx.req?.get('host'), // [!code highlight]    });}

import { Injectable, OnModuleInit } from '@nestjs/common';import { JobQueue, JobQueueService,    RequestContext, ID, LanguageCode, RequestContextService } from '@vendure/core';@Injectable()class ProductExportService implements OnModuleInit {    private jobQueue: JobQueue<{ channelToken: string; languageCode: LanguageCode; }>; // [!code highlight]    constructor(private jobQueueService: JobQueueService,                private requestContextService: RequestContextService) {    }    async onModuleInit() {        this.jobQueue = await this.jobQueueService.createQueue({            name: 'export-products',            process: async job => {                // Reconstruct the RequestContext from the passed data // [!code highlight]                const ctx = await this.requestContextService.create({ // [!code highlight]                    channelOrToken: job.data.channelToken, // [!code highlight]                    languageCode: job.data.languageCode, // [!code highlight]                }) // [!code highlight]                // ... logic to export the product omitted for brevity            },        });    }    exportAllProducts(ctx: RequestContext) {        // Pass only the necessary data // [!code highlight]        return this.jobQueue.add({ // [!code highlight]            channelId: ctx.channel.token, // [!code highlight]            languageCode: ctx.languageCode // [!code highlight]        }); // [!code highlight]    }}

import { Injectable, OnModuleInit } from '@nestjs/common';import { JobQueue, JobQueueService, Product, TransactionalConnection,    SerializedRequestContext, RequestContext } from '@vendure/core';import { JobState } from '@vendure/common/lib/generated-types';import { IsNull } from 'typeorm';@Injectable()class ProductExportService implements OnModuleInit {    private jobQueue: JobQueue<{ ctx: SerializedRequestContext; }>;    constructor(private jobQueueService: JobQueueService,                private connection: TransactionalConnection) {    }    async onModuleInit() {        this.jobQueue = await this.jobQueueService.createQueue({            name: 'export-products',            process: async job => {                const ctx = RequestContext.deserialize(job.data.ctx);                const allProducts = await this.connection.getRepository(ctx, Product).find({                    where: { deletedAt: IsNull() }                });                let successfulExportCount = 0;                for (const product of allProducts) {                    if (job.state === JobState.CANCELLED) { // [!code highlight]                        // If the job has been cancelled, stop processing // [!code highlight]                        // to prevent unnecessary work. // [!code highlight]                        throw new Error('Job was cancelled'); // [!code highlight]                    } // [!code highlight]                    // ... logic to export the product omitted for brevity                    successfulExportCount++;                }                return { successfulExportCount };            },        });    }    exportAllProducts(ctx: RequestContext) {        return this.jobQueue.add({ ctx: ctx.serialize() });    }}

import { Injectable, OnModuleInit } from '@nestjs/common';import { of } from 'rxjs';import { map, catchError } from 'rxjs/operators';import { ID, Product, TransactionalConnection } from '@vendure/core';@Injectable()class ProductVideoService implements OnModuleInit {    // ... omitted (see above)    transcodeForProduct(productId: ID, videoUrl: string) {        const job = await this.jobQueue.add({productId, videoUrl}, {retries: 2});        return job.updates().pipe(            map(update => {                // The returned Observable will emit a value for every update to the job                // such as when the `progress` or `status` value changes.                Logger.info(`Job ${update.id}: progress: ${update.progress}`);                if (update.state === JobState.COMPLETED) {                    Logger.info(`COMPLETED ${update.id}: ${update.result}`);                }                return update.result;            }),            catchError(err => of(err.message)),        );    }}

const job = await this.jobQueue.add({ productId, videoUrl }, { retries: 2 });return job.updates().toPromise()  .then(/* ... */)  .catch(/* ... */);