1 | "use strict";
|
2 | Object.defineProperty(exports, "__esModule", { value: true });
|
3 | exports.WebSocketsController = void 0;
|
4 | const logger_service_1 = require("@nestjs/common/services/logger.service");
|
5 | const metadata_scanner_1 = require("@nestjs/core/metadata-scanner");
|
6 | const rxjs_1 = require("rxjs");
|
7 | const operators_1 = require("rxjs/operators");
|
8 | const constants_1 = require("./constants");
|
9 | const invalid_socket_port_exception_1 = require("./errors/invalid-socket-port.exception");
|
10 | const gateway_metadata_explorer_1 = require("./gateway-metadata-explorer");
|
11 | const compare_element_util_1 = require("./utils/compare-element.util");
|
12 | class WebSocketsController {
|
13 | constructor(socketServerProvider, config, contextCreator, graphInspector, appOptions = {}) {
|
14 | this.socketServerProvider = socketServerProvider;
|
15 | this.config = config;
|
16 | this.contextCreator = contextCreator;
|
17 | this.graphInspector = graphInspector;
|
18 | this.appOptions = appOptions;
|
19 | this.logger = new logger_service_1.Logger(WebSocketsController.name, {
|
20 | timestamp: true,
|
21 | });
|
22 | this.metadataExplorer = new gateway_metadata_explorer_1.GatewayMetadataExplorer(new metadata_scanner_1.MetadataScanner());
|
23 | }
|
24 | connectGatewayToServer(instance, metatype, moduleKey, instanceWrapperId) {
|
25 | const options = Reflect.getMetadata(constants_1.GATEWAY_OPTIONS, metatype) || {};
|
26 | const port = Reflect.getMetadata(constants_1.PORT_METADATA, metatype) || 0;
|
27 | if (!Number.isInteger(port)) {
|
28 | throw new invalid_socket_port_exception_1.InvalidSocketPortException(port, metatype);
|
29 | }
|
30 | this.subscribeToServerEvents(instance, options, port, moduleKey, instanceWrapperId);
|
31 | }
|
32 | subscribeToServerEvents(instance, options, port, moduleKey, instanceWrapperId) {
|
33 | const nativeMessageHandlers = this.metadataExplorer.explore(instance);
|
34 | const messageHandlers = nativeMessageHandlers.map(({ callback, message, methodName }) => ({
|
35 | message,
|
36 | methodName,
|
37 | callback: this.contextCreator.create(instance, callback, moduleKey, methodName),
|
38 | }));
|
39 | this.inspectEntrypointDefinitions(instance, port, messageHandlers, instanceWrapperId);
|
40 | if (this.appOptions.preview) {
|
41 | return;
|
42 | }
|
43 | const observableServer = this.socketServerProvider.scanForSocketServer(options, port);
|
44 | this.assignServerToProperties(instance, observableServer.server);
|
45 | this.subscribeEvents(instance, messageHandlers, observableServer);
|
46 | }
|
47 | subscribeEvents(instance, subscribersMap, observableServer) {
|
48 | const { init, disconnect, connection, server } = observableServer;
|
49 | const adapter = this.config.getIoAdapter();
|
50 | this.subscribeInitEvent(instance, init);
|
51 | this.subscribeConnectionEvent(instance, connection);
|
52 | this.subscribeDisconnectEvent(instance, disconnect);
|
53 | const handler = this.getConnectionHandler(this, instance, subscribersMap, disconnect, connection);
|
54 | adapter.bindClientConnect(server, handler);
|
55 | this.printSubscriptionLogs(instance, subscribersMap);
|
56 | }
|
57 | getConnectionHandler(context, instance, subscribersMap, disconnect, connection) {
|
58 | const adapter = this.config.getIoAdapter();
|
59 | return (...args) => {
|
60 | const [client] = args;
|
61 | connection.next(args);
|
62 | context.subscribeMessages(subscribersMap, client, instance);
|
63 | const disconnectHook = adapter.bindClientDisconnect;
|
64 | disconnectHook &&
|
65 | disconnectHook.call(adapter, client, () => disconnect.next(client));
|
66 | };
|
67 | }
|
68 | subscribeInitEvent(instance, event) {
|
69 | if (instance.afterInit) {
|
70 | event.subscribe(instance.afterInit.bind(instance));
|
71 | }
|
72 | }
|
73 | subscribeConnectionEvent(instance, event) {
|
74 | if (instance.handleConnection) {
|
75 | event
|
76 | .pipe((0, operators_1.distinctUntilChanged)((prev, curr) => (0, compare_element_util_1.compareElementAt)(prev, curr, 0)))
|
77 | .subscribe((args) => instance.handleConnection(...args));
|
78 | }
|
79 | }
|
80 | subscribeDisconnectEvent(instance, event) {
|
81 | if (instance.handleDisconnect) {
|
82 | event
|
83 | .pipe((0, operators_1.distinctUntilChanged)())
|
84 | .subscribe(instance.handleDisconnect.bind(instance));
|
85 | }
|
86 | }
|
87 | subscribeMessages(subscribersMap, client, instance) {
|
88 | const adapter = this.config.getIoAdapter();
|
89 | const handlers = subscribersMap.map(({ callback, message }) => ({
|
90 | message,
|
91 | callback: callback.bind(instance, client),
|
92 | }));
|
93 | adapter.bindMessageHandlers(client, handlers, data => (0, rxjs_1.from)(this.pickResult(data)).pipe((0, operators_1.mergeAll)()));
|
94 | }
|
95 | async pickResult(deferredResult) {
|
96 | const result = await deferredResult;
|
97 | if ((0, rxjs_1.isObservable)(result)) {
|
98 | return result;
|
99 | }
|
100 | if (result instanceof Promise) {
|
101 | return (0, rxjs_1.from)(result);
|
102 | }
|
103 | return (0, rxjs_1.of)(result);
|
104 | }
|
105 | inspectEntrypointDefinitions(instance, port, messageHandlers, instanceWrapperId) {
|
106 | messageHandlers.forEach(handler => {
|
107 | this.graphInspector.insertEntrypointDefinition({
|
108 | type: 'websocket',
|
109 | methodName: handler.methodName,
|
110 | className: instance.constructor?.name,
|
111 | classNodeId: instanceWrapperId,
|
112 | metadata: {
|
113 | port,
|
114 | key: handler.message,
|
115 | message: handler.message,
|
116 | },
|
117 | }, instanceWrapperId);
|
118 | });
|
119 | }
|
120 | assignServerToProperties(instance, server) {
|
121 | for (const propertyKey of this.metadataExplorer.scanForServerHooks(instance)) {
|
122 | Reflect.set(instance, propertyKey, server);
|
123 | }
|
124 | }
|
125 | printSubscriptionLogs(instance, subscribersMap) {
|
126 | const gatewayClassName = instance?.constructor?.name;
|
127 | if (!gatewayClassName) {
|
128 | return;
|
129 | }
|
130 | subscribersMap.forEach(({ message }) => this.logger.log(`${gatewayClassName} subscribed to the "${message}" message`));
|
131 | }
|
132 | }
|
133 | exports.WebSocketsController = WebSocketsController;
|