import { AmqpOptions , DEFAULT_AMQP_OPTIONS } from "./options" ; import { PromiseMap , ResourcePool } from "../containers" ; import { UnimplementedError } from "../errors" ; import { ResultMessage } from "../messages" ; import { GetOptions , ResultBackend } from "../result_backend" ; import { createTimeoutPromise , isNullOrUndefined , promisifyEvent , } from "../utility" ; import * as AmqpLib from "amqplib" ; /**
 * RabbitMQ result backend using RPC and one queue per client.
 */ export class RpcBackend implements ResultBackend { private readonly channels : ResourcePool < AmqpLib . Channel > ; private readonly connection : Promise < AmqpLib . Connection > ; private readonly consumer : Promise < AmqpLib . Channel > ; private readonly consumerTag : Promise < string > ; private readonly options : AmqpOptions ; private promises : PromiseMap < string , Message > ; private readonly routingKey : string ; param routingKey param options public constructor ( routingKey , options ? ) { const DEFAULT_TIMEOUT = 86400000 ; this . options = ( ( ) => { if ( isNullOrUndefined ( options ) ) { return DEFAULT_AMQP_OPTIONS ; } return options ; } ) ( ) ; this . promises = new PromiseMap < string , Message > ( DEFAULT_TIMEOUT ) ; this . routingKey = routingKey ; this . connection = Promise . resolve ( AmqpLib . connect ( this . options ) ) ; this . channels = new ResourcePool < AmqpLib . Channel > ( ( ) => this . connection . then ( ( connection ) => connection . createChannel ( ) ) , ( channel ) => Promise . resolve ( channel . close ( ) ) . then ( ( ) => "closed" ) , 2 ) ; this . consumer = this . channels . get ( ) ; this . consumerTag = this . consumer . then ( ( consumer ) => this . assertQueue ( consumer ) . then ( ( ) => this . createConsumer ( consumer ) ) ) ; } param message returns public async put < T > ( message ) { const toSend = Buffer . from ( JSON . stringify ( message ) , "utf8" ) ; const options = RpcBackend . createPublishOptions ( message ) ; return this . channels . use ( async ( channel ) => { await this . assertQueue ( channel ) ; return this . sendToQueue ( { channel , options , toSend } ) ; } ) ; } param taskId param timeout returns public async get < T > ( { taskId , timeout } ) { const raw = await this . promises . get ( taskId ) ; const result = RpcBackend . parseMessage < T > ( raw ) ; return createTimeoutPromise ( result , timeout ) ; } param taskId returns public async delete ( taskId ) { if ( this . promises . delete ( taskId ) ) { return "deleted" ; } return "no result found" ; } returns see public async disconnect ( ) { await this . end ( ) ; } returns public async end ( ) { const consumer = await this . consumer ; const consumerTag = await this . consumerTag ; this . promises . rejectAll ( new Error ( "disconnecting" ) ) ; await consumer . cancel ( consumerTag ) ; this . channels . return ( consumer ) ; await this . channels . destroyAll ( ) ; const connection = await this . connection ; await connection . close ( ) ; } returns throws public uri ( ) { throw new UnimplementedError ( "Celery.Amqp.Backend.RpcBackend.uri" ) ; } param message returns throws throws private static parseMessage < T > ( message ) { const content = message . content . toString ( "utf8" ) ; const parsed = JSON . parse ( content ) ; return parsed ; } param message returns private static createPublishOptions < T > ( message ) { return { contentEncoding : "utf-8" , contentType : "application/json" , correlationId : message . task_id , persistent : false , priority : 0 , } ; } param channel returns private async assertQueue ( channel ) { return channel . assertQueue ( this . routingKey , { autoDelete : false , durable : false , expires : 86400000 , } ) ; } param channel param options param toSend private async sendToQueue ( { channel , options , toSend } ) { const send = ( ) => channel . sendToQueue ( this . routingKey , toSend , options ) ; while ( ! send ( ) ) { await promisifyEvent < void > ( channel , "drain" ) ; } return "flushed to write buffer" ; } param consumer returns private async createConsumer ( consumer ) { const reply = await consumer . consume ( this . routingKey , ( message ) => this . onMessage ( message ) , { noAck : true } ) ; return reply . consumerTag ; } param maybeMessage private onMessage ( maybeMessage ? ) { if ( isNullOrUndefined ( maybeMessage ) ) { this . promises . rejectAll ( new Error ( "RabbitMQ cancelled consumer" ) ) ; return ; } const message = maybeMessage ; const id = message . properties . correlationId ; this . promises . resolve ( id , message ) ; } } /**
 * Layout of a RabbitMQ message.
 */ interface Message { content : Buffer ; fields : object ; properties : { contentEncoding : string ; contentType : string ; correlationId : string ; headers : object ; priority : number ; } ; }