feat(messaging): throw an error when transport method is unsupported
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, use, createEvent } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { map, tap } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
-import { MsgEffect } from '../../effects/messaging.effects.interface';
+import { MsgEffect, MsgErrorEffect } from '../../effects/messaging.effects.interface';
import { RedisStrategyOptions } from '../../transport/strategies/redis.strategy.interface';
import { runClient, runServer, createMessage, close } from '../../util/messaging.test.util';
@@ -12,6 +12,10 @@ const createOptions = (config: { channel?: string } = {}): RedisStrategyOptions
});
describe('messagingServer::Redis', () => {
+ beforeEach(() => {
+ jest.spyOn(console, 'error').mockImplementation(() => jest.fn());
+ });
+
test('handles RPC event', async () => {
const rpc$: MsgEffect = event$ =>
event$.pipe(
@@ -55,4 +59,31 @@ describe('messagingServer::Redis', () => {
expect(emitResult).toEqual(true);
});
+
+ test('throws an UnsupportedError for unsupported "ackMessage/nackMessage"', async done => {
+ const event$: MsgEffect = (event$, ctx) =>
+ event$.pipe(
+ matchEvent('EVENT_TEST'),
+ tap(event => ctx.client.ackMessage(event.raw)),
+ );
+
+ const error$: MsgErrorEffect = event$ =>
+ event$.pipe(
+ tap(({ event, error }) => {
+ expect(event).toBeUndefined();
+ expect(error).toEqual(error);
+ close(server)(done);
+ }),
+ map(() => createEvent('ERROR')),
+ );
+
+ const options = createOptions();
+ const client = await runClient(Transport.REDIS, options);
+ const server = await runServer(Transport.REDIS, options)(event$, error$);
+ const message = createMessage({ type: 'EVENT_TEST' });
+
+ const emitResult = await client.emitMessage(options.channel, message);
+
+ expect(emitResult).toEqual(true);
+ });
});

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
@@ -5,6 +5,7 @@ import { RedisClient, ClientOpts } from 'redis';
import { TransportLayer, TransportLayerConnection, TransportMessage } from '../transport.interface';
import { RedisStrategyOptions, RedisConnectionStatus } from './redis.strategy.interface';
import { quitRedisClient, subscribeRedisChannel, encodeMessage, decodeMessage, unsubscribeRedisChannel, connectRedisClient } from './redis.strategy.helper';
+import { throwUnsupportedError } from '../transport.error';
class RedisStrategyConnection implements TransportLayerConnection {
private msgSubject$ = new Subject<{ content: string }>();
@@ -87,14 +88,6 @@ class RedisStrategyConnection implements TransportLayerConnection {
).toPromise();
}
- ackMessage = () => {
- throw new Error('Unsupported'); // @TODO
- }
-
- nackMessage = () => {
- throw new Error('Unsupported'); // @TODO
- }
-
close = async () => {
await Promise.all([
quitRedisClient(this.publisher),
@@ -104,6 +97,10 @@ class RedisStrategyConnection implements TransportLayerConnection {
this.closeSubject$.next();
}
+ ackMessage = () => throwUnsupportedError('Redis')('ackMessage');
+
+ nackMessage = () => throwUnsupportedError('Redis')('nackMessage');
+
getChannel = () => this.opts.channel;
}

ADD
packages/messaging/src/transport/transport.error.ts
+import { coreErrorFactory, CoreErrorOptions } from '@marblejs/core';
+import { ExtendableError } from '@marblejs/core/dist/+internal/utils';
+
+export enum ErrorType {
+ UNSUPPORTED_ERROR = 'UnsupportedError',
+}
+
+export class UnsupportedError extends ExtendableError {
+ constructor(public readonly message: string) {
+ super(ErrorType.UNSUPPORTED_ERROR, message);
+ }
+}
+
+export const throwUnsupportedError = (transportName: string) => (method: string) => {
+ const message = `Unsupported operation.`;
+ const detail = `Method "${method}" is unsupported for ${transportName} transport layer.`;
+ const error = new UnsupportedError(`${message} ${detail}`);
+ const coreErrorOptions: CoreErrorOptions = { contextMethod: method, offset: 1 };
+ const coreError = coreErrorFactory(error.message, coreErrorOptions)
+
+ console.error(coreError.stack);
+
+ throw error;
+};

feat(messaging): throw an error when transport method is unsupported