import { getClassMetadata , listModule , listPropertyDataFromClass , MidwayFrameworkType , ConsumerMetadata , MS_CONSUMER_KEY , MSListenerType , RabbitMQListenerOptions , Framework , } from '@midwayjs/decorator' ; import { BaseFramework } from '@midwayjs/core' ; import { IMidwayRabbitMQApplication , IMidwayRabbitMQConfigurationOptions , IMidwayRabbitMQContext , } from './interface' ; import { RabbitMQServer } from './mq' ; import { ConsumeMessage } from 'amqplib' ; @ Framework ( ) export class MidwayRabbitMQFramework extends BaseFramework < IMidwayRabbitMQApplication , IMidwayRabbitMQContext , IMidwayRabbitMQConfigurationOptions > { public app : IMidwayRabbitMQApplication ; public consumerHandlerList = [ ] ; configure ( ) { return this . configService . getConfiguration ( 'rabbitMQServer' ) ; } async applicationInitialize ( options ) { this . app = new RabbitMQServer ( { logger : this . logger , ... this . configurationOptions , } ) as unknown as IMidwayRabbitMQApplication ; } public async run ( ) { await this . app . connect ( this . configurationOptions . url , this . configurationOptions . socketOptions ) ; await this . loadSubscriber ( ) ; this . logger . info ( 'Rabbitmq server start success' ) ; } protected async beforeStop ( ) { await this . app . close ( ) ; } public getFrameworkType ( ) { return MidwayFrameworkType . MS_RABBITMQ ; } private async loadSubscriber ( ) { const subscriberModules = listModule ( MS_CONSUMER_KEY , module => { const metadata = getClassMetadata ( MS_CONSUMER_KEY , module ) ; return metadata . type === MSListenerType . RABBITMQ ; } ) ; for ( const module of subscriberModules ) { const data = listPropertyDataFromClass ( MS_CONSUMER_KEY , module ) ; for ( const methodBindListeners of data ) { for ( const listenerOptions of methodBindListeners ) { await this . app . createConsumer ( listenerOptions , async ( data , channel , channelWrapper ) => { const ctx = { channel , queueName : listenerOptions . queueName , ack : data => { return channelWrapper . ack ( data ) ; } , } as IMidwayRabbitMQContext ; this . app . createAnonymousContext ( ctx ) ; const ins = await ctx . requestContext . getAsync ( module ) ; const fn = await this . applyMiddleware ( async ( ctx ) => { return await ins [ listenerOptions . propertyKey ] . call ( ins , data ) ; } ) ; try { const result = await fn ( ctx ) ; if ( result ) { return channelWrapper . ack ( data ) ; } } catch ( error ) { this . logger . error ( error ) ; } } ) ; } } } } public getFrameworkName ( ) { return 'midway:rabbitmq' ; } }