imp(messaging): shutdown client when httpServer is closed
MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
-import { reader } from '@marblejs/core';
-import { from, Observable } from 'rxjs';
-import { mergeMap, take, map, mapTo } from 'rxjs/operators';
+import { reader, HttpServerEventStreamToken, matchEvent, ServerEvent } from '@marblejs/core';
+import { from, Observable, EMPTY } from 'rxjs';
+import { mergeMap, take, map, mapTo, mergeMapTo } from 'rxjs/operators';
import {
Transport,
TransportMessage,
@@ -65,12 +65,21 @@ export const messagingClient = (config: MessagingClientConfig = {}) => {
take(1),
);
- return reader.map(() => {
+ return reader.map(ask => {
const transportLayer = provideTransportLayer(transport, options);
const conn = transportLayer
.then(server => server.connect())
.then(server => server.consumeResponse().then(() => server));
+ ask(HttpServerEventStreamToken).map(serverEvent$ =>
+ serverEvent$.pipe(
+ matchEvent(ServerEvent.close),
+ take(1),
+ mergeMapTo(conn),
+ mergeMap(conn => conn.close()),
+ ),
+ ).getOrElse(EMPTY).subscribe();
+
return {
emit: emit(conn),
send: send(conn),

MODIFY
packages/messaging/src/listener/messaging.listener.ts
packages/messaging/src/listener/messaging.listener.ts
import {
reader,
- ContextReader,
ContextProvider,
combineMiddlewares,
combineEffects,
createEffectMetadata,
} from '@marblejs/core';
-import { from, Observable, Subscription } from 'rxjs';
-import { map, mergeMap, publish, withLatestFrom, tap } from 'rxjs/operators';
+import { Observable, Subscription } from 'rxjs';
+import { map, publish, withLatestFrom } from 'rxjs/operators';
import {
Transport,
TransportMessage,
@@ -75,14 +74,15 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
effectsSub = subscribeEffects(message$);
};
- return (): ContextReader => reader.map(ask => {
+ return () => reader.map(ask => {
const transportLayer = provideTransportLayer(transport, options);
- from(transportLayer)
- .pipe(
- mergeMap(server => server.connect()),
- tap(conn => conn.consumeMessage()),
- )
- .subscribe(conn => handleConnection(conn, ask));
+ return transportLayer
+ .then(server => server.connect())
+ .then(conn => conn.consumeMessage().then(() => conn))
+ .then(conn => {
+ handleConnection(conn, ask);
+ return conn;
+ });
});
};

imp(messaging): shutdown client when httpServer is closed