feat(websockets): event handler
MODIFY
packages/core/src/http/server/http.server.interface.ts
packages/core/src/http/server/http.server.interface.ts
import * as https from 'https';
import { HttpServerEffect } from '../effects/http.effects.interface';
-import { BoundDependency } from '../../context/context.factory';
+import { ServerConfig } from '../../listener/listener.interface';
import { httpListener } from './http.server.listener';
export const DEFAULT_HOSTNAME = '127.0.0.1';
-export interface CreateServerConfig {
+type HttpListenerFn = ReturnType<typeof httpListener>;
+
+export interface CreateServerConfig extends ServerConfig<HttpServerEffect, HttpListenerFn> {
port?: number;
hostname?: string;
- listener: ReturnType<typeof httpListener>;
- event$?: HttpServerEffect;
options?: ServerOptions;
- dependencies?: BoundDependency<any>[];
}
export interface ServerOptions {

MODIFY
packages/core/src/listener/listener.interface.ts
packages/core/src/listener/listener.interface.ts
import { IO } from 'fp-ts/lib/IO';
import { Reader } from 'fp-ts/lib/Reader';
import { Effect } from '../effects/effects.interface';
-import { Context } from '../context/context.factory';
+import { Context, BoundDependency } from '../context/context.factory';
export interface ListenerConfig<T = any> {
effects?: any[];
@@ -12,10 +12,22 @@ export interface ListenerConfig<T = any> {
export type ListenerHandler = (...args: any[]) => void;
-export interface Listener<T extends ListenerConfig, U extends ListenerHandler> {
+export interface Listener<
+ T extends ListenerConfig = ListenerConfig,
+ U extends ListenerHandler = ListenerHandler,
+> {
(config?: T): Reader<Context, U>;
}
export interface ServerIO<T> extends IO<Promise<T>> {
context: Context;
}
+
+export interface ServerConfig<
+ T extends Effect<any, any, any>,
+ U extends ReturnType<Listener> = ReturnType<Listener>,
+> {
+ event$?: T;
+ listener: U;
+ dependencies?: BoundDependency<any>[];
+}

MODIFY
packages/messaging/src/server/messaging.server.interface.ts
packages/messaging/src/server/messaging.server.interface.ts
-import { BoundDependency, ServerIO } from '@marblejs/core';
+import { ServerConfig, ServerIO } from '@marblejs/core';
import { MsgServerEffect } from '../effects/messaging.effects.interface';
import { TransportStrategy, TransportLayerConnection } from '../transport/transport.interface';
import { messagingListener } from './messaging.server.listener';
-type ConfigurationBase = {
- listener: ReturnType<typeof messagingListener>;
- dependencies?: BoundDependency<any>[];
- event$?: MsgServerEffect;
-}
+type MessagingListenerFn = ReturnType<typeof messagingListener>;
+type ConfigurationBase = ServerConfig<MsgServerEffect, MessagingListenerFn>;
export type CreateMicroserviceConfig =
& TransportStrategy

MODIFY
packages/websockets/src/effects/websocket.effects.interface.ts
packages/websockets/src/effects/websocket.effects.interface.ts
@@ -15,6 +15,9 @@ export interface WsConnectionEffect<
T extends http.IncomingMessage = http.IncomingMessage
> extends WsEffect<T, T, undefined> {}
+export interface WsServerEffect<T extends Event = Event>
+ extends WsEffect<T, any, undefined> {}
+
export interface WsOutputEffect<
T extends Event = Event
> extends WsEffect<T, Event> {}

ADD
packages/websockets/src/server/websocket.server.event.subscriber.ts
+import * as http from 'http';
+import * as WebSocket from 'ws';
+import { Subject, Observable } from 'rxjs';
+import { ServerEventType, ServerEvent, AllServerEvents } from './websocket.server.event';
+import { DEFAULT_HOSTNAME, WebSocketServerConnection } from './websocket.server.interface';
+
+export const subscribeWebSocketEvents = (server: WebSocketServerConnection): {
+ serverEvent$: Observable<AllServerEvents>;
+ serverEventSubject: Subject<AllServerEvents>;
+} => {
+ const serverEventSubject = new Subject<AllServerEvents>();
+
+ server.on(ServerEventType.HEADERS, (headers: string[], req: http.IncomingMessage) =>
+ serverEventSubject.next(ServerEvent.headers(headers, req)),
+ );
+
+ server.on(ServerEventType.CLOSE, () =>
+ serverEventSubject.next(ServerEvent.close()),
+ );
+
+ server.on(ServerEventType.ERROR, (error: Error) =>
+ serverEventSubject.next(ServerEvent.error(error)),
+ );
+
+ server.on(ServerEventType.LISTENING, () => {
+ const hostname = server.options.host ?? DEFAULT_HOSTNAME;
+ const port = (server.address() as WebSocket.AddressInfo).port;
+ serverEventSubject.next(ServerEvent.listening(port, hostname));
+ });
+
+ return {
+ serverEvent$: serverEventSubject.asObservable(),
+ serverEventSubject,
+ };
+};

ADD
packages/websockets/src/server/websocket.server.event.ts
+import * as http from 'http';
+import { createEvent, EventsUnion, Event } from '@marblejs/core';
+import { WebSocketClientConnection } from './websocket.server.interface';
+
+export enum ServerEventType {
+ LISTENING = 'listening',
+ HEADERS = 'headers',
+ CONNECTION = 'connection',
+ CLOSE = 'close',
+ ERROR = 'error',
+}
+
+export const ServerEvent = {
+ listening: createEvent(
+ ServerEventType.LISTENING,
+ (port: number, host: string) => ({ port, host }),
+ ),
+ headers: createEvent(
+ ServerEventType.HEADERS,
+ (headers: string[], req: http.IncomingMessage) => ({ headers, req }),
+ ),
+ connection: createEvent(
+ ServerEventType.CONNECTION,
+ (client: WebSocketClientConnection, req: http.IncomingMessage) => ({ client, req }),
+ ),
+ close: createEvent(
+ ServerEventType.CLOSE,
+ ),
+ error: createEvent(
+ ServerEventType.ERROR,
+ (error: Error) => ({ error }),
+ )
+};
+
+export type AllServerEvents = EventsUnion<typeof ServerEvent>;
+
+export function isListeningEvent(event: Event): event is ReturnType<typeof ServerEvent.listening> {
+ return event.type === ServerEventType.LISTENING;
+}
+
+export function isHeadersEvent(event: Event): event is ReturnType<typeof ServerEvent.headers> {
+ return event.type === ServerEventType.HEADERS;
+}
+
+export function isConnectionEvent(event: Event): event is ReturnType<typeof ServerEvent.connection> {
+ return event.type === ServerEventType.CONNECTION;
+}
+
+export function isCloseEvent(event: Event): event is ReturnType<typeof ServerEvent.close> {
+ return event.type === ServerEventType.CLOSE;
+}
+
+export function isErrorEvent(event: Event): event is ReturnType<typeof ServerEvent.error> {
+ return event.type === ServerEventType.ERROR;
+}

MODIFY
packages/websockets/src/server/websocket.server.interface.ts
packages/websockets/src/server/websocket.server.interface.ts
import * as WebSocket from 'ws';
import { Observable } from 'rxjs';
-import { BoundDependency, Event } from '@marblejs/core';
-import { WsConnectionEffect } from '../effects/websocket.effects.interface';
+import { Event, ServerConfig } from '@marblejs/core';
+import { WsConnectionEffect, WsServerEffect } from '../effects/websocket.effects.interface';
import { webSocketListener } from './websocket.server.listener';
-export interface WebSocketServerConfig {
+export const DEFAULT_HOSTNAME = '127.0.0.1';
+
+type WebSocketListenerFn = ReturnType<typeof webSocketListener>;
+
+export interface WebSocketServerConfig extends ServerConfig<WsServerEffect, WebSocketListenerFn> {
options?: WebSocket.ServerOptions;
- listener: ReturnType<typeof webSocketListener>;
- dependencies?: BoundDependency<any>[];
connection$?: WsConnectionEffect;
}

feat(websockets): event handler