fix(core): rxjs operator order
MODIFY
packages/core/src/http/router/http.router.resolver.v2.ts
packages/core/src/http/router/http.router.resolver.v2.ts
@@ -37,17 +37,16 @@ export const resolveRouting = (
const errorSubject = new Subject<{ error: Error; req: HttpRequest }>();
const outputFlow$ = outputSubject.asObservable().pipe(
- takeUntil(close$),
mergeMap(data => {
const stream = output$ ? output$(of(data), ctx) : of(data.res);
return stream.pipe(
map(res => ([res, data.req] as [HttpEffectResponse, HttpRequest])),
);
}),
+ takeUntil(close$),
);
const errorFlow$ = errorSubject.asObservable().pipe(
- takeUntil(close$),
map(data => isHttpRequestError(data.error) ? { ...data, error: data.error.error } : data),
mergeMap(data => {
const stream = error$ ? error$(of(data), ctx) : defaultError$(of(data), ctx);
@@ -55,6 +54,7 @@ export const resolveRouting = (
map(res => ([res, data.req] as [HttpEffectResponse, HttpRequest])),
);
}),
+ takeUntil(close$),
);
const subscribeOutput = (stream$: Observable<[HttpEffectResponse, HttpRequest]>) =>
@@ -93,10 +93,8 @@ export const resolveRouting = (
e$ => middleware(e$, ctx),
e$ => decorate ? decorateEffect(e$, errorSubject) : e$,
e$ => effect(e$, ctx),
- e$ => e$.pipe(
- takeUntil(close$),
catchError((error, stream) => processError(stream)(error)),
- ),
+ takeUntil(close$),
);
const processError = (originStream$: Observable<any>) => (error: any) => {
@@ -118,7 +116,7 @@ export const resolveRouting = (
},
() => {
const type = 'RouterResolver';
- const message = `Output stream completes`;
+ const message = `Effect stream completes`;
logger({ tag: LoggerTag.HTTP, type, message, level: LoggerLevel.DEBUG })();
}
);

MODIFY
packages/core/src/http/server/http.server.event.subscriber.ts
packages/core/src/http/server/http.server.event.subscriber.ts
import * as http from 'http';
import * as net from 'net';
import { Subject, Observable } from 'rxjs';
-import { takeWhile } from 'rxjs/operators';
import { HttpServer } from '../http.interface';
-import { ServerEventType, ServerEvent, AllServerEvents, isCloseEvent } from './http.server.event';
+import { ServerEventType, ServerEvent, AllServerEvents } from './http.server.event';
import { DEFAULT_HOSTNAME } from './http.server.interface';
export const subscribeServerEvents = (hostname?: string) => (httpServer: HttpServer): Observable<AllServerEvents> => {
@@ -50,7 +49,5 @@ export const subscribeServerEvents = (hostname?: string) => (httpServer: HttpSer
event$.next(ServerEvent.listening(serverAddressInfo.port, hostname ?? DEFAULT_HOSTNAME));
});
- return event$
- .asObservable()
- .pipe(takeWhile(e => !isCloseEvent(e), true));
+ return event$.asObservable();
};

MODIFY
packages/core/src/http/server/http.server.ts
packages/core/src/http/server/http.server.ts
@@ -2,6 +2,7 @@ import * as http from 'http';
import * as https from 'https';
import { flow } from 'fp-ts/lib/function';
import { Subject, merge, EMPTY } from 'rxjs';
+import { takeWhile } from 'rxjs/operators';
import { createContext, lookup, registerAll, bindTo, resolve } from '../../context/context.factory';
import { createEffectContext } from '../../effects/effectsContext.factory';
import { isTestingMetadataOn } from '../../+internal/testing';
@@ -15,6 +16,7 @@ import { subscribeServerEvents } from './http.server.event.subscriber';
import { HttpServerClientToken, HttpServerEventStreamToken, HttpRequestMetadataStorageToken, HttpRequestBusToken } from './http.server.tokens';
import { httpRequestMetadataStorage } from './http.server.metadata.storage';
import { CreateServerConfig } from './http.server.interface';
+import { isCloseEvent } from './http.server.event';
export const createServer = async (config: CreateServerConfig) => {
const { listener, event$, port, hostname, dependencies = [], options = {} } = config;
@@ -50,6 +52,8 @@ export const createServer = async (config: CreateServerConfig) => {
listening$(serverEvent$, ctx),
error$(serverEvent$, ctx),
close$(serverEvent$, ctx),
+ ).pipe(
+ takeWhile(e => !isCloseEvent(e), true)
).subscribe();
const listen: ServerIO<HttpServer> = () => new Promise((resolve, reject) => {

fix(core): rxjs operator order