import * as amqplib from "amqplib" ; import { CeleryBackend } from "." ; export default class AMQPBackend implements CeleryBackend { opts : { [ key ] : any ; } ; connect : Promise < amqplib . Connection > ; channel : Promise < amqplib . Channel > ; constructor param { string } url param { object } opts constructor ( url , opts ) { this . opts = opts ; this . connect = amqplib . connect ( url , opts ) ; this . channel = this . connect . then ( conn => conn . createChannel ( ) ) . then ( ch => ch . assertExchange ( "default" , "direct" , { durable : true , autoDelete : true , internal : false , arguments : null } ) . then ( ( ) => Promise . resolve ( ch ) ) ) ; } method returns { Promise } public isReady ( ) { return this . connect ; } method returns { Promise } public disconnect ( ) { return this . connect . then ( conn => conn . close ( ) ) ; } method param { String } taskId param { any } result param { String } state returns { Promise } public storeResult ( taskId , result , state ) { const queue = taskId . replace ( /-/g , "" ) ; return this . channel . then ( ch => ch . assertQueue ( queue , { durable : true , autoDelete : true , exclusive : false , arguments : { "x-expires" : this . opts . CELERY_RESULT_EXPIRES || 86400000 } } ) . then ( ( ) => Promise . resolve ( ch ) ) ) . then ( ch => ch . publish ( "" , queue , Buffer . from ( JSON . stringify ( { status : state , result : state == 'FAILURE' ? null : result , traceback : null , children : [ ] , task_id : taskId , date_done : new Date ( ) . toISOString ( ) } ) ) , { contentType : "application/json" , contentEncoding : "utf-8" } ) ) ; } method param { String } taskId returns { Promise } public getTaskMeta ( taskId ) { const queue = taskId . replace ( /-/g , "" ) ; return this . channel . then ( ch => ch . assertQueue ( queue , { durable : true , autoDelete : true , exclusive : false , arguments : { "x-expires" : this . opts . CELERY_RESULT_EXPIRES || 86400000 } } ) . then ( ( ) => Promise . resolve ( ch ) ) ) . then ( ch => ch . get ( queue , { noAck : false } ) ) . then ( msg => { if ( msg === false ) { return null ; } if ( msg . properties . contentType !== "application/json" ) { throw new Error ( `unsupported content type ${ msg . properties . contentType }` ) ; } if ( msg . properties . contentEncoding !== "utf-8" ) { throw new Error ( `unsupported content encoding ${ msg . properties . contentEncoding }` ) ; } const body = msg . content . toString ( "utf-8" ) ; return JSON . parse ( body ) ; } ) ; } }