test(messaging): #messagingClient - timeout
MODIFY
packages/@integration/src/messaging/client.ts
packages/@integration/src/messaging/client.ts
import { logger$ } from '@marblejs/middleware-logger';
+import { requestValidator$, t } from '@marblejs/middleware-io';
import { messagingClient, MessagingClient, Transport } from '@marblejs/messaging';
import {
r,
@@ -14,7 +15,6 @@ import {
} from '@marblejs/core';
import { merge, forkJoin } from 'rxjs';
import { tap, map, mergeMap, mapTo } from 'rxjs/operators';
-import { requestValidator$, t } from '@marblejs/middleware-io';
const port = process.env.PORT
? Number(process.env.PORT)
@@ -37,22 +37,30 @@ const rootValiadtor$ = requestValidator$({
}),
});
-const test$ = r.pipe(
- r.matchPath('/test'),
+const buffer$ = r.pipe(
+ r.matchPath('/buffer'),
r.matchType('GET'),
- r.useEffect((req$, { ask }) => req$.pipe(
- mergeMap(() => useContext(ClientToken)(ask).emit({ type: 'TEST' })),
+ r.useEffect((req$, { ask }) => {
+ const client = useContext(ClientToken)(ask);
+
+ return req$.pipe(
+ mergeMap(() => client.emit({ type: 'BUFFER' })),
mapTo(({ body: 'OK' })),
- )),
+ );
+ }),
);
const timeout$ = r.pipe(
r.matchPath('/timeout'),
r.matchType('GET'),
- r.useEffect((req$, { ask }) => req$.pipe(
- mergeMap(() => useContext(ClientToken)(ask).send({ type: 'TIMEOUT' })),
+ r.useEffect((req$, { ask }) => {
+ const client = useContext(ClientToken)(ask);
+
+ return req$.pipe(
+ mergeMap(() => client.send({ type: 'TIMEOUT' })),
mapTo(({ body: 'OK' })),
- )),
+ );
+ }),
);
const fib$ = r.pipe(
@@ -87,7 +95,7 @@ export const server = createServer({
port,
httpListener: httpListener({
middlewares: [logger$()],
- effects: [test$, timeout$, fib$],
+ effects: [buffer$, timeout$, fib$],
}),
dependencies: [
bindEagerlyTo(ClientToken)(client),

MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
@@ -19,11 +19,11 @@ const fibonacci$: MsgEffect = event$ =>
)),
);
-const test$: MsgEffect = event$ =>
+const buffer$: MsgEffect = event$ =>
event$.pipe(
- matchEvent('TEST'),
+ matchEvent('BUFFER'),
bufferCount(2),
- mapTo({ type: 'TEST_RESULT' }),
+ mapTo({ type: 'BUFFER_RESULT' }),
);
const timeout$: MsgEffect = event$ =>
@@ -43,7 +43,7 @@ export const microservice = createMicroservice({
queueOptions: { durable: false },
},
messagingListener: messagingListener({
- effects: [fibonacci$, test$, timeout$],
+ effects: [fibonacci$, buffer$, timeout$],
}),
});

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 { Transport, TransportMessage } from '../transport/transport.interface';
import { createAmqpStrategy } from '../transport/strategies/amqp.strategy';
@@ -13,6 +14,7 @@ describe('#messagingClient', () => {
host: 'amqp://localhost:5672',
queue: 'test_queue_client',
queueOptions: { durable: false },
+ timeout: 1000,
},
};
@@ -23,7 +25,7 @@ describe('#messagingClient', () => {
correlationId: correlationId,
});
- test('emits event to consumer', async done => {
+ test('emits event', async done => {
const client = await runClient();
const server = await runServer();
@@ -37,7 +39,7 @@ describe('#messagingClient', () => {
await client.close();
});
- test('sends command to consumer', async () => {
+ test('sends command', async () => {
const client = await runClient();
const server = await runServer();
@@ -50,9 +52,23 @@ describe('#messagingClient', () => {
});
const result = await client.send(1).toPromise();
+
expect(result).toEqual(2);
+
await client.close();
});
+
+ test('timeouts sent command', async () => {
+ const client = await runClient();
+ const server = await runServer();
+
+ const result = client.send(1).toPromise();
+
+ await expect(result).rejects.toEqual(new TimeoutError());
+
+ await client.close();
+ await server.close();
+ });
});
});

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
@@ -23,13 +23,16 @@ export const messagingClient = (config: MessagingClientConfig) => {
);
}
- const send = (config: TransportLayerConfig, conn: TransportLayerConnection) => <T, U>(msg: T): Observable<U> =>
- from(conn.sendMessage(conn.getChannel(), { data: msgTransformer.encode(msg as any) })).pipe(
+ const send = (config: TransportLayerConfig, conn: TransportLayerConnection) => <T, U>(msg: T): Observable<U> => {
+ const channel = conn.getChannel();
+ const message: TransportMessage<Buffer> = { data: msgTransformer.encode(msg as any) };
+
+ return from(conn.sendMessage(channel, message)).pipe(
timeout(config.timeout),
- map(m => m as TransportMessage<Buffer>),
map(m => msgTransformer.decode(m.data) as U),
take(1),
);
+ }
const close = (conn: TransportLayerConnection) => async () => {
await conn.close();

test(messaging): #messagingClient - timeout