feat(middleware-io): allow to pass single event to middleware
MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
import { of } from 'rxjs';
import { map, tap, catchError } from 'rxjs/operators';
-import { matchEvent, use, act } from '@marblejs/core';
+import { matchEvent, act } from '@marblejs/core';
import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { createMicroservice, messagingListener, Transport, MsgEffect, reply } from '@marblejs/messaging';
import { IO } from 'fp-ts/lib/IO';
+import { pipe } from 'fp-ts/lib/pipeable';
const fib = (n: number): number =>
(n === 0 || n === 1) ? n : fib(n - 1) + fib(n - 2);
@@ -12,8 +13,8 @@ const fib = (n: number): number =>
const fibonacci$: MsgEffect = event$ =>
event$.pipe(
matchEvent('FIB'),
- act(event => of(event).pipe(
- use(eventValidator$(t.number)),
+ act(event => pipe(
+ eventValidator$(t.number)(event),
tap(event => { if (event.payload >= 45) throw new Error('Too high number!') }),
map(event => fib(event.payload)),
map(payload => reply(event)({ type: 'FIB_RESULT', payload })),

MODIFY
packages/@integration/src/websockets/websockets.server.ts
packages/@integration/src/websockets/websockets.server.ts
import { WsEffect, webSocketListener, createWebSocketServer } from '@marblejs/websockets';
import { buffer, map } from 'rxjs/operators';
-import { matchEvent, use } from '@marblejs/core';
+import { matchEvent, act } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
+import { flow } from 'fp-ts/lib/function';
const sum$: WsEffect = event$ =>
event$.pipe(
@@ -11,10 +12,12 @@ const sum$: WsEffect = event$ =>
const add$: WsEffect = (event$, ctx) =>
event$.pipe(
matchEvent('ADD'),
- use(eventValidator$(t.number)),
+ act(flow(
+ eventValidator$(t.number),
buffer(sum$(event$, ctx)),
map(addEvents => addEvents.reduce((a, e) => e.payload + a, 0)),
map(payload => ({ type: 'SUM_RESULT', payload })),
+ ))
);
export const webSocketServer = createWebSocketServer({

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
-import { Event, matchEvent, use, combineEffects, act, EventError } from '@marblejs/core';
+import { Event, matchEvent, combineEffects, act, EventError } from '@marblejs/core';
import { wait, NamedError } from '@marblejs/core/dist/+internal/utils';
import { eventValidator$, t } from '@marblejs/middleware-io';
-import { throwError, of } from 'rxjs';
+import { throwError } from 'rxjs';
import { delay, map, tap, ignoreElements } from 'rxjs/operators';
+import { flow } from 'fp-ts/lib/function';
import { MsgEffect } from '../effects/messaging.effects.interface';
import { runEventBus, runEventBusClient } from '../util/messaging.test.util';
import { reply } from '../reply/reply';
@@ -12,9 +13,11 @@ describe('#eventBus', () => {
const rpc$: MsgEffect = event$ =>
event$.pipe(
matchEvent('RPC_TEST'),
- use(eventValidator$(t.number)),
+ act(flow(
+ eventValidator$(t.number),
delay(10),
map(event => reply(event)({ type: 'RPC_TEST_RESULT', payload: event.payload + 1 })),
+ ))
);
const eventBus = await runEventBus({ effects: [rpc$] });
@@ -33,17 +36,21 @@ describe('#eventBus', () => {
const rpc1$: MsgEffect = event$ =>
event$.pipe(
matchEvent('RPC_1_TEST'),
- use(eventValidator$(t.number)),
+ act(flow(
+ eventValidator$(t.number),
delay(10),
map(event => reply(event)({ type: 'RPC_1_TEST_RESULT', payload: event.payload + 1 })),
+ ))
);
const rpc2$: MsgEffect = event$ =>
event$.pipe(
matchEvent('RPC_2_TEST'),
- use(eventValidator$(t.number)),
+ act(flow(
+ eventValidator$(t.number),
delay(10),
map(event => reply(event)({ type: 'RPC_2_TEST_RESULT', payload: event.payload + 1 })),
+ ))
);
const eventBus = await runEventBus({ effects: [combineEffects(rpc1$, rpc2$)] });
@@ -76,8 +83,8 @@ describe('#eventBus', () => {
const rpc1$: MsgEffect = event$ =>
event$.pipe(
matchEvent('RPC_TEST'),
- act(event => of(event).pipe(
- use(eventValidator$(t.number)),
+ act(flow(
+ eventValidator$(t.number),
delay(10),
)),
);

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
-import { matchEvent, use } from '@marblejs/core';
+import { matchEvent, act } from '@marblejs/core';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { forkJoin, TimeoutError } from 'rxjs';
import { map, tap, delay, mapTo } from 'rxjs/operators';
+import { flow } from 'fp-ts/lib/function';
import { TransportLayerConnection } from '../../transport/transport.interface';
import { MsgEffect } from '../../effects/messaging.effects.interface';
import { MessagingClient } from '../../client/messaging.client.interface';
@@ -36,9 +37,11 @@ describe('messagingServer::AMQP', () => {
const increment$: MsgEffect = event$ =>
event$.pipe(
matchEvent('INCREMENT'),
- use(eventValidator$(t.number)),
+ act(flow(
+ eventValidator$(t.number),
delay(50),
map(event => ({ ...event, type: 'INCREMENT_RESULT', payload: event.payload + 1 })),
+ )),
);
// when
@@ -97,9 +100,11 @@ describe('messagingServer::AMQP', () => {
const increment$: MsgEffect = event$ =>
event$.pipe(
matchEvent('INCREMENT'),
- use(eventValidator$(t.number)),
+ act(flow(
+ eventValidator$(t.number),
delay(50),
map(event => ({ ...event, type: 'INCREMENT_RESULT', payload: event.payload + 1 })),
+ ))
);
// then

MODIFY
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
-import { matchEvent, use } from '@marblejs/core';
+import { matchEvent, act } from '@marblejs/core';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { forkJoin, TimeoutError } from 'rxjs';
import { map, tap, delay, mapTo } from 'rxjs/operators';
+import { flow } from 'fp-ts/lib/function';
import { TransportLayerConnection } from '../../transport/transport.interface';
import { MsgEffect } from '../../effects/messaging.effects.interface';
import { MessagingClient } from '../../client/messaging.client.interface';
@@ -40,9 +41,11 @@ describe('messagingServer::Redis', () => {
const increment$: MsgEffect = event$ =>
event$.pipe(
matchEvent('INCREMENT'),
- use(eventValidator$(t.number)),
+ act(flow(
+ eventValidator$(t.number),
delay(50),
map(event => ({ ...event, type: 'INCREMENT_RESULT', payload: event.payload + 1 })),
+ )),
);
// when
@@ -101,9 +104,11 @@ describe('messagingServer::Redis', () => {
const increment$: MsgEffect = event$ =>
event$.pipe(
matchEvent('INCREMENT'),
- use(eventValidator$(t.number)),
+ act(flow(
+ eventValidator$(t.number),
delay(50),
map(event => ({ ...event, type: 'INCREMENT_RESULT', payload: event.payload + 1 })),
+ )),
);
// then

MODIFY
packages/middleware-io/src/io.event.middleware.ts
packages/middleware-io/src/io.event.middleware.ts
import { Event, EventError, ValidatedEvent, isEventCodec } from '@marblejs/core';
-import { Observable, of, throwError } from 'rxjs';
+import { Observable, of, throwError, isObservable } from 'rxjs';
import { pipe } from 'fp-ts/lib/pipeable';
import { mergeMap, catchError, map } from 'rxjs/operators';
import { Schema, ValidatorOptions, validator$ } from './io.middleware';
@@ -36,6 +36,8 @@ export const eventValidator$ = <U extends Schema>(schema: U, options?: Validator
)),
) as Observable<ValidationResult<U>>;
- return (event$: Observable<Event>) =>
- event$.pipe(mergeMap(validate));
+ return (input: Observable<Event> | Event) =>
+ isObservable(input)
+ ? input.pipe(mergeMap(validate))
+ : validate(input);
};

MODIFY
packages/middleware-io/src/specs/io.event.middleware.spec.ts
packages/middleware-io/src/specs/io.event.middleware.spec.ts
@@ -94,4 +94,27 @@ describe('#eventValidator$', () => {
}],
}));
});
+
+ test('allows to pass single event outside Observable stream', async () => {
+ // given
+ const input = {
+ type: 'TEST_EVENT',
+ payload: {
+ name: 'test',
+ age: 100,
+ }
+ };
+
+ const schema = t.type({
+ name: t.string,
+ age: t.number,
+ });
+
+ // when
+ const result = await eventValidator$(schema)(input).toPromise();
+
+ // then
+ expect(result.type).toEqual(input.type);
+ expect(result.payload).toEqual(input.payload);
+ });
});

feat(middleware-io): allow to pass single event to middleware