import { BehaviorSubject, Observable, Observer, Subject } from 'rxjs'; import { createServer } from 'http'; import { Server, Socket as SocketForConnectedClient } from 'socket.io'; import { io, Socket as ClientSocket } from 'socket.io-client'; import * as fs from 'fs' import { v4 as uuidv4 } from 'uuid' import { ConnectionState, Transport, TransportEvent, TransportMessage } from '../interface/connector.interface'; import { ConnectedClientSocket, ConnectedServerSocket } from '../transport/websocket'; import { EventMessage } from '../interface/transport.interface'; export function startSocketServer(port: number): Observable { return new Observable((observer) => { try { console.log(`Socket Server ${port} Started....`) let httpServer = createServer(); let socketServer = new Server(httpServer) // something wrong here socketServer.on('connection', (socket) => { observer.next(socket) }) socketServer.engine.on("connection_error", (err) => { console.log(err.req); // the request object console.log(err.code); // the error code, for example 1 console.log(err.message); // the error message, for example "Session ID unknown" console.log(err.context); // some additional error context }); // Start the socketServer httpServer.listen(port) } catch (error) { observer.error(error) } }) } export async function startClientSocketConnection(serverUrl: string): Promise { return new Promise((resolve, reject) => { try { // let clientSocket = io(serverUrl) let clientSocket: ClientSocket = io(serverUrl, { reconnection: true, // Enable automatic reconnections reconnectionAttempts: 1000, // Retry up to 10 times reconnectionDelay: 500, // Start with a 500ms delay reconnectionDelayMax: 10000, // Delay can grow to a max of 10 seconds randomizationFactor: 0.3, }) resolve(clientSocket) } catch (error) { reject(error) } }) } // After establishing connection to the server, set up the credentials, confirm whether or not if there's any credentials, if not ask for one from the server export function handleClientSocketConnection(socket: ClientSocket, serversConnected: ConnectedServerSocket[]): Observable { return new Observable((eventNotification: Observer) => { let clientName!: string let buffer: any[] = [] let receiverProfileInfo!: ConnectedServerSocket // Listen for a connection event socket.on('connect', () => { console.log('Connected to the server:', socket.id) if (clientName) { checkOwnClientInfo(clientName).then((profile: ConnectedServerSocket) => { receiverProfileInfo = profile socket.emit('profile', { name: 'Old Client', data: profile }) }).catch((error) => { socket.emit('profile', { name: 'New Client', data: null }) }) } else { socket.emit('profile', { name: 'New Client', data: null }) } }); // Listen for messages from the server. Generally here's the responses socket.on('message', (msg: any) => { console.log(`Websocket Client Transport Receieve Msg`, msg.id) if (receiverProfileInfo) { // publish to event eventNotification.next({ id: uuidv4(), event: 'New Message', data: msg }) } else { // Do nothing. just store in local array first. Cannot process without information. but then again, don['t need information if acting as client // but for consistency sake, will impose the standard buffer.push(msg) // store locally for now } }) socket.on('profile', (data: { name: string, message: any }) => { console.log(data) if (data.name == 'New Profile') { console.log(`Assigned client Name: ${(data.message as ConnectedServerSocket).id}`) receiverProfileInfo = data.message as ConnectedServerSocket writeFile(data.message as ConnectedServerSocket, (data.message as ConnectedServerSocket).id).then(() => { clientName = receiverProfileInfo.id // broadcast event to allow retransmission to release buffer eventNotification.next({ id: uuidv4(), event: `New Server`, data: { clientId: (data.message as ConnectedServerSocket).id, message: `New Websocket Channel ${(data.message as ConnectedServerSocket).id} established.` } as EventMessage }) }).catch((error) => { }) // do nothing at the moment. // Update websocket instance record receiverProfileInfo = { id: (data.message as ConnectedServerSocket).id, dateCreated: new Date(), socketInstance: socket, connectionState: new BehaviorSubject(`ONLINE`) } serversConnected.push(receiverProfileInfo) } if (data.name == 'Adjusted Profile') { console.log(`Assigned client Name: ${(data.message as ConnectedServerSocket).id}`) receiverProfileInfo = data.message as ConnectedServerSocket writeFile(data.message as ConnectedServerSocket, (data.message as ConnectedServerSocket).id).then(() => { // broadcast event to allow retransmission to release buffer eventNotification.next({ id: uuidv4(), event: 'Client Reconnected', data: { clientId: (data.message as ConnectedServerSocket).id, message: `Existing Websocket Channel ${(data.message as ConnectedServerSocket).id} re-established.` } as EventMessage }) }).catch((error) => { }) // do nothing at the moment. // Update websocket instance record let clientObj: ConnectedServerSocket | undefined = serversConnected.find(obj => obj.id === (data.message as ConnectedServerSocket).id) if (clientObj) { clientObj.socketInstance = receiverProfileInfo.socketInstance clientObj.connectionState.next('ONLINE') } } if (data.name == 'Error') { console.log(`Server cannot find credentials`, data.message) // logic to request for new credentials setTimeout(() => { socket.emit('profile', { name: 'New Client', data: null }) }, 2000) } }) // Handle disconnection socket.on('disconnect', () => { console.log('Websocket Client disconnected from the server'); if (receiverProfileInfo) { eventNotification.next({ id: uuidv4(), event: `Client Disconnected`, data: { clientId: receiverProfileInfo.id, message: `Server for Channel ${receiverProfileInfo.id} disconnected.` } as EventMessage }) receiverProfileInfo.connectionState.next(`OFFLINE`) } }); }) } // For SERVER Usage: set up socket listeners to start listening for different events export function handleNewSocketClient(socket: SocketForConnectedClient, connectedClientSocket: ConnectedClientSocket[]): Observable { return new Observable((event: Observer) => { console.log(`Setting up listeners for socket:${socket.id}`) // returns the socket client instance // listen to receiver's initiotion first before assigning 'credentials' socket.on(`profile`, (message: { name: string, data: any }) => { if (message.name == 'New Client') { let clientInstance: ConnectedClientSocket = { id: uuidv4(), // client should only be assigned at this level. And is passed around for reference pointing dateCreated: new Date(), socketInstance: socket, connectionState: new BehaviorSubject(`ONLINE`) } // send to receiver for reference socket.emit('profile', { name: `New Profile`, message: { id: clientInstance.id } }) // publish first event notification event.next({ id: uuidv4(), event: `New Client`, data: { clientId: clientInstance.id, message: `New Client Connected. Adapter ID assigned: ${clientInstance.id}`, payload: clientInstance } as EventMessage }) // Update connected clientInstance info to adapter connectedClientSocket.push(clientInstance) startListening(socket, clientInstance, event) } else { // update first let clientInstance: ConnectedClientSocket | undefined = connectedClientSocket.find(obj => obj.id === message.data.id) if (clientInstance) { console.log(`Socket Client ${clientInstance.id} Found`) socket.emit('profile', { name: 'Adjusted Profile', message: { id: clientInstance.id } }) // replace socket instance since the previous has been terminated clientInstance.socketInstance = socket // need to start listening again, because it's assigned a different socket instance this time round startListening(socket, clientInstance, event) event.next({ id: uuidv4(), event: 'Client Reconnected', data: { clientId: clientInstance.id, message: `Client ${clientInstance.id} connection re-established`, payload: clientInstance } as EventMessage }) // Resume operation if (clientInstance.connectionState.getValue() == 'OFFLINE') { clientInstance.connectionState.next(`ONLINE`) } } else { console.log(`Profile Not Found`) socket.emit('profile', { name: 'Error', message: 'Receiver Profile Not found' }) } } }) }) } // Specifically to write receiver profile information export async function writeFile(data: ConnectedServerSocket, filename: string): Promise { return new Promise((resolve, reject) => { // Write JSON data to a file fs.writeFile(`${filename}.json`, JSON.stringify(data, null, 2), (err) => { if (err) { console.error('Error writing file', err); reject(false) } else { console.log('File has been written'); resolve(true) } }); }) } // Check if filename exists. Return profile information if there's any export async function checkOwnClientInfo(filename?: string): Promise { return new Promise((resolve, reject) => { // Check if the file exists if (fs.existsSync(`${filename}.json`)) { try { // Read the file contents const fileData = fs.readFileSync(`${filename}.json`, 'utf8'); // If the file is empty, return an error if (fileData.trim() === "") { throw new Error("File is empty"); } // Parse and return the data if present const jsonData = JSON.parse(fileData); resolve(jsonData) } catch (err) { // Handle parsing errors or other file-related errors console.error("Error reading or parsing file:", err); reject(''); } } else { console.error("File does not exist"); reject(''); } }) } export function startListening(socket: SocketForConnectedClient, client: ConnectedClientSocket, eventListener: Observer): void { /* Generally, we don't need this unless in the case of being the receiver */ socket.on('message', (message: any) => { eventListener.next({ id: uuidv4(), event: 'New Message', data: { id: uuidv4(), dateCreated: new Date(), transport: Transport.Websocket, target: client.id, // this ref to be associated with the client/channel payload: message } as TransportMessage }) }) socket.on('disconnect', () => { eventListener.next({ id: uuidv4(), event: 'Server Disconnected', data: { clientID: client.id, time: new Date() } }) eventListener.error(`Client ${client.id} disconnected. Terminating this observable event for this client socket...`) eventListener.complete() }) }