import * as Redis from "ioredis" ; import * as urllib from "url" ; import { v4 } from "uuid" ; import { CeleryBroker } from "." ; import { Message } from "../message" ; private param { String } urlString function redisOptsFromUrl ( urlString ) { const redisOpts = { } as Redis . RedisOptions ; try { const redisUrl = urllib . parse ( urlString ) ; redisOpts . port = + redisUrl . port || 6379 ; redisOpts . host = redisUrl . hostname ; redisOpts . db = redisUrl . pathname ? + redisUrl . pathname . split ( "/" ) [ 1 ] : 0 ; if ( redisUrl . auth ) { [ , redisOpts . password ] = redisUrl . auth . split ( ":" ) ; } } catch ( e ) { throw new Error ( e . message ) ; } return redisOpts ; } class RedisMessage extends Message { private raw : object ; constructor ( payload ) { super ( Buffer . from ( payload [ "body" ] , "base64" ) , payload [ "content-type" ] , payload [ "content-encoding" ] , payload [ "properties" ] , payload [ "headers" ] ) ; this . raw = payload ; } } export default class RedisBroker implements CeleryBroker { redis : Redis . Redis ; channels = [ ] ; closing = false ; constructor param { string } url param { object } opts constructor ( url , opts ) { this . redis = new Redis ( { ... redisOptsFromUrl ( url ) , ... opts } ) ; } method returns { Promise } public isReady ( ) { return new Promise ( ( resolve , reject ) => { if ( this . redis . status === "ready" ) { resolve ( ) ; } else { let handleError ; const handleReady = ( ) => { this . redis . removeListener ( "error" , handleError ) ; resolve ( ) ; } ; handleError = err => { this . redis . removeListener ( "ready" , handleReady ) ; reject ( err ) ; } ; this . redis . once ( "ready" , handleReady ) ; this . redis . once ( "error" , handleError ) ; } } ) ; } method returns { Promise } public disconnect ( ) { this . closing = true ; return Promise . all ( this . channels ) . then ( ( ) => this . redis . quit ( ) ) ; } method returns { Promise } public publish ( body , exchange , routingKey , headers , properties ) { const messageBody = JSON . stringify ( body ) ; const contentType = "application/json" ; const contentEncoding = "utf-8" ; const message = { body : Buffer . from ( messageBody ) . toString ( "base64" ) , "content-type" : contentType , "content-encoding" : contentEncoding , headers , properties : { body_encoding : "base64" , delivery_info : { exchange : exchange , routing_key : routingKey } , delivery_mode : 2 , delivery_tag : v4 ( ) , ... properties } } ; return this . redis . lpush ( routingKey , JSON . stringify ( message ) ) ; } method param { string } queue param { Function } callback returns { Promise } public subscribe ( queue , callback ) { const promiseCount = 1 ; return this . isReady ( ) . then ( ( ) => { for ( let index = 0 ; index < promiseCount ; index += 1 ) { this . channels . push ( new Promise ( resolve => this . receive ( index , resolve , queue , callback ) ) ) ; } return Promise . all ( this . channels ) ; } ) ; } private param { number } index param { Fucntion } resolve param { string } queue param { Function } callback private receive ( index , resolve , queue , callback ) { process . nextTick ( ( ) => this . recieveOneOnNextTick ( index , resolve , queue , callback ) ) ; } private param { number } index param { Function } resolve param { String } queue param { Function } callback returns { Promise } private recieveOneOnNextTick ( index , resolve , queue , callback ) { if ( this . closing ) { resolve ( ) ; return ; } return this . receiveOne ( queue ) . then ( body => { if ( body ) { callback ( body ) ; } Promise . resolve ( ) ; } ) . then ( ( ) => this . receive ( index , resolve , queue , callback ) ) . catch ( err => console . log ( err ) ) ; } private param { string } queue return { Promise } private receiveOne ( queue ) { return this . redis . brpop ( queue , "5" ) . then ( result => { if ( ! result ) { return null ; } const [ queue , item ] = result ; const rawMsg = JSON . parse ( item ) ; if ( rawMsg [ "content-type" ] !== "application/json" ) { throw new Error ( `queue ${ queue } item: unsupported content type ${ rawMsg [ "content-type" ] }` ) ; } if ( rawMsg . properties . body_encoding !== "base64" ) { throw new Error ( `queue ${ queue } item: unsupported body encoding ${ rawMsg . properties . body_encoding }` ) ; } if ( rawMsg [ "content-encoding" ] !== "utf-8" ) { throw new Error ( `queue ${ queue } item: unsupported content encoding ${ rawMsg [ "content-encoding" ] }` ) ; } return new RedisMessage ( rawMsg ) ; } ) ; } }