♻️Removed unused websockets
This commit is contained in:
Anton SHEPILOV
@@ -1,4 +1,3 @@
|
||||
export * from "./consumes";
|
||||
export * from "./prefix";
|
||||
export * from "./service-name";
|
||||
export * from "./realtime";
|
||||
|
||||
@@ -1,41 +0,0 @@
|
||||
import { CreateResult } from "../../api/crud-service";
|
||||
import { RealtimeEntityEvent, RealtimeEntityActionType } from "../../../services/realtime/types";
|
||||
import { websocketEventBus } from "../../../services/realtime/bus";
|
||||
import { getRealtimeRecipients, getRoom, RealtimeRecipients } from ".";
|
||||
|
||||
/**
|
||||
*
|
||||
* @param path the path to push the notification to
|
||||
* @param resourcePath the path of the resource itself
|
||||
*/
|
||||
export function RealtimeCreated<T>(recipients: RealtimeRecipients<T>): MethodDecorator {
|
||||
// eslint-disable-next-line @typescript-eslint/ban-types
|
||||
return function (target: Object, propertyKey: string, descriptor: PropertyDescriptor): void {
|
||||
const originalMethod = descriptor.value;
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
descriptor.value = async function (...args: any[]) {
|
||||
const result: CreateResult<T> = await originalMethod.apply(this, args);
|
||||
// context should always be the last arg
|
||||
const context = args && args[args.length - 1];
|
||||
|
||||
if (!(result instanceof CreateResult)) {
|
||||
return result;
|
||||
}
|
||||
|
||||
getRealtimeRecipients(recipients, result.entity, context).forEach(
|
||||
({ room, path, resource }) => {
|
||||
websocketEventBus.publish<T>(RealtimeEntityActionType.Created, {
|
||||
type: result.type,
|
||||
room: getRoom(room, result, context),
|
||||
resourcePath: path || "/",
|
||||
entity: resource,
|
||||
result,
|
||||
} as RealtimeEntityEvent<T>);
|
||||
},
|
||||
);
|
||||
|
||||
return result;
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -1,41 +0,0 @@
|
||||
import { getRealtimeRecipients, getRoom, RealtimeRecipients } from "..";
|
||||
import { DeleteResult } from "../../api/crud-service";
|
||||
import { RealtimeEntityEvent, RealtimeEntityActionType } from "../../../services/realtime/types";
|
||||
import { websocketEventBus } from "../../../services/realtime/bus";
|
||||
|
||||
/**
|
||||
*
|
||||
* @param path the path to push the notification to
|
||||
* @param resourcePath the path of the resource itself
|
||||
*/
|
||||
export function RealtimeDeleted<T>(recipients: RealtimeRecipients<T>): MethodDecorator {
|
||||
// eslint-disable-next-line @typescript-eslint/ban-types
|
||||
return function (target: Object, propertyKey: string, descriptor: PropertyDescriptor): void {
|
||||
const originalMethod = descriptor.value;
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
descriptor.value = async function (...args: any[]) {
|
||||
const result: DeleteResult<T> = await originalMethod.apply(this, args);
|
||||
const context = args && args[args.length - 1];
|
||||
|
||||
if (!(result instanceof DeleteResult)) {
|
||||
return result;
|
||||
}
|
||||
|
||||
if (result.deleted)
|
||||
getRealtimeRecipients(recipients, result.entity, context).forEach(
|
||||
({ room, path, resource }) => {
|
||||
websocketEventBus.publish<T>(RealtimeEntityActionType.Deleted, {
|
||||
type: result.type,
|
||||
room: getRoom(room, result, context),
|
||||
resourcePath: path,
|
||||
entity: resource,
|
||||
result,
|
||||
} as RealtimeEntityEvent<T>);
|
||||
},
|
||||
);
|
||||
|
||||
return result;
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -1,67 +0,0 @@
|
||||
import { ResourcePath } from "../../../../../core/platform/services/realtime/types";
|
||||
import { EntityTarget, ExecutionContext } from "../../api/crud-service";
|
||||
|
||||
export * from "./created";
|
||||
export * from "./deleted";
|
||||
export * from "./updated";
|
||||
export * from "./saved";
|
||||
|
||||
export type RealtimeRecipients<T> =
|
||||
| RealtimeRecipient<T>
|
||||
| RealtimeRecipient<T>[]
|
||||
| ((type: T, context?: ExecutionContext) => RealtimeRecipient<T>[] | RealtimeRecipient<T>);
|
||||
|
||||
export type RealtimeRecipient<T> = {
|
||||
room: RealtimePath<T>;
|
||||
path?: string;
|
||||
resource?: any | T;
|
||||
};
|
||||
|
||||
export function getRealtimeRecipients<T>(
|
||||
recipients: RealtimeRecipients<T>,
|
||||
type: T,
|
||||
context?: ExecutionContext,
|
||||
): RealtimeRecipient<T>[] {
|
||||
if (typeof recipients === "function") recipients = recipients(type, context);
|
||||
if (!recipients) return [];
|
||||
if (!(recipients as RealtimeRecipient<T>[]).length)
|
||||
recipients = [recipients as RealtimeRecipient<T>];
|
||||
|
||||
return (recipients as RealtimeRecipient<T>[]).map(recipient => {
|
||||
return {
|
||||
resource: recipient.resource || type,
|
||||
path: recipient.path || "/",
|
||||
room: recipient.room,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
export type RealtimePath<T> = string | ResourcePath | ResourcePathResolver<T>;
|
||||
export interface ResourcePathResolver<T> {
|
||||
(type: T, context?: ExecutionContext): ResourcePath;
|
||||
}
|
||||
export interface PathResolver<T> {
|
||||
(type: T, context?: ExecutionContext): string;
|
||||
}
|
||||
|
||||
export type RealtimeEntity<T> = null | ((type: T, context?: ExecutionContext) => T);
|
||||
|
||||
export function getRoom<T>(
|
||||
path: RealtimePath<T> = ResourcePath.default(),
|
||||
entity: EntityTarget<T>,
|
||||
context: ExecutionContext,
|
||||
): ResourcePath {
|
||||
if (typeof path === "string") {
|
||||
return ResourcePath.get(path);
|
||||
}
|
||||
|
||||
return typeof path === "function" ? path(entity.entity, context) : path;
|
||||
}
|
||||
|
||||
export function getPath<T>(
|
||||
path: string | PathResolver<T> = "/",
|
||||
entity: EntityTarget<T>,
|
||||
context: ExecutionContext,
|
||||
): string {
|
||||
return typeof path === "function" ? path(entity.entity, context) : path;
|
||||
}
|
||||
@@ -1,41 +0,0 @@
|
||||
import { getRealtimeRecipients, getRoom, RealtimeRecipients } from "..";
|
||||
import { SaveResult } from "../../api/crud-service";
|
||||
import { RealtimeEntityEvent, RealtimeEntityActionType } from "../../../services/realtime/types";
|
||||
import { websocketEventBus } from "../../../services/realtime/bus";
|
||||
|
||||
/**
|
||||
*
|
||||
* @param path the path to push the notification to
|
||||
* @param resourcePath the path of the resource itself
|
||||
*/
|
||||
export function RealtimeSaved<T>(recipients: RealtimeRecipients<T>): MethodDecorator {
|
||||
// eslint-disable-next-line @typescript-eslint/ban-types
|
||||
return function (target: Object, propertyKey: string, descriptor: PropertyDescriptor): void {
|
||||
const originalMethod = descriptor.value;
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
descriptor.value = async function (...args: any[]) {
|
||||
const result: SaveResult<T> = await originalMethod.apply(this, args);
|
||||
// context should always be the last arg
|
||||
const context = args && args[args.length - 1];
|
||||
|
||||
if (!(result instanceof SaveResult)) {
|
||||
return result;
|
||||
}
|
||||
|
||||
getRealtimeRecipients(recipients, result.entity, context).forEach(
|
||||
({ room, path, resource }) => {
|
||||
websocketEventBus.publish<T>(RealtimeEntityActionType.Saved, {
|
||||
type: result.type,
|
||||
room: getRoom(room, result, context),
|
||||
resourcePath: path,
|
||||
entity: resource,
|
||||
result,
|
||||
} as RealtimeEntityEvent<T>);
|
||||
},
|
||||
);
|
||||
|
||||
return result;
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -1,40 +0,0 @@
|
||||
import { getRealtimeRecipients, getRoom, RealtimeRecipients } from "..";
|
||||
import { UpdateResult } from "../../api/crud-service";
|
||||
import { RealtimeEntityEvent, RealtimeEntityActionType } from "../../../services/realtime/types";
|
||||
import { websocketEventBus } from "../../../services/realtime/bus";
|
||||
|
||||
/**
|
||||
*
|
||||
* @param path the path to push the notification to
|
||||
* @param resourcePath the path of the resource itself
|
||||
*/
|
||||
export function RealtimeUpdated<T>(recipients: RealtimeRecipients<T>): MethodDecorator {
|
||||
// eslint-disable-next-line @typescript-eslint/ban-types
|
||||
return function (target: Object, propertyKey: string, descriptor: PropertyDescriptor): void {
|
||||
const originalMethod = descriptor.value;
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
descriptor.value = async function (...args: any[]) {
|
||||
const result: UpdateResult<T> = await originalMethod.apply(this, args);
|
||||
const context = args && args[args.length - 1];
|
||||
|
||||
if (!(result instanceof UpdateResult)) {
|
||||
return result;
|
||||
}
|
||||
|
||||
getRealtimeRecipients(recipients, result.entity, context).forEach(
|
||||
({ room, path, resource }) => {
|
||||
websocketEventBus.publish<T>(RealtimeEntityActionType.Updated, {
|
||||
type: result.type,
|
||||
room: getRoom(room, result, context),
|
||||
resourcePath: path,
|
||||
entity: resource,
|
||||
result,
|
||||
} as RealtimeEntityEvent<T>);
|
||||
},
|
||||
);
|
||||
|
||||
return result;
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -1,78 +0,0 @@
|
||||
import { TdriveServiceProvider } from "../../framework";
|
||||
import { WebSocketUser, WebSocket } from "../../services/websocket/types";
|
||||
import RealtimeEntityManager from "./services/entity-manager";
|
||||
import { RealtimeEntityEvent } from "./types";
|
||||
|
||||
export interface RealtimeServiceAPI extends TdriveServiceProvider {
|
||||
/**
|
||||
* Get the realtime event bus instance
|
||||
*/
|
||||
getBus(): RealtimeEventBus;
|
||||
|
||||
/**
|
||||
* Get the room manager
|
||||
*/
|
||||
getRoomManager(): RealtimeRoomManager;
|
||||
|
||||
/**
|
||||
* Get the entity manager
|
||||
*/
|
||||
getEntityManager(): RealtimeEntityManager;
|
||||
|
||||
/**
|
||||
* Create a token for a room
|
||||
*/
|
||||
sign(items: { room: string }[], userId: string): { room: string; token: string }[];
|
||||
}
|
||||
|
||||
export interface RealtimeRoomManager {
|
||||
/**
|
||||
* Get the number of connected users in the given room
|
||||
* @param room name of the room
|
||||
*/
|
||||
getConnectedUsers(room: string): number;
|
||||
|
||||
/**
|
||||
* Add the user to a room
|
||||
*
|
||||
* @param websocket
|
||||
* @param room
|
||||
* @param user
|
||||
*/
|
||||
join(websocket: WebSocket, room: string, user: WebSocketUser): void;
|
||||
|
||||
/**
|
||||
* Remove the user from a room
|
||||
*
|
||||
* @param websocket
|
||||
* @param room
|
||||
* @param user
|
||||
*/
|
||||
leave(websocket: WebSocket, room: string, user: WebSocketUser): void;
|
||||
|
||||
/**
|
||||
* Remove a user from all the room he joined
|
||||
*
|
||||
* @param websocket
|
||||
* @param user
|
||||
*/
|
||||
leaveAll(websocket: WebSocket, user: WebSocketUser): void;
|
||||
}
|
||||
|
||||
export interface RealtimeEventBus {
|
||||
/**
|
||||
* Subscribes to a topic
|
||||
*
|
||||
* @param topic Topic to subscribe to
|
||||
* @param listener Listener to call when event is published in the given topic
|
||||
*/
|
||||
subscribe<Entity>(topic: string, listener: (event: RealtimeEntityEvent<Entity>) => void): this;
|
||||
|
||||
/**
|
||||
* Publish event in a topic
|
||||
*
|
||||
* @param topic Topic to publish event to
|
||||
* @param event Event to publish
|
||||
*/
|
||||
publish<Entity>(topic: string, event: RealtimeEntityEvent<Entity>): boolean;
|
||||
}
|
||||
@@ -1,22 +0,0 @@
|
||||
/**
|
||||
* EventBus is used internally to exchange events between decorators and realtime manager
|
||||
*/
|
||||
import { EventEmitter } from "events";
|
||||
import { RealtimeEventBus } from "./api";
|
||||
import { RealtimeEntityEvent } from "./types";
|
||||
|
||||
class WebsocketEventBus extends EventEmitter implements RealtimeEventBus {
|
||||
constructor() {
|
||||
super();
|
||||
}
|
||||
|
||||
subscribe<Entity>(topic: string, listener: (event: RealtimeEntityEvent<Entity>) => void): this {
|
||||
return this.on(topic, listener);
|
||||
}
|
||||
|
||||
publish<Entity>(topic: string, event: RealtimeEntityEvent<Entity>): boolean {
|
||||
return this.emit(topic, event);
|
||||
}
|
||||
}
|
||||
|
||||
export const websocketEventBus = new WebsocketEventBus();
|
||||
@@ -1,76 +0,0 @@
|
||||
import { Consumes, ServiceName, TdriveService } from "../../framework";
|
||||
import { SkipCLI } from "../../framework/decorators/skip";
|
||||
import { localEventBus } from "../../framework/event-bus";
|
||||
import WebSocketAPI from "../../services/websocket/provider";
|
||||
import AuthService from "../auth/provider";
|
||||
import { RealtimeEventBus, RealtimeRoomManager, RealtimeServiceAPI } from "./api";
|
||||
import { websocketEventBus } from "./bus";
|
||||
import RealtimeEntityManager from "./services/entity-manager";
|
||||
import RoomManagerImpl from "./services/room-manager";
|
||||
import { RealtimeBaseBusEvent, RealtimeLocalBusEvent } from "./types";
|
||||
|
||||
@Consumes(["websocket", "auth"])
|
||||
@ServiceName("realtime")
|
||||
export default class RealtimeService
|
||||
extends TdriveService<RealtimeServiceAPI>
|
||||
implements RealtimeServiceAPI
|
||||
{
|
||||
private roomManager: RoomManagerImpl;
|
||||
private entityManager: RealtimeEntityManager;
|
||||
private auth: AuthService;
|
||||
version = "1";
|
||||
|
||||
api(): RealtimeServiceAPI {
|
||||
return this;
|
||||
}
|
||||
|
||||
@SkipCLI()
|
||||
async doStart(): Promise<this> {
|
||||
const ws = this.context.getProvider<WebSocketAPI>("websocket");
|
||||
this.auth = this.context.getProvider<AuthService>("auth");
|
||||
|
||||
this.roomManager = new RoomManagerImpl(ws, this.auth);
|
||||
this.roomManager.init();
|
||||
this.entityManager = new RealtimeEntityManager(ws);
|
||||
this.entityManager.init();
|
||||
|
||||
localEventBus.subscribe("realtime:event", (event: RealtimeBaseBusEvent<any>) => {
|
||||
event.data._type = event.type;
|
||||
ws.getIo().to(event.room).emit("realtime:event", { name: event.room, data: event.data });
|
||||
});
|
||||
|
||||
localEventBus.subscribe("realtime:publish", (data: RealtimeLocalBusEvent<any>) => {
|
||||
this.getBus().publish(data.topic, data.event);
|
||||
});
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
getBus(): RealtimeEventBus {
|
||||
return websocketEventBus;
|
||||
}
|
||||
|
||||
getRoomManager(): RealtimeRoomManager {
|
||||
return this.roomManager;
|
||||
}
|
||||
|
||||
getEntityManager(): RealtimeEntityManager {
|
||||
return this.entityManager;
|
||||
}
|
||||
|
||||
sign(items: { room: string }[], sub: string) {
|
||||
return items.map(item => {
|
||||
const token = this.auth.sign({
|
||||
sub,
|
||||
name: item.room,
|
||||
//Fixme, reduce this number but also in frontend reload the websocket token after this ends
|
||||
iat: Math.round(new Date().getTime() / 1000) + 60 * 60 * 24 * 31,
|
||||
nbf: Math.round(new Date().getTime() / 1000) - 60,
|
||||
});
|
||||
return {
|
||||
...item,
|
||||
token,
|
||||
};
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -1,54 +0,0 @@
|
||||
import { logger } from "../../../framework/logger";
|
||||
import { RealtimeEntityActionType, RealtimeEntityEvent } from "../types";
|
||||
import WebSocketAPI from "../../../services/websocket/provider";
|
||||
import { websocketEventBus } from "../bus";
|
||||
|
||||
const REALTIME_RESOURCE = "realtime:resource";
|
||||
|
||||
export default class RealtimeEntityManager {
|
||||
constructor(private ws: WebSocketAPI) {}
|
||||
|
||||
init(): void {
|
||||
websocketEventBus.subscribe(RealtimeEntityActionType.Created, event => {
|
||||
this.pushResourceEvent(event, RealtimeEntityActionType.Created);
|
||||
});
|
||||
|
||||
websocketEventBus.subscribe(RealtimeEntityActionType.Updated, event => {
|
||||
this.pushResourceEvent(event, RealtimeEntityActionType.Updated);
|
||||
});
|
||||
|
||||
websocketEventBus.subscribe(RealtimeEntityActionType.Deleted, event => {
|
||||
this.pushResourceEvent(event, RealtimeEntityActionType.Deleted);
|
||||
});
|
||||
|
||||
websocketEventBus.subscribe(RealtimeEntityActionType.Saved, event => {
|
||||
this.pushResourceEvent(event, RealtimeEntityActionType.Saved);
|
||||
});
|
||||
|
||||
websocketEventBus.subscribe(RealtimeEntityActionType.Event, event => {
|
||||
this.pushResourceEvent(event, RealtimeEntityActionType.Event);
|
||||
});
|
||||
}
|
||||
|
||||
private pushResourceEvent(
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
event: RealtimeEntityEvent<any>,
|
||||
action: RealtimeEntityActionType,
|
||||
): void {
|
||||
event.room.path.forEach(path => {
|
||||
logger.info(`Pushing ${action} entity to room ${path}`);
|
||||
const message: unknown = {
|
||||
action,
|
||||
room: path,
|
||||
type: event.type,
|
||||
path: event.resourcePath,
|
||||
resource: event.entity,
|
||||
};
|
||||
if (logger.isLevelEnabled("debug")) {
|
||||
logger.debug(`Entity to push to room ${path}: %o`, message);
|
||||
}
|
||||
|
||||
this.ws.getIo().to(path).emit(REALTIME_RESOURCE, message);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -1,2 +0,0 @@
|
||||
export * from "./entity-manager";
|
||||
export * from "./room-manager";
|
||||
@@ -1,164 +0,0 @@
|
||||
import { logger } from "../../../framework/logger";
|
||||
import {
|
||||
JoinLeaveRoomError,
|
||||
JoinLeaveRoomSuccess,
|
||||
JoinRoomEvent,
|
||||
LeaveRoomEvent,
|
||||
ClientEvent,
|
||||
WebsocketRoomSignature,
|
||||
} from "../types";
|
||||
import { RealtimeRoomManager } from "../api";
|
||||
import WebSocketAPI from "../../../services/websocket/provider";
|
||||
import { WebSocketUser, WebSocket } from "../../../services/websocket/types";
|
||||
import AuthService from "../../auth/provider";
|
||||
|
||||
export default class RoomManager implements RealtimeRoomManager {
|
||||
constructor(private ws: WebSocketAPI, private auth: AuthService) {}
|
||||
|
||||
init(): void {
|
||||
this.ws.onUserConnected(event => {
|
||||
logger.info(`User ${event.user.id} is connected`);
|
||||
|
||||
event.socket.on("realtime:join", async (joinEvent: JoinRoomEvent) => {
|
||||
const canJoin =
|
||||
joinEvent.name.indexOf("previous::") === 0 || //Compatibility with old collections
|
||||
(await this.userCanJoinRoom(event.user, joinEvent));
|
||||
|
||||
if (canJoin) {
|
||||
this.join(event.socket, joinEvent.name, event.user);
|
||||
} else {
|
||||
this.sendError("join", event.socket, {
|
||||
name: joinEvent.name,
|
||||
message: "User is not authorized to join room",
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
event.socket.on("realtime:leave", (leaveEvent: LeaveRoomEvent) => {
|
||||
this.leave(event.socket, leaveEvent.name, event.user);
|
||||
});
|
||||
|
||||
event.socket.on("realtime:event", async (clientEvent: ClientEvent) => {
|
||||
const canEmit =
|
||||
clientEvent.name.indexOf("previous::") === 0 || //Compatibility with old collections
|
||||
(await this.userCanEmitInRoom(event.user, clientEvent));
|
||||
if (canEmit) {
|
||||
this.sendEvent(clientEvent.name, clientEvent.data);
|
||||
} else {
|
||||
this.sendError("event", event.socket, {
|
||||
name: clientEvent.name,
|
||||
message: "User is not authorized to emit in this room",
|
||||
});
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
this.ws.onUserDisconnected(event => {
|
||||
logger.info(`User ${event.user.id} is disconnected`);
|
||||
this.leaveAll(event.socket, event.user);
|
||||
});
|
||||
}
|
||||
|
||||
getConnectedUsers(room: string): number {
|
||||
if (this.ws.getIo().sockets.adapter.rooms.has(room)) {
|
||||
return this.ws.getIo().sockets.adapter.rooms.get(room).entries.length;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check of the user can join room
|
||||
*
|
||||
* @param user
|
||||
* @param joinEvent
|
||||
* @return Promise<boolean> true if can join, false otherwise. Never rejects.
|
||||
*/
|
||||
async userCanJoinRoom(user: WebSocketUser, joinEvent: JoinRoomEvent): Promise<boolean> {
|
||||
logger.info(`Checking if user ${user.id} can join room ${joinEvent.name} with token`);
|
||||
|
||||
try {
|
||||
//Public rooms we just check the user is logged in
|
||||
if (
|
||||
joinEvent.name.startsWith("/users/") || //User update room
|
||||
joinEvent.name === "/ping"
|
||||
) {
|
||||
return !!this.auth.verifyToken(joinEvent.token)?.sub;
|
||||
}
|
||||
|
||||
//Retro-compatibility for mobile up to february 2021 (to remove after this date)
|
||||
if (joinEvent.token === "tdrive") return true;
|
||||
|
||||
const signature = this.auth.verifyTokenObject<WebsocketRoomSignature>(joinEvent.token);
|
||||
|
||||
return (
|
||||
signature &&
|
||||
signature.name === joinEvent.name &&
|
||||
signature.sub === user.id &&
|
||||
signature.iat > Math.round(new Date().getTime() / 1000)
|
||||
);
|
||||
} catch (err) {
|
||||
console.log(err);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
userCanEmitInRoom = this.userCanJoinRoom;
|
||||
|
||||
async join(websocket: WebSocket, room: string, user: WebSocketUser): Promise<void> {
|
||||
logger.info(`User ${user.id} is joining room ${room}`);
|
||||
|
||||
try {
|
||||
await websocket.join(room);
|
||||
} catch (err) {
|
||||
if (err) {
|
||||
logger.error(`Error while joining room ${room}`, err);
|
||||
this.sendError("join", websocket, {
|
||||
name: room,
|
||||
message: "Error while joining room",
|
||||
});
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
this.sendSuccess("join", websocket, { name: room });
|
||||
logger.info(`User ${user.id} joined room ${room}`);
|
||||
}
|
||||
|
||||
async leave(websocket: WebSocket, room: string, user: WebSocketUser): Promise<void> {
|
||||
logger.info(`User ${user.id} is leaving room ${room}`);
|
||||
|
||||
try {
|
||||
await websocket.leave(room);
|
||||
} catch (err) {
|
||||
if (err) {
|
||||
logger.error(`Error while leaving room ${room}`, err);
|
||||
this.sendError("leave", websocket, {
|
||||
name: room,
|
||||
message: "Error while leaving room",
|
||||
});
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
this.sendSuccess("leave", websocket, { name: room });
|
||||
logger.info(`User ${user.id} left room ${room}`);
|
||||
}
|
||||
|
||||
leaveAll(websocket: WebSocket, user: WebSocketUser): void {
|
||||
logger.info(`Leaving rooms for user ${user.id}`);
|
||||
websocket.rooms.forEach(room => websocket.leave(room));
|
||||
}
|
||||
|
||||
sendError(event: string, websocket: WebSocket, error: JoinLeaveRoomError): void {
|
||||
websocket.emit(`realtime:${event}:error`, error);
|
||||
}
|
||||
|
||||
sendSuccess(event: string, websocket: WebSocket, success: JoinLeaveRoomSuccess): void {
|
||||
websocket.emit(`realtime:${event}:success`, success);
|
||||
}
|
||||
|
||||
sendEvent(path: string, data: any): void {
|
||||
this.ws.getIo().to(path).emit("realtime:event", { name: path, data: data });
|
||||
}
|
||||
}
|
||||
@@ -1,83 +0,0 @@
|
||||
import { EntityOperationResult } from "../../framework/api/crud-service";
|
||||
|
||||
export enum RealtimeEntityActionType {
|
||||
Created = "created",
|
||||
Saved = "saved",
|
||||
Updated = "updated",
|
||||
Deleted = "deleted",
|
||||
Event = "event",
|
||||
}
|
||||
|
||||
export interface WebsocketRoomSignature {
|
||||
name: string; //Ws room name
|
||||
sub: string; //User id
|
||||
iat: number; //Deadline
|
||||
}
|
||||
|
||||
export class ResourcePath {
|
||||
constructor(readonly path: string[] = ["/"], readonly name: string = "default") {}
|
||||
|
||||
static default(): ResourcePath {
|
||||
return new ResourcePath();
|
||||
}
|
||||
|
||||
static get(path: string | string[]): ResourcePath {
|
||||
return new ResourcePath(typeof path === "string" ? [path] : path);
|
||||
}
|
||||
}
|
||||
export class RealtimeEntityEvent<Entity> {
|
||||
// the type of the resource
|
||||
type: string;
|
||||
// the room to push the resource to
|
||||
room: ResourcePath;
|
||||
// the unique full path of the resource, ie where we can get it
|
||||
resourcePath: string | null;
|
||||
// the input resource
|
||||
entity: Entity;
|
||||
// the action result which fired this event
|
||||
result: EntityOperationResult<Entity> | null;
|
||||
}
|
||||
|
||||
export class RealtimeApplicationEvent {
|
||||
action: "configure" | "close_configure";
|
||||
application: unknown;
|
||||
form: unknown;
|
||||
hidden_data: unknown;
|
||||
}
|
||||
|
||||
export class RealtimeBaseBusEvent<T> {
|
||||
room: string;
|
||||
type: string;
|
||||
data: T;
|
||||
}
|
||||
|
||||
export class RealtimeLocalBusEvent<Entity> {
|
||||
event: RealtimeEntityEvent<Entity>;
|
||||
topic: string;
|
||||
}
|
||||
|
||||
export class JoinRoomEvent {
|
||||
name: string;
|
||||
token?: string;
|
||||
}
|
||||
|
||||
export class LeaveRoomEvent {
|
||||
name: string;
|
||||
}
|
||||
|
||||
export class ClientEvent {
|
||||
name: string;
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
data: any;
|
||||
}
|
||||
|
||||
export interface JoinLeaveRoomError {
|
||||
name: string;
|
||||
message?: string;
|
||||
type?: string;
|
||||
}
|
||||
|
||||
export interface JoinLeaveRoomSuccess {
|
||||
name: string;
|
||||
message?: string;
|
||||
}
|
||||
@@ -1,47 +0,0 @@
|
||||
import { Consumes, ServiceName, TdriveService } from "../../framework";
|
||||
import WebServerAPI from "../webserver/provider";
|
||||
import WebSocketAPI from "./provider";
|
||||
import { WebSocketService } from "./services";
|
||||
import { AdaptersConfiguration } from "./types";
|
||||
import FastifyIO from "fastify-socket.io";
|
||||
|
||||
@Consumes(["webserver"])
|
||||
@ServiceName("websocket")
|
||||
export default class WebSocket extends TdriveService<WebSocketAPI> {
|
||||
private service: WebSocketService;
|
||||
name = "websocket";
|
||||
version = "1";
|
||||
|
||||
api(): WebSocketAPI {
|
||||
return this.service;
|
||||
}
|
||||
|
||||
async doInit(): Promise<this> {
|
||||
const webserver = this.context.getProvider<WebServerAPI>("webserver");
|
||||
const fastify = webserver.getServer();
|
||||
|
||||
const options = {
|
||||
path: this.configuration.get<string>("path", "/socket"),
|
||||
};
|
||||
|
||||
fastify.register(FastifyIO, {
|
||||
...options,
|
||||
allowEIO3: true,
|
||||
cors: {
|
||||
//Allow all origins
|
||||
origin: (origin, callback) => callback(null, origin),
|
||||
credentials: true,
|
||||
},
|
||||
});
|
||||
|
||||
this.service = new WebSocketService({
|
||||
server: fastify,
|
||||
options,
|
||||
ready: webserver.onReady.bind(webserver),
|
||||
adapters: this.configuration.get<AdaptersConfiguration>("adapters"),
|
||||
auth: this.configuration.get<{ secret: string }>("auth.jwt"),
|
||||
});
|
||||
|
||||
return this;
|
||||
}
|
||||
}
|
||||
@@ -1,17 +0,0 @@
|
||||
import { EventEmitter } from "events";
|
||||
import socketIO from "socket.io";
|
||||
import { TdriveServiceProvider } from "../../framework";
|
||||
import { User } from "../../../../utils/types";
|
||||
import { WebsocketUserEvent, WebSocket, WebSocketUser } from "./types";
|
||||
|
||||
export default interface WebSocketAPI extends TdriveServiceProvider, EventEmitter {
|
||||
getIo(): socketIO.Server;
|
||||
|
||||
isConnected(user: User): boolean;
|
||||
|
||||
onUserConnected(listener: (event: WebsocketUserEvent) => void): this;
|
||||
|
||||
onUserDisconnected(listener: (event: WebsocketUserEvent) => void): this;
|
||||
|
||||
getUser(websocket: WebSocket): WebSocketUser;
|
||||
}
|
||||
@@ -1,105 +0,0 @@
|
||||
import socketIO from "socket.io";
|
||||
// eslint-disable-next-line @typescript-eslint/no-var-requires
|
||||
const { createClient } = require("redis");
|
||||
import SocketIORedis from "@socket.io/redis-adapter";
|
||||
import jwt from "jsonwebtoken";
|
||||
import WebSocketAPI from "../provider";
|
||||
import {
|
||||
WebSocket,
|
||||
WebSocketServiceConfiguration,
|
||||
WebSocketUser,
|
||||
WebsocketUserEvent,
|
||||
} from "../types";
|
||||
import { EventEmitter } from "events";
|
||||
import { User } from "../../../../../utils/types";
|
||||
import { JwtType } from "../../types";
|
||||
|
||||
export class WebSocketService extends EventEmitter implements WebSocketAPI {
|
||||
version: "1";
|
||||
private io: socketIO.Server;
|
||||
|
||||
constructor(serviceConfiguration: WebSocketServiceConfiguration) {
|
||||
super();
|
||||
|
||||
serviceConfiguration.ready(() => {
|
||||
this.io = serviceConfiguration.server.io;
|
||||
|
||||
if (serviceConfiguration.adapters?.types?.includes("redis")) {
|
||||
const pubClient = createClient(serviceConfiguration.adapters.redis);
|
||||
const subClient = pubClient.duplicate();
|
||||
this.io.adapter(SocketIORedis.createAdapter(pubClient, subClient));
|
||||
}
|
||||
|
||||
this.io.on("connection", socket => {
|
||||
socket.on("authenticate", message => {
|
||||
if (message.token) {
|
||||
jwt.verify(
|
||||
message.token as string,
|
||||
serviceConfiguration.auth.secret as string,
|
||||
(err, decoded) => {
|
||||
if (err) {
|
||||
socket.emit("unauthorized", { err });
|
||||
socket.disconnect();
|
||||
return;
|
||||
}
|
||||
(socket as unknown as WebSocket).decoded_token = decoded as JwtType;
|
||||
const user = this.getUser(socket as WebSocket);
|
||||
|
||||
socket.emit("authenticated");
|
||||
|
||||
socket.on("message", message => {
|
||||
this.io.emit("message", message);
|
||||
});
|
||||
|
||||
this.emit("user:connected", {
|
||||
user,
|
||||
socket,
|
||||
event: "user:connected",
|
||||
} as WebsocketUserEvent);
|
||||
|
||||
socket.on("disconnect", () =>
|
||||
this.emit("user:disconnected", {
|
||||
user,
|
||||
socket,
|
||||
event: "user:disconnected",
|
||||
} as WebsocketUserEvent),
|
||||
);
|
||||
},
|
||||
);
|
||||
} else {
|
||||
socket.emit("unauthorized", { err: "No token provided" });
|
||||
socket.disconnect();
|
||||
return;
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
onUserConnected(listener: (event: WebsocketUserEvent) => void): this {
|
||||
return this.on("user:connected", listener);
|
||||
}
|
||||
|
||||
onUserDisconnected(listener: (event: WebsocketUserEvent) => void): this {
|
||||
return this.on("user:disconnected", listener);
|
||||
}
|
||||
|
||||
getUser(socket: WebSocket): WebSocketUser {
|
||||
return {
|
||||
id: socket.decoded_token.sub,
|
||||
identity_provider_id: socket.decoded_token.provider_id,
|
||||
email: socket.decoded_token.email,
|
||||
token: socket.decoded_token,
|
||||
allow_tracking: socket.decoded_token.track,
|
||||
};
|
||||
}
|
||||
|
||||
getIo(): socketIO.Server {
|
||||
return this.io;
|
||||
}
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-unused-vars
|
||||
isConnected(user: User): boolean {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -1,39 +0,0 @@
|
||||
import socketIO, { Socket } from "socket.io";
|
||||
import { RedisAdapterOptions } from "@socket.io/redis-adapter";
|
||||
import { User } from "../../../../utils/types";
|
||||
import { JwtType } from "../types";
|
||||
import { FastifyInstance } from "fastify";
|
||||
|
||||
export interface AdaptersConfiguration {
|
||||
types: Array<string>;
|
||||
redis: RedisAdapterOptions;
|
||||
}
|
||||
|
||||
export interface WebSocketServiceConfiguration {
|
||||
server: FastifyInstance;
|
||||
// eslint-disable-next-line @typescript-eslint/ban-types
|
||||
ready: Function;
|
||||
options?: socketIO.ServerOptions | { path: string };
|
||||
adapters?: AdaptersConfiguration;
|
||||
auth?: { secret: string };
|
||||
}
|
||||
|
||||
export interface WebSocketUser extends User {
|
||||
token?: DecodedToken;
|
||||
}
|
||||
|
||||
export interface WebSockets {
|
||||
[index: string]: WebSocket[];
|
||||
}
|
||||
|
||||
export interface WebSocket extends Socket {
|
||||
decoded_token: DecodedToken;
|
||||
}
|
||||
|
||||
export interface WebsocketUserEvent {
|
||||
event: string;
|
||||
socket: WebSocket;
|
||||
user: WebSocketUser;
|
||||
}
|
||||
|
||||
type DecodedToken = JwtType;
|
||||
Reference in New Issue
Block a user