refactor(messaging): #messagingClient as async dependency
MODIFY
packages/@integration/src/messaging/client.ts
packages/@integration/src/messaging/client.ts
@@ -2,7 +2,6 @@ import { logger$ } from '@marblejs/middleware-logger';
import { messagingClient, MessagingClient, Transport } from '@marblejs/messaging';
import {
r,
- bindTo,
createServer,
createContextToken,
matchEvent,
@@ -11,6 +10,7 @@ import {
HttpServerEffect,
use,
useContext,
+ bindEagerlyTo,
} from '@marblejs/core';
import { merge, forkJoin } from 'rxjs';
import { tap, map, mergeMap, mapTo } from 'rxjs/operators';
@@ -81,7 +81,7 @@ export const server = createServer({
effects: [test$, fib$],
}),
dependencies: [
- bindTo(ClientToken)(client),
+ bindEagerlyTo(ClientToken)(client),
],
event$: (...args) => merge(
listening$(...args),

MODIFY
packages/messaging/src/client/messaging.client.interface.ts
packages/messaging/src/client/messaging.client.interface.ts
@@ -2,9 +2,9 @@ import { Observable } from 'rxjs';
import { TransportMessageTransformer, TransportStrategy } from '../transport/transport.interface';
export interface MessagingClient {
- emit: <T>(data: T) => Observable<boolean>;
send: <T, U>(data: T) => Observable<U>;
- close: () => Observable<any>;
+ emit: <T>(data: T) => Promise<void>;
+ close: () => Promise<void>;
}
type ConfigurationBase = {

MODIFY
packages/messaging/src/client/messaging.client.spec.ts
packages/messaging/src/client/messaging.client.spec.ts
@@ -16,7 +16,7 @@ describe('#messagingClient', () => {
},
};
- const runClient = () => messagingClient(clientOptions)(createContext());
+ const runClient = async () => await messagingClient(clientOptions)(createContext());
const runServer = () => createAmqpStrategy(clientOptions.options).connect({ isConsumer: true });
const createMessage = (data: any, correlationId?: string): TransportMessage<Buffer> => ({
data: Buffer.from(JSON.stringify(data)),
@@ -24,7 +24,7 @@ describe('#messagingClient', () => {
});
test('emits event to consumer', async done => {
- const client = runClient();
+ const client = await runClient();
const server = await runServer();
server.message$.subscribe(async msg => {
@@ -33,13 +33,12 @@ describe('#messagingClient', () => {
done();
});
- const result = await client.emit({ test: 'test' }).toPromise();
- expect(result).toEqual(true);
- await client.close().toPromise();
+ await client.emit({ test: 'test' });
+ await client.close();
});
test('sends command to consumer', async () => {
- const client = runClient();
+ const client = await runClient();
const server = await runServer();
server.message$.subscribe(async msg => {
@@ -52,7 +51,7 @@ describe('#messagingClient', () => {
const result = await client.send(1).toPromise();
expect(result).toEqual(2);
- await client.close().toPromise();
+ await client.close();
});
});

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
@@ -3,7 +3,7 @@ import * as R from 'fp-ts/lib/Reader';
import { pipe } from 'fp-ts/lib/pipeable';
import { reader, ServerEventStreamToken, matchEvent, ServerEvent, AllServerEvents } from '@marblejs/core';
import { from, Observable, EMPTY } from 'rxjs';
-import { mergeMap, take, map, mapTo, mergeMapTo } from 'rxjs/operators';
+import { mergeMap, take, map } from 'rxjs/operators';
import { TransportMessage, TransportLayerConnection } from '../transport/transport.interface';
import { provideTransportLayer } from '../transport/transport.provider';
import { jsonTransformer } from '../transport/transport.transformer';
@@ -16,44 +16,34 @@ export const messagingClient = (config: MessagingClientConfig) => {
msgTransformer = jsonTransformer,
} = config;
- const emit = (conn: Promise<TransportLayerConnection>) => <T>(msg: T) =>
- from(conn).pipe(
- mergeMap(c => c.emitMessage(
- c.getChannel(),
+ const emit = (conn: TransportLayerConnection) => async <T>(msg: T) => {
+ await conn.emitMessage(
+ conn.getChannel(),
{ data: msgTransformer.encode(msg as any) },
- )),
- mapTo(true),
- take(1),
);
+ }
- const send = (conn: Promise<TransportLayerConnection>) => <T, U>(msg: T): Observable<U> =>
- from(conn).pipe(
- mergeMap(c => c.sendMessage(
- c.getChannel(),
- { data: msgTransformer.encode(msg as any) },
- )),
+ const send = (conn: TransportLayerConnection) => <T, U>(msg: T): Observable<U> =>
+ from(conn.sendMessage(conn.getChannel(), { data: msgTransformer.encode(msg as any) })).pipe(
map(m => m as TransportMessage<Buffer>),
map(m => msgTransformer.decode(m.data) as U),
take(1),
);
- const close = (conn: Promise<TransportLayerConnection>) => () =>
- from(conn).pipe(
- mergeMap(c => c.close()),
- take(1),
- );
+ const close = (conn: TransportLayerConnection) => async () => {
+ await conn.close();
+ }
- const teardownOnClose$ = (conn: Promise<TransportLayerConnection>) => (event$: Observable<AllServerEvents>) =>
+ const teardownOnClose$ = (conn: TransportLayerConnection) => (event$: Observable<AllServerEvents>) =>
event$.pipe(
matchEvent(ServerEvent.close),
take(1),
- mergeMapTo(conn),
- mergeMap(conn => conn.close()),
+ mergeMap(() => conn.close()),
);
- return pipe(reader, R.map(ask => {
+ return pipe(reader, R.map(async ask => {
const transportLayer = provideTransportLayer(transport, options);
- const connection = transportLayer.connect({ isConsumer: false });
+ const connection = await transportLayer.connect({ isConsumer: false });
pipe(
ask(ServerEventStreamToken),

refactor(messaging): #messagingClient as async dependency