import * as url from "url" ; import RedisBroker from "./redis" ; import AMQPBroker from "./amqp" ; export interface CeleryBroker { isReady : ( ) => Promise < any > ; disconnect : ( ) => Promise < any > ; publish : ( body : object | [ Array < any > , object , object ] , exchange : string , routingKey : string , headers : object , properties : object ) => Promise < any > ; subscribe : ( queue : string , callback : Function ) => Promise < any > ; } private constant const supportedProtocols = [ "redis" , "amqp" , "amqps" ] ; private param { String } uri returns { String } throws function getProtocol ( uri ) { const protocol = url . parse ( uri ) . protocol . slice ( 0 , - 1 ) ; if ( supportedProtocols . indexOf ( protocol ) === - 1 ) { throw new Error ( `Unsupported type: ${ protocol }` ) ; } return protocol ; } param { String } CELERY_BROKER param { object } CELERY_BROKER_OPTIONS param { string } CELERY_QUEUE returns { CeleryBroker } export function newCeleryBroker ( CELERY_BROKER , CELERY_BROKER_OPTIONS , CELERY_QUEUE = "celery" ) { const brokerProtocol = getProtocol ( CELERY_BROKER ) ; if ( brokerProtocol === "redis" ) { return new RedisBroker ( CELERY_BROKER , CELERY_BROKER_OPTIONS ) ; } if ( [ 'amqp' , 'amqps' ] . indexOf ( brokerProtocol ) > - 1 ) { return new AMQPBroker ( CELERY_BROKER , CELERY_BROKER_OPTIONS , CELERY_QUEUE ) ; } throw new Error ( "unsupprted celery broker" ) ; }