import { Injectable } from '@nestjs/common' ; import { IEvent , IEventPublisher , IMessageSource } from '@nestjs/cqrs' ; import { TCPClient } from 'geteventstore-promise' ; import * as http from 'http' ; import { ConfigService } from 'nestjs-config' ; import { Subject } from 'rxjs' ; import { AggregateRoot } from '../domain/models/aggregate-root' ; class description @ Injectable ( ) export class EventStore implements IEventPublisher , IMessageSource { private _category : string ; private _eventHandlers : object ; private _eventStoreHostUrl : string ; constructor ( config , private readonly client ) { this . _category = config . get ( 'eventstore' ) . category ; this . _eventStoreHostUrl = config . get ( 'eventstore' ) . protocol + `://${ config . get ( 'eventstore' ) . hostname }:${ config . get ( 'eventstore' ) . httpPort }/streams/` ; } async publish < T extends IEvent > ( event ) { if ( 'id' in event === false ) { throw new Error ( 'Not a DomainEvent' ) ; } const message = JSON . parse ( JSON . stringify ( event ) ) ; const id = message . id ; const streamName = `${ this . _category }-${ id }` ; const type = event . constructor . name ; const metadata = { _aggregate_id : id , _ocurred_on : new Date ( ) . getTime ( ) , } ; try { await this . client . writeEvent ( streamName , type , event , metadata ) ; } catch ( err ) { console . trace ( err ) ; } } async read < T extends AggregateRoot > ( aggregate , id ) { const streamName = `${ this . _category }-${ id }` ; try { const entity = Reflect . construct ( aggregate , [ ] ) ; const response = await this . client . getEvents ( streamName ) ; const events = response . map ( event => { const eventType = event . eventType ; const data = event . data ; return this . _eventHandlers [ eventType ] ( ... Object . values ( data ) ) ; } ) ; if ( events . length === 0 ) { return null ; } entity . loadFromHistory ( events ) ; return entity ; } catch ( err ) { console . trace ( err ) ; } return null ; } description param subject async bridgeEventsTo < T extends IEvent > ( subject ) { const streamName = `$ce-${ this . _category }` ; const onEvent = async ( ... event ) => { const eventUrl = this . _eventStoreHostUrl + `${ event [ 1 ] . metadata . $o }/${ event [ 1 ] . data . split ( '@' ) [ 0 ] }` ; const requestOptions = { headers : { Accept : 'application/vnd.eventstore.atom+json' , Authorization : 'Basic YWRtaW46Y2hhbmdlaXQ=' , } , } ; http . get ( eventUrl , requestOptions , res => { res . setEncoding ( 'utf8' ) ; let rawData = '' ; res . on ( 'data' , chunk => { rawData += chunk ; } ) ; res . on ( 'end' , ( ) => { try { const message = JSON . parse ( rawData ) ; const eventType = message . content . eventType ; const data = message . content . data ; event = this . _eventHandlers [ eventType ] ( ... Object . values ( data ) ) ; subject . next ( event ) ; } catch ( err ) { console . trace ( err ) ; } } ) ; } ) ; } ; const onDropped = ( subscription , reason , error ) => { console . trace ( subscription , reason , error ) ; } ; try { await this . client . subscribeToStream ( streamName , onEvent , onDropped , false ) ; } catch ( err ) { console . trace ( err ) ; } } addEventHandlers ( eventHandlers ) { this . _eventHandlers = { ... this . _eventHandlers , ... eventHandlers , } ; } }