feat(websockets): integration with core Logger dependency
MODIFY
packages/websockets/src/effects/websocket.effects.interface.ts
packages/websockets/src/effects/websocket.effects.interface.ts
@@ -8,10 +8,8 @@ export interface WsMiddlewareEffect<
> extends WsEffect<I, O> {}
export interface WsErrorEffect<
- T extends Error = Error,
- U = Event,
- V = Event
-> extends WsEffect<{ event?: U; error: T }, V> {}
+ Err extends Error = Error,
+> extends WsEffect<Err, Event> {}
export interface WsConnectionEffect<
T extends http.IncomingMessage = http.IncomingMessage

MODIFY
packages/websockets/src/error/websocket.error.effect.ts
packages/websockets/src/error/websocket.error.effect.ts
-import { EventError } from '@marblejs/core';
import { map } from 'rxjs/operators';
import { WsErrorEffect } from '../effects/websocket.effects.interface';
-const DEFAULT_ERROR_CHANNEL = 'ERROR';
-
-const errorFactory = (message: string | undefined, data: any | undefined) => ({
- message, data,
-});
-
-export const defaultError$: WsErrorEffect<EventError> = event$ =>
+export const defaultError$: WsErrorEffect = event$ =>
event$.pipe(
- map(({ error, event }) => ({
- type: event ? event.type : DEFAULT_ERROR_CHANNEL,
- error: errorFactory(
- error ? error.message : undefined,
- error ? error.data : undefined,
- )
+ map(error => ({
+ type: 'UNHANDLED_ERROR',
+ error: {
+ name: error.name,
+ message: error.message,
+ },
})),
);

MODIFY
packages/websockets/src/server/websocket.server.listener.ts
packages/websockets/src/server/websocket.server.listener.ts
-import { Observable, fromEvent, of } from 'rxjs';
-import { map, takeUntil } from 'rxjs/operators';
+import { Observable, fromEvent, of, Subject } from 'rxjs';
+import { map, takeUntil, catchError } from 'rxjs/operators';
import {
combineEffects,
combineMiddlewares,
@@ -7,6 +7,11 @@ import {
createListener,
Event,
EventError,
+ EffectContext,
+ LoggerTag,
+ useContext,
+ LoggerToken,
+ LoggerLevel,
} from '@marblejs/core';
import { pipe } from 'fp-ts/lib/pipeable';
import { WsEffect, WsErrorEffect, WsMiddlewareEffect, WsOutputEffect } from '../effects/websocket.effects.interface';
@@ -18,7 +23,7 @@ import { WebSocketClientConnection } from './websocket.server.interface';
export interface WebSocketListenerConfig {
effects?: WsEffect<any, any>[];
middlewares?: WsMiddlewareEffect<any, any>[];
- error$?: WsErrorEffect<Error, any, any>;
+ error$?: WsErrorEffect;
eventTransformer?: EventTransformer<Event, any>;
output$?: WsOutputEffect;
}
@@ -40,34 +45,68 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
eventTransformer = jsonTransformer as EventTransformer<Event, any>,
} = config ?? {};
- const middleware$ = combineMiddlewares(...middlewares);
- const effect$ = combineEffects(...effects);
+ const logger = useContext(LoggerToken)(ask);
+ const combinedMiddlewares = combineMiddlewares(...middlewares); // @TODO: create inputLogger$
+ const combinedEffects = combineEffects(...effects);
+
+ const processError$ = (ctx: EffectContext<WebSocketClientConnection>) => (error: Error) =>
+ pipe(
+ of(error),
+ e$ => error$(e$, ctx),
+ // e$ => errorLogger$(e$, ctx), // @TODO
+ );
const handle = (client: WebSocketClientConnection) => {
+ const eventSubject = new Subject<Event>();
const ctx = createEffectContext({ ask, client });
const close$ = fromEvent(client, 'close');
const message$ = fromEvent<MessageEvent>(client, 'message');
- const stream = pipe(
+ const incomingEvent$ = pipe(
message$,
e$ => e$.pipe(map(e => eventTransformer.decode(e.data))),
- e$ => middleware$(e$, ctx),
- e$ => effect$(e$, ctx),
+ e$ => combinedMiddlewares(e$, ctx),
+ e$ => e$.pipe(catchError(processError$(ctx))),
+ );
+
+ const outgoingEvent$ = pipe(
+ eventSubject.asObservable(),
+ e$ => combinedEffects(e$, ctx),
e$ => output$(e$, ctx),
+ // e$ => outputLogger$(e$, ctx), // @TODO
+ e$ => e$.pipe(catchError(processError$(ctx))),
+ );
+
+ const subscribeIncomingEvent = (input$: Observable<Event>) =>
+ input$
+ .pipe(takeUntil(close$))
+ .subscribe(
+ (event: Event) => eventSubject.next(event),
+ (error: EventError) => {
+ const type = 'ServerListener';
+ const message = `Unexpected error for IncomingEvent stream: "${error.name}", "${error.message}"`;
+ logger({ tag: LoggerTag.WEBSOCKETS, type, message, level: LoggerLevel.ERROR })();
+ subscribeIncomingEvent(input$);
+ },
+ () => subscribeIncomingEvent(input$),
);
- const subscribe = (input$: Observable<Event>) =>
+ const subscribeOutgoingEvent = (input$: Observable<Event>) =>
input$
.pipe(takeUntil(close$))
.subscribe(
(event: Event) => client.sendResponse(event),
(error: EventError) => {
- error$(of({ event: error.event, error }), ctx).subscribe(client.sendResponse);
- subscribe(stream);
+ const type = 'ServerListener';
+ const message = `Unexpected error OutgoingEvent stream: "${error.name}", "${error.message}"`;
+ logger({ tag: LoggerTag.WEBSOCKETS, type, message, level: LoggerLevel.ERROR })();
+ subscribeOutgoingEvent(input$);
},
+ () => subscribeOutgoingEvent(input$),
);
- subscribe(stream);
+ subscribeIncomingEvent(incomingEvent$);
+ subscribeOutgoingEvent(outgoingEvent$);
};
handle.eventTransformer = eventTransformer;

MODIFY
packages/websockets/src/server/websocket.server.ts
packages/websockets/src/server/websocket.server.ts
@@ -3,7 +3,22 @@ import * as WebSocket from 'ws';
import { Observable, of } from 'rxjs';
import { map } from 'rxjs/operators';
import { flow } from 'fp-ts/lib/function';
-import { lookup, registerAll, createContext, createEffectContext, Context, resolve, ServerIO } from '@marblejs/core';
+import {
+ bindTo,
+ Context,
+ createContext,
+ createEffectContext,
+ registerAll,
+ resolve,
+ ServerIO,
+ mockLogger,
+ LoggerToken,
+ LoggerTag,
+ lookup,
+ logger,
+ logContext,
+} from '@marblejs/core';
+import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
import { createServer, handleServerBrokenConnections, handleClientBrokenConnection } from '../server/websocket.server.helper';
import { handleBroadcastResponse, handleResponse } from '../response/websocket.response.handler';
import { WebSocketConnectionError } from '../error/websocket.error.model';
@@ -26,8 +41,14 @@ export const createWebSocketServer = async (config: WebSocketServerConfig) => {
);
};
+ const boundLogger = bindTo(LoggerToken)(isTestEnv() ? mockLogger : logger);
+
const context = await flow(
- registerAll([ ...dependencies ]),
+ registerAll([
+ boundLogger,
+ ...dependencies,
+ ]),
+ logContext(LoggerTag.WEBSOCKETS),
resolve,
)(createContext());
@@ -40,11 +61,15 @@ export const createWebSocketServer = async (config: WebSocketServerConfig) => {
}, listener.eventTransformer);
const listen: ServerIO<WebSocketServerConnection> = async () => {
+ // @TODO: log message that WebSocket server started listening
+
server.on('connection', (client: WebSocketClientConnection) => {
client.sendResponse = handleResponse(client, listener.eventTransformer);
client.sendBroadcastResponse = handleBroadcastResponse(server, listener.eventTransformer);
client.isAlive = true;
+ // @TODO: log message that there is an incoming WebSocket connection
+
handleClientBrokenConnection(client).subscribe();
listener(client);
});

feat(websockets): integration with core Logger dependency