feat(integration): CQRS
MODIFY
packages/@integration/package.json
packages/@integration/package.json
"main": "src/index.ts",
"scripts": {
"start:http": "PORT=1337 ts-node --project tsconfig.json src/http/index.ts",
+ "start:cqrs": "PORT=1337 ts-node --project tsconfig.json src/cqrs/index.ts",
"start:websockets": "ts-node --project tsconfig.json src/websockets/http.server.ts",
"start:messaging:server": "PORT=1337 ts-node --project tsconfig.json src/messaging/server.ts",
"start:messaging:client": "ts-node --project tsconfig.json src/messaging/client.ts",

ADD
packages/@integration/src/cqrs/domain/Offer.command.ts
+import { createEvent, EventsUnion } from '@marblejs/core';
+
+export enum OfferCommandType {
+ GENERATE_OFFER_DOCUMENT = 'GENERATE_OFFER_DOCUMENT',
+};
+
+export const OfferCommand = {
+ generateOffer: createEvent(
+ OfferCommandType.GENERATE_OFFER_DOCUMENT,
+ (offerId: string) => ({ offerId }),
+ ),
+};
+
+export type OfferCommand = EventsUnion<typeof OfferCommand>;

ADD
packages/@integration/src/cqrs/domain/Offer.event.ts
+import { createEvent, EventsUnion } from '@marblejs/core';
+
+export enum OfferEventType {
+ OFFER_CREATED = 'OFFER_CREATED',
+ OFFER_UPDATED = 'OFFER_UPDATED',
+ OFFER_DOCUMENT_SAVED = 'OFFER_DOCUMENT_CREATED',
+ OFFER_DOCUMENT_CREATED = 'OFFER_DOCUMENT_SAVED',
+};
+
+export const OfferEvent = {
+ offerCreated: createEvent(
+ OfferEventType.OFFER_CREATED,
+ (offerId: string) => ({ offerId }),
+ ),
+ offerUpdated: createEvent(
+ OfferEventType.OFFER_UPDATED,
+ (offerId: string) => ({ offerId }),
+ ),
+ offerDocumentCreated: createEvent(
+ OfferEventType.OFFER_DOCUMENT_CREATED,
+ (offerId: string) => ({ offerId }),
+ ),
+ offerDocumentSaved: createEvent(
+ OfferEventType.OFFER_DOCUMENT_SAVED,
+ ),
+};
+
+export type OfferEvent = EventsUnion<typeof OfferEvent>;

ADD
packages/@integration/src/cqrs/effects/eventbus.effects.ts
+import { map, tap, delay } from 'rxjs/operators';
+import { matchEvent, useContext, LoggerToken, LoggerTag } from '@marblejs/core';
+import { MsgEffect, EVENT_BUS_CHANNEL } from '@marblejs/messaging';
+import { OfferCommand } from '../domain/Offer.command';
+import { OfferEvent } from '../domain/Offer.event';
+
+const tag = LoggerTag.EVENT_BUS;
+
+export const generateOfferDocument$: MsgEffect = (event$, ctx) => {
+ const logger = useContext(LoggerToken)(ctx.ask);
+
+ return event$.pipe(
+ matchEvent(OfferCommand.generateOffer),
+ map(event => event.payload.offerId),
+ tap(logger({ tag, type: 'generateOfferDocument$', message: 'Generating offer document...'})),
+ delay(5 * 1000), // simulate processing
+ map(OfferEvent.offerDocumentCreated),
+ map(event => ({ ...event, metadata: { replyTo: EVENT_BUS_CHANNEL }})),
+ );
+}
+
+export const saveOfferDocument$: MsgEffect = (event$, ctx) => {
+ const logger = useContext(LoggerToken)(ctx.ask);
+
+ return event$.pipe(
+ matchEvent(OfferEvent.offerDocumentCreated),
+ map(event => event.payload.offerId),
+ tap(logger({ tag, type: 'saveOfferDocument$', message: 'Saving offer document...'})),
+ delay(5 * 1000), // simulate processing
+ map(OfferEvent.offerDocumentSaved),
+ );
+}

ADD
packages/@integration/src/cqrs/effects/http.effects.ts
+import { r, HttpStatus, useContext, use } from '@marblejs/core';
+import { tap, mapTo, map } from 'rxjs/operators';
+import { EventBusClientToken } from '@marblejs/messaging';
+import { requestValidator$, t } from '@marblejs/middleware-io';
+import { OfferCommand } from '../domain/Offer.command';
+
+const generateDocumentValidator$ = requestValidator$({
+ params: t.type({
+ id: t.string,
+ }),
+});
+
+export const postDocumentsGenerate$ = r.pipe(
+ r.matchPath('/documents/:id/generate'),
+ r.matchType('POST'),
+ r.useEffect((req$, ctx) => {
+ const eventBusClient = useContext(EventBusClientToken)(ctx.ask);
+
+ return req$.pipe(
+ use(generateDocumentValidator$),
+ map(req => OfferCommand.generateOffer(req.params.id)),
+ tap(eventBusClient.emit),
+ mapTo({ status: HttpStatus.ACCEPTED }),
+ );
+ }));

ADD
packages/@integration/src/cqrs/index.ts
+import { IO } from 'fp-ts/lib/IO';
+import { isTestEnv, getPortEnv } from '@marblejs/core/dist/+internal/utils';
+import { httpListener, createServer, bindEagerlyTo } from '@marblejs/core';
+import { messagingListener, EventBusClientToken, eventBusClient, EventBusToken, eventBus } from '@marblejs/messaging';
+import { bodyParser$ } from '@marblejs/middleware-body';
+import { logger$ } from '@marblejs/middleware-logger';
+import { postDocumentsGenerate$ } from './effects/http.effects';
+import { generateOfferDocument$, saveOfferDocument$ } from './effects/eventbus.effects';
+
+const messagingListenerReader = messagingListener({
+ effects: [
+ generateOfferDocument$,
+ saveOfferDocument$,
+ ],
+});
+
+const httpListenerReader = httpListener({
+ middlewares: [
+ logger$({ silent: isTestEnv() }),
+ bodyParser$(),
+ ],
+ effects: [
+ postDocumentsGenerate$,
+ ],
+});
+
+export const server = createServer({
+ port: getPortEnv(),
+ httpListener: httpListenerReader,
+ dependencies: [
+ bindEagerlyTo(EventBusClientToken)(eventBusClient),
+ bindEagerlyTo(EventBusToken)(eventBus({ messagingListener: messagingListenerReader })),
+ ],
+});
+
+const main: IO<void> = async () =>
+ !isTestEnv() && await (await server)();
+
+main();

MODIFY
packages/@integration/src/http/effects/api.effects.ts
packages/@integration/src/http/effects/api.effects.ts
@@ -3,7 +3,6 @@ import { requestValidator$, t } from '@marblejs/middleware-io';
import { throwError } from 'rxjs';
import { map, mergeMap } from 'rxjs/operators';
import { user$ } from './user.effects';
-import { eventBus$ } from './event.effect';
import { static$ } from './static.effects';
const rootValiadtor$ = requestValidator$({
@@ -45,5 +44,5 @@ const notFound$ = r.pipe(
export const api$ = combineRoutes(
'/api/:version',
- [ root$, user$, eventBus$, static$, notImplemented$, notFound$ ],
+ [ root$, user$, static$, notImplemented$, notFound$ ],
);

DELETE
packages/@integration/src/http/effects/event.effect.ts
-import { r, HttpStatus, combineRoutes, useContext } from '@marblejs/core';
-import { EventBusClientToken } from '@marblejs/messaging';
-import { tap, mapTo, map, mergeMap } from 'rxjs/operators';
-
-const emit$ = r.pipe(
- r.matchPath('/emit'),
- r.matchType('GET'),
- r.useEffect((req$, ctx) => {
- const eventBusClient = useContext(EventBusClientToken)(ctx.ask);
-
- return req$.pipe(
- tap(() => eventBusClient.emit({ type: 'TEST' })),
- mapTo({ status: HttpStatus.ACCEPTED }),
- );
- }));
-
-const send$ = r.pipe(
- r.matchPath('/send'),
- r.matchType('GET'),
- r.useEffect((req$, ctx) => {
- const eventBusClient = useContext(EventBusClientToken)(ctx.ask);
-
- return req$.pipe(
- mergeMap(() => eventBusClient.send({ type: 'TEST' })),
- map(body => ({ body })),
- );
- }));
-
-export const eventBus$ = combineRoutes(
- '/event',
- [ emit$, send$ ],
-);

DELETE
packages/@integration/src/http/event.listener.ts
-import { map } from 'rxjs/operators';
-import { matchEvent } from '@marblejs/core';
-import { messagingListener, MsgEffect } from '@marblejs/messaging';
-
-const test$: MsgEffect = event$ =>
- event$.pipe(
- matchEvent('TEST'),
- map(event => ({ ...event, type: 'TEST_RESULT', payload: 'some_result' })),
- );
-
-export default messagingListener({
- effects: [test$],
-});

DELETE
packages/@integration/src/http/http.listener.ts
-import { httpListener } from '@marblejs/core';
-import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
-import { bodyParser$ } from '@marblejs/middleware-body';
-import { logger$ } from '@marblejs/middleware-logger';
-import { cors$ } from './middlewares/cors.middleware';
-import { api$ } from './effects/api.effects';
-
-export default httpListener({
- middlewares: [
- logger$({ silent: isTestEnv() }),
- bodyParser$(),
- cors$,
- ],
- effects: [api$]
-});

MODIFY
packages/@integration/src/http/index.ts
packages/@integration/src/http/index.ts
import { IO } from 'fp-ts/lib/IO';
-import { createServer, bindEagerlyTo } from '@marblejs/core';
-import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
-import { EventBusToken, EventBusClientToken, eventBus, eventBusClient } from '@marblejs/messaging';
-import httpListener from './http.listener';
-import messagingListener from './event.listener';
+import { createServer, httpListener } from '@marblejs/core';
+import { isTestEnv, getPortEnv } from '@marblejs/core/dist/+internal/utils';
+import { logger$ } from '@marblejs/middleware-logger';
+import { bodyParser$ } from '@marblejs/middleware-body';
+import { api$ } from './effects/api.effects';
+import { cors$ } from './middlewares/cors.middleware';
-const port = process.env.PORT
- ? Number(process.env.PORT)
- : undefined;
+const httpListenerReader = httpListener({
+ middlewares: [
+ logger$({ silent: isTestEnv() }),
+ bodyParser$(),
+ cors$,
+ ],
+ effects: [
+ api$,
+ ],
+});
export const server = createServer({
- port,
- httpListener,
- dependencies: [
- bindEagerlyTo(EventBusClientToken)(eventBusClient),
- bindEagerlyTo(EventBusToken)(eventBus({ messagingListener })),
- ],
+ port: getPortEnv(),
+ httpListener: httpListenerReader,
});
const main: IO<void> = async () =>

MODIFY
packages/core/src/+internal/utils/any.util.ts
packages/core/src/+internal/utils/any.util.ts
+import { pipe } from 'fp-ts/lib/pipeable';
+import * as O from 'fp-ts/lib/Option';
+
export const isNonNullable = <T>(value: T): value is NonNullable<T> =>
value !== undefined && value !== null;
@@ -6,3 +9,9 @@ export const isNullable = <T>(value: T) =>
export const isTestEnv = () =>
process.env.NODE_ENV === 'test';
+
+export const getPortEnv = () => pipe(
+ O.fromNullable(process.env.PORT),
+ O.map(Number),
+ O.toUndefined,
+);

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -4,7 +4,9 @@ export * from './client/messaging.client.interface';
// transport
export * from './transport/transport.interface';
-export { AmqpConnectionStatus } from './transport/strategies/amqp.strategy.interface';
+export { AmqpStrategyOptions, AmqpConnectionStatus } from './transport/strategies/amqp.strategy.interface';
+export { RedisStrategyOptions, RedisConnectionStatus } from './transport/strategies/redis.strategy.interface';
+export { LocalStrategyOptions, EVENT_BUS_CHANNEL } from './transport/strategies/local.strategy.interface';
// effects
export * from './effects/messaging.effects.interface';

MODIFY
packages/messaging/src/transport/strategies/local.strategy.interface.ts
packages/messaging/src/transport/strategies/local.strategy.interface.ts
import { Transport } from '../transport.interface';
+export const EVENT_BUS_CHANNEL = 'event_bus';
+
export interface LocalStrategy {
transport: Transport.LOCAL;
options: LocalStrategyOptions;

MODIFY
packages/messaging/src/transport/strategies/local.strategy.ts
packages/messaging/src/transport/strategies/local.strategy.ts
@@ -3,7 +3,7 @@ import { share, take, filter } from 'rxjs/operators';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { TransportLayer, TransportLayerConnection, TransportMessage, Transport, DEFAULT_TIMEOUT, TransportLayerConfig } from '../transport.interface';
import { throwUnsupportedError } from '../transport.error';
-import { LocalStrategyOptions } from './local.strategy.interface';
+import { LocalStrategyOptions, EVENT_BUS_CHANNEL } from './local.strategy.interface';
class LocalStrategyConnection implements TransportLayerConnection {
private errorSubject$ = new Subject<Error>();
@@ -82,7 +82,7 @@ class LocalStrategy implements TransportLayer {
get config() {
return ({
host: 'localhost',
- channel: 'event_bus',
+ channel: EVENT_BUS_CHANNEL,
timeout: this.options.timeout ?? DEFAULT_TIMEOUT,
});
}

feat(integration): CQRS