feat(core): act operator
MODIFY
packages/@integration/src/cqrs/effects/eventbus.effects.ts
packages/@integration/src/cqrs/effects/eventbus.effects.ts
+import { of } from 'rxjs';
import { map, tap, delay } from 'rxjs/operators';
-import { matchEvent, useContext, LoggerToken, LoggerTag } from '@marblejs/core';
+import { act, matchEvent, useContext, LoggerToken, LoggerTag } from '@marblejs/core';
import { MsgEffect, EVENT_BUS_CHANNEL, reply } from '@marblejs/messaging';
import { OfferCommand } from '../domain/Offer.command';
import { OfferEvent } from '../domain/Offer.event';
@@ -11,12 +12,14 @@ export const generateOfferDocument$: MsgEffect = (event$, ctx) => {
return event$.pipe(
matchEvent(OfferCommand.generateOffer),
+ act(event => of(event).pipe(
map(event => event.payload.offerId),
tap(logger({ tag, type: 'generateOfferDocument$', message: 'Generating offer document...'})),
delay(5 * 1000),
map(offerId => reply(EVENT_BUS_CHANNEL)(
OfferEvent.offerDocumentCreated(offerId),
)),
+ )),
);
}
@@ -25,9 +28,11 @@ export const saveOfferDocument$: MsgEffect = (event$, ctx) => {
return event$.pipe(
matchEvent(OfferEvent.offerDocumentCreated),
+ act(event => of(event).pipe(
map(event => event.payload.offerId),
tap(logger({ tag, type: 'saveOfferDocument$', message: 'Saving offer document...'})),
delay(5 * 1000),
map(OfferEvent.offerDocumentSaved),
+ )),
);
}

ADD
packages/core/src/operators/act/act.operator.spec.ts
+import { Observable, of, concat } from 'rxjs';
+import { filter, tap, mergeMap } from 'rxjs/operators';
+import { Event } from '../../event/event.interface';
+import { Marbles } from '../../+internal/testing';
+import { act } from './act.operator';
+
+describe('#act operator', () => {
+ test('emits output event from async (Observable) call function', () => {
+ // given
+ const event1: Event = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
+ const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act(event => of(event)),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ]);
+ });
+
+ test('emits multiple output events', () => {
+ // given
+ const event1: Event = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
+ const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act((event: any) => of(event).pipe(
+ mergeMap(e => e.payload === 2
+ ? concat(of(e), of(e))
+ : of(e)),
+ )),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b----c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['-a-(bb)-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ]);
+ });
+
+ test('emits filtered output event', () => {
+ // given
+ const event1: Event = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
+ const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act((event: any) => of(event).pipe(
+ filter(e => e.payload >= 2),
+ )),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['---b-c-d---', { b: event2, c: event3, d: event4 }],
+ ]);
+ });
+
+ test('catches errored event => maps to default => continues stream', () => {
+ // given
+ const event1: Event = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
+ const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
+ const eventErrored: Event = { type: 'TEST_EVENT_2', error: { name: 'Error', message: 'something went wrong' } };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act((event: any) => of(event).pipe(
+ tap(e => { if (e.payload === 2) throw new Error('something went wrong') }),
+ )),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: event1, b: eventErrored, c: event3, d: event4 }],
+ ]);
+ });
+
+ test('catches errored event => maps to custom event => continues stream', () => {
+ // given
+ const event1: Event = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
+ const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
+ const eventErrored: Event = { type: 'TEST_ERROR', error: 'something went wrong' };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act(
+ (event: any) => of(event).pipe(
+ tap(e => { if (e.payload === 2) throw new Error('something went wrong') }),
+ ),
+ (error: Error) => ({
+ type: 'TEST_ERROR',
+ error: error.message,
+ }),
+ ),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: event1, b: eventErrored, c: event3, d: event4 }],
+ ]);
+ });
+
+ test('catches errored event => maps to custom event (based on errored one) => continues stream', () => {
+ // given
+ const event1: Event = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
+ const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
+ const eventErrored: Event = { type: 'TEST_EVENT_2', error: 'something went wrong' };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act(
+ (event: any) => of(event).pipe(
+ tap(e => { if (e.payload === 2) throw new Error('something went wrong') }),
+ ),
+ (error: Error, event: Event) => ({
+ type: event.type,
+ error: error.message,
+ }),
+ ),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b-c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['-a-b-c-d---', { a: event1, b: eventErrored, c: event3, d: event4 }],
+ ]);
+ });
+
+ test('catches errored event => maps to observable of new events => continues stream', () => {
+ // given
+ const event1: Event = { type: 'TEST_EVENT_1', payload: 1 };
+ const event2: Event = { type: 'TEST_EVENT_2', payload: 2 };
+ const event3: Event = { type: 'TEST_EVENT_3', payload: 3 };
+ const event4: Event = { type: 'TEST_EVENT_4', payload: 4 };
+ const eventErrored: Event = { type: 'TEST_ERROR' };
+
+ // when
+ const effect$ = (event$: Observable<Event>) =>
+ event$.pipe(
+ act(
+ (event: any) => of(event).pipe(
+ tap(e => { if (e.payload === 2) throw new Error('something went wrong') }),
+ ),
+ () => concat(
+ of(eventErrored),
+ of(eventErrored),
+ ),
+ ),
+ );
+
+ // then
+ Marbles.assertEffect(effect$, [
+ ['-a-b----c-d---', { a: event1, b: event2, c: event3, d: event4 }],
+ ['-a-(bb)-c-d---', { a: event1, b: eventErrored, c: event3, d: event4 }],
+ ]);
+ });
+});

ADD
packages/core/src/operators/act/act.operator.ts
+import * as O from 'fp-ts/lib/Option';
+import { pipe } from 'fp-ts/lib/pipeable';
+import { Observable, of, defer, isObservable } from 'rxjs';
+import { mergeMap, catchError } from 'rxjs/operators';
+import { Event } from '../../event/event.interface';
+
+export function act<
+ InputEvent extends Event,
+ CallEvent extends Event,
+>(
+ callFn: (event: InputEvent) => Observable<CallEvent>,
+): (source: Observable<InputEvent>) => Observable<CallEvent>;
+
+export function act<
+ InputEvent extends Event,
+ CallEvent extends Event,
+ ErrorEvent extends Event,
+>(
+ callFn: (event: InputEvent) => Observable<CallEvent>,
+ errorFn: (error: any, event: InputEvent) => ErrorEvent | Observable<ErrorEvent>,
+): (source: Observable<InputEvent>) => Observable<CallEvent | ErrorEvent>;
+
+export function act<
+ InputEvent extends Event,
+ CallEvent extends Event,
+ ErrorEvent extends Event,
+>(
+ callFn: (event: InputEvent) => Observable<CallEvent>,
+ errorFn?: (error: any, event: InputEvent) => ErrorEvent | Observable<ErrorEvent>,
+) {
+
+ const getDefaultErrorEvent = (error: Error) => (event: Event) => of({
+ type: event.type,
+ error: { name: error.name, message: error.message },
+ } as ErrorEvent);
+
+ return (source: Observable<InputEvent>): Observable<CallEvent | ErrorEvent> =>
+ source.pipe(
+ mergeMap(event => defer(() =>
+ callFn(event).pipe(
+ catchError(error => pipe(
+ O.fromNullable(errorFn),
+ O.map(fn => fn(error, event)),
+ O.map(res => !isObservable(res) ? of(res) : res),
+ O.getOrElse(() => getDefaultErrorEvent(error)(event)),
+ )),
+ ),
+ )),
+ );
+}

MODIFY
packages/core/src/operators/index.ts
packages/core/src/operators/index.ts
export { use } from './use/use.operator';
+export { act } from './act/act.operator';
export { switchToProtocol } from './switchToProtocol/switchToProtocol.operator';
export { matchEvent } from './matchEvent/matchEvent.operator';

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -85,7 +85,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
connection.message$,
e$ => e$.pipe(map(decode)),
e$ => combinedMiddlewares(e$, ctx),
- e$ => e$.pipe(catchError(error => defer(() => processError(incomingEvent$)(error)))),
+ e$ => defer(() => processError(e$)),
);
const outgoingEvent$ = pipe(
@@ -93,21 +93,22 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
e$ => combinedEffects(e$, ctx),
e$ => output$(e$, ctx),
e$ => outputLogger$(e$, ctx),
- e$ => e$.pipe(catchError(error => defer(() => processError(outgoingEvent$)(error)))),
+ e$ => defer(() => processError(e$)),
);
const errorEvent$ = pipe(
errorSubject.asObservable(),
e$ => error$(e$, ctx),
- e$ => errorLogger$(e$, ctx),
+ e$ => defer(() => errorLogger$(e$, ctx)),
);
- const processError = (originStream$: Observable<any>) => (error: Error) => {
+ const processError = <T>(origin$: Observable<T>): Observable<T> =>
+ origin$.pipe(catchError(error => {
errorSubject.next(error);
- return originStream$;
- };
+ return processError(origin$);
+ }));
- const subscribeIncomingEvent = (event$: Observable<Event<unknown, any, string>>) =>
+ const subscribeIncomingEvent = (event$: Observable<Event>) =>
event$
.pipe(takeUntil(connection.close$))
.subscribe(

feat(core): act operator