import { DEFAULT_REDIS_OPTIONS , RedisOptions } from "./options" ; import { PromiseMap , ResourcePool } from "../containers" ; import { ResultMessage , Status } from "../messages" ; import { GetOptions , ResultBackend } from "../result_backend" ; import { createTimeoutPromise , isNullOrUndefined } from "../utility" ; import * as IoRedis from "ioredis" ; /**
 * Redis in-memory database result backend.
 * Uses connection pooling with one connection per active subscriber.
 */ export class RedisBackend implements ResultBackend { private static readonly TIMEOUT : number = 86400000 ; private static readonly UUID_REGEX : RegExp = /^celery-task-meta-([A-Fa-f\d]{8}-[A-Fa-f\d]{4}-[A-Fa-f\d]{4}-[A-Fa-f\d]{4}-[A-Fa-f\d]{12})$/ ; private static readonly UUID_INDEX : number = 1 ; private readonly options : RedisOptions ; private readonly pool : ResourcePool < IoRedis . Redis > ; private readonly results : PromiseMap < string , string > = new PromiseMap < string , string > ( RedisBackend . TIMEOUT ) ; private readonly subscriber : Promise < IoRedis . Redis > ; param options returns public constructor ( options ? ) { this . options = ( ( ) => { if ( isNullOrUndefined ( options ) ) { return DEFAULT_REDIS_OPTIONS ; } return options ; } ) ( ) ; this . pool = new ResourcePool < IoRedis . Redis > ( ( ) => this . options . createClient ( { keyPrefix : "celery-task-meta-" } ) , async ( connection ) => { const response = await connection . quit ( ) ; connection . disconnect ( ) ; return response ; } , 2 ) ; this . subscriber = this . pool . get ( ) . then ( async ( subscriber ) => { await subscriber . psubscribe ( "celery-task-meta-*" ) ; subscriber . on ( "pmessage" , ( _ , channel , message ) => this . onMessage ( channel , message ) ) ; return subscriber ; } ) ; } param message returns public async put < T > ( message ) { const key = message . task_id ; const toPut = JSON . stringify ( message ) ; return this . pool . use ( async ( client ) : Promise < string > => client . multi ( ) . setex ( key , RedisBackend . TIMEOUT / 1000 , toPut ) . publish ( key , toPut ) . exec ( ) ) ; } param taskId param timeout returns public async get < T > ( { taskId , timeout } ) { const listen = async ( ) : Promise < ResultMessage < T > > => { const raw = await this . results . get ( taskId ) ; return JSON . parse ( raw ) ; } ; if ( this . results . has ( taskId ) ) { return listen ( ) ; } return this . pool . use ( async ( client ) => { const response = ( async ( ) => { const raw = await client . get ( taskId ) ; if ( isNullOrUndefined ( raw ) ) { return listen ( ) ; } const parsed = JSON . parse ( raw ) ; if ( parsed . status !== Status . Success ) { return listen ( ) ; } return parsed ; } ) ( ) ; return createTimeoutPromise ( response , timeout ) ; } ) ; } param taskId returns public async delete ( taskId ) { return this . pool . use ( async ( client ) : Promise < string > => { this . results . delete ( taskId ) ; return client . del ( taskId ) ; } ) ; } returns see public async disconnect ( ) { await this . end ( ) ; } returns public async end ( ) { const subscriber = await this . subscriber ; await subscriber . punsubscribe ( "celery-task-meta-*" ) ; this . pool . return ( subscriber ) ; await this . pool . destroyAll ( ) ; } returns public uri ( ) { return this . options . createUri ( ) ; } param channel param message throws private onMessage ( channel , message ) { const maybeId = channel . match ( RedisBackend . UUID_REGEX ) ; if ( isNullOrUndefined ( maybeId ) ) { throw new Error ( `channel ${ channel } is not a celery result` ) ; } const id = maybeId [ RedisBackend . UUID_INDEX ] ; this . results . resolve ( id , message ) ; } }