import { FailoverStrategy } from "./config" ; import { MessageBroker } from "./message_broker" ; import { CompressionMime , ContentEncodingMime , ContentTypeMime , TaskHeaders , TaskProperties , } from "./messages" ; import { NullBackend } from "./null_backend" ; import * as Packer from "./packer" ; import { Result } from "./result" ; import { ResultBackend } from "./result_backend" ; import { isNullOrUndefined } from "./utility" ; import * as Os from "os" ; import * as Process from "process" ; import * as Uuid from "uuid" ; see Client export class Task < T > { private readonly appId : string ; private readonly backend : ResultBackend ; private broker : MessageBroker ; private readonly brokers : Array < MessageBroker > ; private readonly deliveryMode : "persistent" | "transient" = "persistent" ; private readonly failoverStrategy : FailoverStrategy ; private readonly name : string ; private readonly queue : string = "celery" ; private readonly timeLimit : [ number | null , number | null ] = [ null , null ] ; param appId param backend param brokers param deliveryMode param failoverStrategy param hardTimeLimit param name param queue param softTimeLimit returns public constructor ( { appId , backend , brokers , deliveryMode = "persistent" , failoverStrategy , hardTimeLimit , name , queue = "celery" , softTimeLimit , } ) { this . appId = appId ; this . backend = backend ; this . broker = failoverStrategy ( brokers ) ; this . brokers = brokers ; this . deliveryMode = deliveryMode ; this . failoverStrategy = failoverStrategy ; this . name = name ; this . queue = queue ; this . timeLimit = Task . getTimeLimit ( { hard : hardTimeLimit , soft : softTimeLimit , } ) ; } param args param compression param eta param expires param ignoreResult param kwargs param priority param queue param serializer returns public applyAsync ( { args , compression = Packer . Compressor . Identity , eta , expires , ignoreResult = false , kwargs , priority = 0 , queue = this . queue , serializer = Packer . Serializer . Json , } ) { const backend = ( ( ) => { if ( ignoreResult ) { return new NullBackend ( ) ; } return this . backend ; } ) ( ) ; const id = Uuid . v4 ( ) ; const result = new Result < T > ( id , backend ) ; const [ packer , encoding ] = Task . createPacker ( serializer , compression ) ; const body = Task . packBody ( { args , kwargs , packer } ) ; const etaStr = Task . dateOrNull ( eta ) ; const expiresStr = Task . dateOrNull ( expires ) ; const publishOptions = { body , "content-encoding" : ContentEncodingMime . Utf8 , "content-type" : Task . getContentTypeMime ( serializer ) , headers : this . createHeaders ( { args , compression , eta : etaStr , expires : expiresStr , id , kwargs , } ) , properties : this . createProperties ( { deliveryMode : this . getDeliveryMode ( ) , encoding , id , priority , queue , } ) , } ; const tryPublish = async ( ) : Promise < string > => { try { return this . broker . publish ( publishOptions ) ; } catch { this . broker = this . failoverStrategy ( this . brokers ) ; return tryPublish ( ) ; } } ; tryPublish ( ) ; return result ; } param date returns private static dateOrNull ( date ? ) { if ( isNullOrUndefined ( date ) ) { return null ; } return date . toISOString ( ) ; } param soft param hard returns private static getTimeLimit ( { soft , hard } ) { if ( ! isNullOrUndefined ( soft ) && ! isNullOrUndefined ( hard ) ) { return [ soft , hard ] ; } else if ( ! isNullOrUndefined ( soft ) ) { return [ soft , null ] ; } else if ( ! isNullOrUndefined ( hard ) ) { return [ null , hard ] ; } return [ null , null ] ; } returns private getDeliveryMode ( ) { if ( this . deliveryMode === "transient" ) { return 1 ; } return 2 ; } param serializer param compressor returns private static createPacker ( serializer , compressor ) { const encoder = Task . selectEncoder ( compressor ) ; const realCompressor = ( ( ) => { if ( compressor === Packer . Compressor . Gzip ) { return Packer . Compressor . Zlib ; } return compressor ; } ) ( ) ; return [ Packer . createPacker ( { compressor : realCompressor , encoder , serializer , } ) , encoder ] ; } param args param kwargs param packer returns private static packBody ( { args , kwargs , packer } ) { return packer . pack ( [ args , kwargs , { callbacks : null , chain : null , chord : null , errbacks : null } , ] ) ; } param serializer returns private static getContentTypeMime ( serializer ) { switch ( serializer ) { case Packer . Serializer . Json : return ContentTypeMime . Json ; case Packer . Serializer . Yaml : return ContentTypeMime . Yaml ; } } returns private static getOrigin ( ) { return `${ Process . pid }@${ Os . hostname ( ) }` ; } param args param compression param eta param expires param kwargs param id returns private createHeaders ( { args , compression , eta , expires , kwargs , id } ) { const base = { argsrepr : JSON . stringify ( args ) , eta , expires , group : null , id , kwargsrepr : JSON . stringify ( kwargs ) , lang : "py" , origin : Task . getOrigin ( ) , parent_id : null , retries : 0 , root_id : id , shadow : null , task : this . name , timelimit : this . timeLimit , } ; if ( compression === Packer . Compressor . Identity ) { return base ; } return { ... base , compression : CompressionMime . Zlib , } ; } param deliveryMode param encoding param id param priority param queue returns private createProperties ( { deliveryMode , encoding , id , priority , queue , } ) { return { body_encoding : Task . getEncodingMime ( encoding ) , correlation_id : id , delivery_info : { exchange : "" , routing_key : queue , } , delivery_mode : deliveryMode , delivery_tag : "celery" , priority , reply_to : this . appId , } ; } param compressor returns private static selectEncoder ( compressor ) { if ( compressor === Packer . Compressor . Identity ) { return Packer . Encoder . Plaintext ; } return Packer . Encoder . Base64 ; } param encoding returns private static getEncodingMime ( encoding ) { switch ( encoding ) { case Packer . Encoder . Base64 : return ContentEncodingMime . Base64 ; case Packer . Encoder . Plaintext : return ContentEncodingMime . Utf8 ; } } } /**
 * Task creation options.
 */ export interface TaskOptions { appId : string ; backend : ResultBackend ; brokers : Array < MessageBroker > ; deliveryMode ? : "persistent" | "transient" ; failoverStrategy : FailoverStrategy ; hardTimeLimit ? : number ; name : string ; queue ? : string ; softTimeLimit ? : number ; } /**
 * Task invocation options.
 */ export interface TaskApplyOptions { args : Args ; compression ? : Packer . Compressor ; eta ? : Date ; expires ? : Date ; ignoreResult ? : boolean ; kwargs : KeywordArgs ; priority ? : Priority ; queue ? : string ; serializer ? : Packer . Serializer ; } export type Args = Array < any > ; export interface KeywordArgs { [ key ] : any | undefined ; } export type Priority = number ;