feat(messaging): messagingClient - detect error event response and throw exception
MODIFY
packages/core/src/+internal/utils/error.util.ts
packages/core/src/+internal/utils/error.util.ts
-export class ExtendableError extends Error {
+export class NamedError extends Error {
constructor(public name: string, message: string) {
super(message);
}
}
-export const isExtendableError = (data: any): data is ExtendableError =>
+export const isNamedError = (data: any): data is NamedError =>
!!data?.name && !!data?.message;
export const isError = (data: any): data is Error =>

MODIFY
packages/core/src/error/error.model.ts
packages/core/src/error/error.model.ts
import { Event } from '../event/event.interface';
-import { ExtendableError } from '../+internal/utils';
+import { NamedError } from '../+internal/utils';
export enum ErrorType {
CORE_ERROR = 'CoreError',
@@ -7,7 +7,7 @@ export enum ErrorType {
EVENT_ERROR = 'EventError',
}
-export class CoreError extends ExtendableError {
+export class CoreError extends NamedError {
constructor(
public readonly message: string,
options: {
@@ -21,7 +21,7 @@ export class CoreError extends ExtendableError {
}
}
-export class ContextError extends ExtendableError {
+export class ContextError extends NamedError {
constructor(
public readonly message: string,
) {
@@ -29,7 +29,7 @@ export class ContextError extends ExtendableError {
}
}
-export class EventError extends ExtendableError {
+export class EventError extends NamedError {
constructor(
public readonly event: Event,
public readonly message: string,

MODIFY
packages/core/src/http/error/http.error.model.ts
packages/core/src/http/error/http.error.model.ts
import chalk from 'chalk';
import { HttpStatus, HttpRequest } from '../http.interface';
-import { ExtendableError } from '../../+internal/utils';
+import { NamedError } from '../../+internal/utils';
import { HttpEffectResponse } from '../effects/http.effects.interface';
import { coreErrorFactory } from '../../error/error.factory';
@@ -9,7 +9,7 @@ export enum HttpErrorType {
HTTP_REQUEST_ERROR = 'HttpRequestError',
}
-export class HttpError extends ExtendableError {
+export class HttpError extends NamedError {
constructor(
public readonly message: string,
public readonly status: HttpStatus,
@@ -21,7 +21,7 @@ export class HttpError extends ExtendableError {
}
}
-export class HttpRequestError extends ExtendableError {
+export class HttpRequestError extends NamedError {
constructor(
public readonly request: HttpRequest,
public readonly error: Error,

MODIFY
packages/core/src/operators/act/act.operator.spec.ts
packages/core/src/operators/act/act.operator.spec.ts
@@ -2,7 +2,7 @@ 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 { ExtendableError } from '../../+internal/utils';
+import { NamedError } from '../../+internal/utils';
import { act } from './act.operator';
describe('#act operator', () => {
@@ -85,7 +85,7 @@ describe('#act operator', () => {
const effect$ = (event$: Observable<Event>) =>
event$.pipe(
act((event: any) => of(event).pipe(
- tap(e => { if (e.payload === 2) throw new ExtendableError(error.name, error.message) }),
+ tap(e => { if (e.payload === 2) throw new NamedError(error.name, error.message) }),
)),
);

MODIFY
packages/messaging/src/client/messaging.client.spec.ts
packages/messaging/src/client/messaging.client.spec.ts
import { TimeoutError } from 'rxjs';
-import { createContext } from '@marblejs/core';
+import { NamedError } from '@marblejs/core/dist/+internal/utils';
import { Transport, TransportMessage } from '../transport/transport.interface';
import { createAmqpStrategy } from '../transport/strategies/amqp.strategy';
+import { createTestContext } from '../util/messaging.test.util';
import { messagingClient } from './messaging.client';
import { MessagingClientConfig } from './messaging.client.interface';
@@ -18,7 +19,7 @@ describe('#messagingClient', () => {
},
};
- const runClient = async () => await messagingClient(clientOptions)(createContext());
+ const runClient = async () => await messagingClient(clientOptions)(createTestContext());
const runServer = () => createAmqpStrategy(clientOptions.options).connect({ isConsumer: true });
const createMessage = (data: any, correlationId?: string): TransportMessage<Buffer> => ({
data: Buffer.from(JSON.stringify(data)),
@@ -30,45 +31,67 @@ describe('#messagingClient', () => {
const server = await runServer();
server.message$.subscribe(async msg => {
- expect(msg.data.toString()).toEqual(JSON.stringify({ test: 'test' }));
+ expect(msg.data.toString()).toEqual(JSON.stringify({ type: 'TEST' }));
await server.close();
done();
});
- await client.emit({ test: 'test' });
+ await client.emit({ type: 'TEST' });
await client.close();
});
- test('sends command', async () => {
+ test('sends RPC event', async () => {
const client = await runClient();
const server = await runServer();
server.message$.subscribe(async msg => {
- const count = Number(msg.data.toString()) + 1;
- const response = createMessage(count, msg.correlationId);
+ const inEvent = JSON.parse(msg.data.toString());
+ const outEvent = { ...inEvent, payload: inEvent.payload + 1 };
+ const response = createMessage(outEvent, msg.correlationId);
await server.emitMessage(msg.replyTo || '', response);
await server.close();
});
- const result = await client.send(1).toPromise();
+ const result = await client.send({ type: 'TEST', payload: 1 }).toPromise();
- expect(result).toEqual(2);
+ expect(result).toEqual({ type: 'TEST', payload: 2 });
await client.close();
});
- test('timeouts sent command', async () => {
+ test('timeouts sent RPC event', async () => {
const client = await runClient();
const server = await runServer();
- const result = client.send(1).toPromise();
+ const result = client.send({ type: 'TEST' }).toPromise();
await expect(result).rejects.toEqual(new TimeoutError());
await client.close();
await server.close();
});
+
+ test('throws an error for sent RPC event', async () => {
+ const client = await runClient();
+ const server = await runServer();
+ const error = { name: 'TestErrorName', message: 'TestErrorMessage' };
+
+ server.message$.subscribe(async msg => {
+ const inEvent = JSON.parse(msg.data.toString());
+ const outEvent = { ...inEvent, error };
+ const response = createMessage(outEvent, msg.correlationId);
+
+ await server.emitMessage(msg.replyTo || '', response);
+ await server.close();
+ });
+
+ const result = client.send(1).toPromise();
+
+ await expect(result).rejects.toEqual(new NamedError(error.name, error.message));
+
+ await client.close();
+ });
});
});

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
import * as O from 'fp-ts/lib/Option';
-import * as R from 'fp-ts/lib/Reader';
import { pipe } from 'fp-ts/lib/pipeable';
-import { reader, HttpServerEventStreamToken, matchEvent, ServerEvent, AllServerEvents } from '@marblejs/core';
-import { from, Observable, EMPTY } from 'rxjs';
+import {
+ Event,
+ HttpServerEventStreamToken,
+ matchEvent,
+ ServerEvent,
+ AllServerEvents,
+ useContext,
+ LoggerToken,
+ LoggerTag,
+ LoggerLevel,
+ isEventError,
+ EventError,
+ createReader,
+} from '@marblejs/core';
+import { isNamedError, NamedError } from '@marblejs/core/dist/+internal/utils';
+import { from, Observable, EMPTY, throwError, of } from 'rxjs';
import { mergeMap, take, map, timeout } from 'rxjs/operators';
-import { TransportMessage, TransportLayerConnection, DEFAULT_TIMEOUT } from '../transport/transport.interface';
+import { TransportMessage, DEFAULT_TIMEOUT } from '../transport/transport.interface';
import { provideTransportLayer } from '../transport/transport.provider';
import { jsonTransformer } from '../transport/transport.transformer';
import { MessagingClient, MessagingClientConfig } from './messaging.client.interface';
@@ -16,49 +29,76 @@ export const messagingClient = (config: MessagingClientConfig) => {
msgTransformer = jsonTransformer,
} = config;
- const emit = (conn: TransportLayerConnection) => async <T>(msg: T) => {
- await conn.emitMessage(
- conn.getChannel(),
+ return createReader(async ask => {
+ const logger = useContext(LoggerToken)(ask);
+ const transportLayer = provideTransportLayer(transport, options);
+ const connection = await transportLayer.connect({ isConsumer: false });
+
+ const emit = async <T>(msg: T) => {
+ await connection.emitMessage(
+ connection.getChannel(),
{ data: msgTransformer.encode(msg as any) },
);
}
- const send = (conn: TransportLayerConnection) => <T, U extends Event>(msg: T): Observable<U> => {
- const channel = conn.getChannel();
+ const send = <T, U extends Event>(msg: T): Observable<U> => {
+ const channel = connection.getChannel();
const message: TransportMessage<Buffer> = { data: msgTransformer.encode(msg as any) };
- return from(conn.sendMessage(channel, message)).pipe(
+ const catchErrorEvent = <V extends Event>(e: V) => {
+ if (!e.error) {
+ return of(e);
+ }
+
+ if (isEventError(e.error)) {
+ return throwError(new EventError(e, e.error.message, e.error.data));
+ }
+
+ if (isNamedError(e.error)) {
+ return throwError(new NamedError(e.error.name, e.error.message));
+ }
+
+ const parsedError = JSON.stringify(e.error);
+
+ logger({
+ tag: LoggerTag.MESSAGING,
+ type: 'messagingClient',
+ message: `Caught an error with invalid structure: ${parsedError}`,
+ level: LoggerLevel.WARN,
+ })();
+
+ return throwError(new Error(parsedError));
+ }
+
+ return from(connection.sendMessage(channel, message)).pipe(
timeout(config.options.timeout || DEFAULT_TIMEOUT),
map(m => msgTransformer.decode(m.data) as U),
+ mergeMap(catchErrorEvent),
take(1),
);
}
- const close = (conn: TransportLayerConnection) => async () => {
- await conn.close();
+ const close = async () => {
+ await connection.close();
}
- const teardownOnClose$ = (conn: TransportLayerConnection) => (event$: Observable<AllServerEvents>) =>
+ const teardownOnClose = (event$: Observable<AllServerEvents>) =>
event$.pipe(
matchEvent(ServerEvent.close),
take(1),
- mergeMap(() => conn.close()),
+ mergeMap(() => connection.close()),
);
- return pipe(reader, R.map(async ask => {
- const transportLayer = provideTransportLayer(transport, options);
- const connection = await transportLayer.connect({ isConsumer: false });
-
pipe(
ask(HttpServerEventStreamToken),
- O.map(teardownOnClose$(connection)),
+ O.map(teardownOnClose),
O.getOrElse(() => EMPTY as Observable<any>),
).subscribe();
return {
- emit: emit(connection),
- send: send(connection),
- close: close(connection),
+ emit,
+ send,
+ close,
} as MessagingClient;
- }));
+ });
};

MODIFY
packages/messaging/src/effects/messaging.effects.helper.ts
packages/messaging/src/effects/messaging.effects.helper.ts
import { Event, EventMetadata } from '@marblejs/core';
-import { isString, ExtendableError } from '@marblejs/core/dist/+internal/utils';
+import { isString, NamedError } from '@marblejs/core/dist/+internal/utils';
-export class MissingEventTypeError extends ExtendableError {
+export class MissingEventTypeError extends NamedError {
constructor() {
super('MissingEventTypeError', `#reply - Missing type literal`);
}

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
-import { Event, matchEvent, use, combineEffects } from '@marblejs/core';
-import { wait } from '@marblejs/core/dist/+internal/utils';
+import { Event, matchEvent, use, 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 { delay, map, tap } from 'rxjs/operators';
import { MsgEffect } from '../effects/messaging.effects.interface';
import { runEventBus, runEventBusClient } from '../util/messaging.test.util';
@@ -71,6 +72,40 @@ describe('#eventBus', () => {
await eventBusClient.close();
});
+ test('handles parallel RPC with one errored event', async () => {
+ const rpc1$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('RPC_TEST'),
+ act(event => of(event).pipe(
+ use(eventValidator$(t.number)),
+ delay(10),
+ )),
+ );
+
+ const eventBus = await runEventBus(rpc1$);
+ const eventBusClient = await runEventBusClient();
+
+ const event1: Event = { type: 'RPC_TEST', payload: 1 };
+ const event2: Event = { type: 'RPC_TEST', payload: '2' };
+ const event3: Event = { type: 'RPC_TEST', payload: 3 };
+ const event4: Event = { type: 'RPC_TEST', payload: 4 };
+
+ const res1 = eventBusClient.send(event1).toPromise();
+ const res2 = eventBusClient.send(event2).toPromise();
+ const res3 = eventBusClient.send(event3).toPromise();
+ const res4 = eventBusClient.send(event4).toPromise();
+
+ await Promise.all([
+ expect(res1).resolves.toEqual(event1),
+ expect(res2).rejects.toBeInstanceOf(EventError),
+ expect(res3).resolves.toEqual(event3),
+ expect(res4).resolves.toEqual(event4),
+ ]);
+
+ await eventBus.close();
+ await eventBusClient.close();
+ });
+
test('handles published event', async done => {
const foo$: MsgEffect = event$ =>
event$.pipe(
@@ -92,4 +127,76 @@ describe('#eventBus', () => {
await eventBusClient.emit(event);
});
+
+ test('handles RPC event error with direct mapping', async () => {
+ const error = new NamedError('TestError_1', 'TestErrorMessage_1');
+
+ const rpc$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('RPC_TEST'),
+ act(
+ () => throwError(error),
+ (error, event) => reply(event)({
+ type: 'RPC_TEST_ERROR',
+ error: { name: error.name, message: error.message },
+ }),
+ ),
+ );
+
+ const eventBus = await runEventBus(rpc$);
+ const eventBusClient = await runEventBusClient();
+ const event: Event = { type: 'RPC_TEST' };
+
+ const result = eventBusClient.send(event).toPromise();
+ await expect(result).rejects.toEqual(error);
+
+ await eventBus.close();
+ await eventBusClient.close();
+ });
+
+ test('handles RPC event error with invalid direct mapping', async () => {
+ const error = new NamedError('TestError_2', 'TestErrorMessage_2');
+
+ const rpc$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('RPC_TEST'),
+ act(
+ () => throwError(error),
+ (error, event) => reply(event)({
+ type: 'RPC_TEST_ERROR',
+ error: { test: error.name },
+ }),
+ ),
+ );
+
+ const eventBus = await runEventBus(rpc$);
+ const eventBusClient = await runEventBusClient();
+ const event: Event = { type: 'RPC_TEST' };
+
+ const result = eventBusClient.send(event).toPromise();
+ await expect(result).rejects.toEqual(new Error(`{"test":"${error.name}"}`));
+
+ await eventBus.close();
+ await eventBusClient.close();
+ });
+
+ test('handles RPC event error with indirect mapping', async () => {
+ const error = new NamedError('TestError_3', 'TestErrorMessage_3');
+
+ const rpc$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('RPC_TEST'),
+ act(() => throwError(error)),
+ );
+
+ const eventBus = await runEventBus(rpc$);
+ const eventBusClient = await runEventBusClient();
+ const event: Event = { type: 'RPC_TEST' };
+
+ const result = eventBusClient.send(event).toPromise();
+ await expect(result).rejects.toEqual(error);
+
+ await eventBus.close();
+ await eventBusClient.close();
+ });
});

MODIFY
packages/messaging/src/transport/transport.error.ts
packages/messaging/src/transport/transport.error.ts
import { coreErrorFactory, CoreErrorOptions } from '@marblejs/core';
-import { ExtendableError } from '@marblejs/core/dist/+internal/utils';
+import { NamedError } from '@marblejs/core/dist/+internal/utils';
export enum ErrorType {
UNSUPPORTED_ERROR = 'UnsupportedError',
}
-export class UnsupportedError extends ExtendableError {
+export class UnsupportedError extends NamedError {
constructor(public readonly message: string) {
super(ErrorType.UNSUPPORTED_ERROR, message);
}

MODIFY
packages/messaging/src/util/messaging.test.util.ts
packages/messaging/src/util/messaging.test.util.ts
@@ -11,13 +11,12 @@ import { eventBus, eventBusClient } from '../eventbus/messaging.eventBus.reader'
// event bus
export const runEventBus =
async (effect$?: MsgEffect, output$?: MsgOutputEffect, error$?: MsgErrorEffect) => {
- const boundLogger = bindTo(LoggerToken)(mockLogger);
- const context = flow(register(boundLogger))(createContext());
+ const context = createTestContext();
const listener = messagingListener({ effects: effect$ ? [effect$] : [], output$, error$ });
return eventBus({ listener })(context);
};
-export const runEventBusClient = () => eventBusClient(createContext());
+export const runEventBusClient = () => eventBusClient(createTestContext());
// microservice
export const runMicroservice = (transport: any, options: any) =>
@@ -35,3 +34,7 @@ export const createMessage = (data: any): TransportMessage<Buffer> => ({
correlationId: uuid(),
});
+export const createTestContext = () => {
+ const boundLogger = bindTo(LoggerToken)(mockLogger);
+ return flow(register(boundLogger))(createContext());
+};

MODIFY
packages/middleware-io/src/io.error.ts
packages/middleware-io/src/io.error.ts
-import { ExtendableError } from '@marblejs/core/dist/+internal/utils';
+import { NamedError } from '@marblejs/core/dist/+internal/utils';
export enum ErrorType {
IO_ERROR = 'IOError',
}
-export class IOError extends ExtendableError {
+export class IOError extends NamedError {
constructor(
public readonly message: string,
public readonly data: object,

MODIFY
packages/websockets/src/error/websocket.error.model.ts
packages/websockets/src/error/websocket.error.model.ts
-import { ExtendableError } from '@marblejs/core/dist/+internal/utils';
+import { NamedError } from '@marblejs/core/dist/+internal/utils';
import { WebSocketStatus } from '../websocket.interface';
-export class WebSocketConnectionError extends ExtendableError {
+export class WebSocketConnectionError extends NamedError {
constructor(
public readonly message: string,
public readonly status: WebSocketStatus | number,

feat(messaging): messagingClient - detect error event response and throw exception