import { CloudEvent } from 'cloudevents' ; import { IDomainEvent } from '../domain/IDomainEvent' ; import { CloudEventFactory , IBrokerInterface } from '@micheleangioni/node-messagebrokers' ; import ILogger from '../infrastructure/logger/ILogger' ; import { GroupedByAggregateCloudEvents , GroupedByAggregateEvents } from './declarations' ; export default class EventPublisher { constructor ( private readonly messageBroker , private readonly logger ) { } param { string } source param { IDomainEvent [ } events public async publish ( source , events ) { const groupedEvents = events . reduce ( ( acc , event ) => { if ( ! acc [ event . getEventAggregate ( ) ] ) { acc [ event . getEventAggregate ( ) ] = [ event ] ; } else { acc [ event . getEventAggregate ( ) ] . push ( event ) ; } return acc ; } , { } ) ; const groupedCloudevents = Object . keys ( groupedEvents ) . reduce ( ( acc , aggregate ) => { acc [ aggregate ] = groupedEvents [ aggregate ] . map ( ( domainEvent ) => this . convertDomainEventToCloudEvent ( source , domainEvent ) ) ; return acc ; } , { } ) ; const promises = Object . keys ( groupedCloudevents ) . map ( ( aggregate ) => { return this . messageBroker . sendMessage ( aggregate , groupedCloudevents [ aggregate ] ) ; } ) ; await Promise . all ( promises ) ; if ( events . length === 1 ) { this . logger . info ( { data : events [ 0 ] . getEventData ( ) , message : `Successfully published ${ events [ 0 ] . getEventAggregate ( ) } ${ events [ 0 ] . getEventName ( ) } event` , type : 'sns' , } ) ; } else { Object . keys ( groupedEvents ) . forEach ( ( aggregate ) => { this . logger . info ( { message : `Successfully published ${ groupedEvents [ aggregate ] . length } ${ aggregate } events` , type : 'sns' , } ) ; } ) ; } } private convertDomainEventToCloudEvent ( source , event ) { return CloudEventFactory . createV1 ( event . getEventAggregate ( ) , event . getEventName ( ) , source , event . getEventData ( ) ) ; } }