Worker i kolejka zadań

import { bootstrapWorker } from '@deenruv/core';
import { config } from './deenruv-config';

bootstrapWorker(config)
    .then(worker => worker.startJobQueue())
    .catch(err => {
        console.log(err);
    });

import { bootstrapWorker, mergeConfig } from '@deenruv/core';
import { config } from './deenruv-config';

const transcoderConfig = mergeConfig(config, {
    jobQueueOptions: {
        activeQueues: ['transcode-video'],
    },
});

bootstrapWorker(transcoderConfig)
    .then(worker => worker.startJobQueue())
    .catch(err => {
        console.log(err);
    });

import { bootstrap, JobQueueService } from '@deenruv/core';
import { config } from './deenruv-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 '@deenruv/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 '@deenruv/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 '@deenruv/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 '@deenruv/core';
import { ProductVideoService } from './services/product-video.service';
import { ProductVideoResolver } from './api/product-video.resolver';
import { adminApiExtensions } from './api/api-extensions';

@DeenruvPlugin({
    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 '@deenruv/core';

@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();
                // ... logic to export the product omitted for brevity
            },
        });
    }

    exportAllProducts(ctx: RequestContext) {
        return this.jobQueue.add({ ctx: RequestContext.serialize(ctx) });
    }
}

import { Injectable, OnModuleInit } from '@nestjs/common';
import {
    JobQueue,
    JobQueueService,
    Product,
    TransactionalConnection,
    SerializedRequestContext,
    RequestContext,
    Job,
    JobState,
} from '@deenruv/core';
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) {
                        // If the job has been cancelled, stop processing
                        // to prevent unnecessary work.
                        throw new Error('Job was cancelled');
                    }

                    // ... logic to export the product omitted for brevity
                    successfulExportCount++;
                }
                return { successfulExportCount };
            },
        });
    }

    exportAllProducts(ctx: RequestContext) {
        return this.jobQueue.add({ ctx: RequestContext.serialize(ctx) });
    }
}

import { Injectable, OnModuleInit } from '@nestjs/common';
import { of } from 'rxjs';
import { map, catchError } from 'rxjs/operators';
import { ID, Product, TransactionalConnection } from '@deenruv/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(/* ... */);