|
|
@@ -1,4 +1,4 @@
|
|
|
-import { BehaviorSubject, Observable, Observer, Subject } from 'rxjs';
|
|
|
+import { Observable, Observer } from 'rxjs';
|
|
|
import { createServer } from 'http';
|
|
|
import { Server, Socket as SocketForConnectedClient } from 'socket.io';
|
|
|
import { io, Socket as SocketForConnectedServer } from 'socket.io-client';
|
|
|
@@ -7,8 +7,8 @@ import { v4 as uuidv4 } from 'uuid'
|
|
|
import { ConnectedSocketClient, ConnectedSocketServer } from '../transport/websocket';
|
|
|
import ConsoleLogger from './log.utils';
|
|
|
import path from 'path';
|
|
|
-import { ClientObject, ConnectionState, GeneralEvent, TransportMessage } from '../interface/interface';
|
|
|
-import { addClientToDB, checkIfClientExists, checkOwnClientInfo, writeFile } from './general.utils';
|
|
|
+import { ClientObject, GeneralEvent, TransportMessage } from '../interface/interface';
|
|
|
+import { addClientToDB, checkIfClientExists, checkOwnClientInfo } from './general.utils';
|
|
|
const console: ConsoleLogger = new ConsoleLogger(`SocketUtils`, ['transport'])
|
|
|
|
|
|
export function startSocketServer(port: number): Observable<SocketForConnectedClient> {
|
|
|
@@ -63,69 +63,71 @@ export async function startClientSocketConnection(serverUrl: string): Promise<So
|
|
|
export function handleClientSocketConnection(transportServiceId: string, socket: SocketForConnectedServer, serversConnected: ConnectedSocketServer[]): Observable<GeneralEvent<any>> {
|
|
|
return new Observable((eventNotification: Observer<GeneralEvent<any>>) => {
|
|
|
let buffer: any[] = []
|
|
|
- let receiverProfileInfo!: ConnectedSocketServer
|
|
|
+ let selfId!: string
|
|
|
+ let clientProfileInfo!: ConnectedSocketServer // this information refers to the server itself
|
|
|
|
|
|
// Listen for a connection event
|
|
|
socket.on('connect', () => {
|
|
|
console.log({ message: `Connected to the server ${socket.id} ` })
|
|
|
- if (receiverProfileInfo?.clientId) {
|
|
|
- checkOwnClientInfo(receiverProfileInfo.clientId).then((profile: { id: string }) => {
|
|
|
- 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
|
|
|
+ checkOwnClientInfo(`clientprofile`).then((profile: { id: string, clientId: string }) => {
|
|
|
+ socket.emit('handshaking', {
|
|
|
+ id: profile.id,
|
|
|
})
|
|
|
- }
|
|
|
- });
|
|
|
+ }).catch((error) => {
|
|
|
+ console.error({ message: `Client profile is not defined!`, details: error })
|
|
|
+ })
|
|
|
|
|
|
- // Listen for messages from the server. Generally here's the responses
|
|
|
- socket.on('message', (msg: any) => {
|
|
|
- // console.log(`Websocket Client Transport Receieve Msg`, msg)
|
|
|
- if (receiverProfileInfo) {
|
|
|
- // publish to event
|
|
|
- eventNotification.next({
|
|
|
- id: uuidv4(),
|
|
|
- type: `Transport Event`,
|
|
|
- event: 'New Message',
|
|
|
- date: new Date(),
|
|
|
- data: {
|
|
|
+
|
|
|
+ // Listen for messages from the server. Generally here's the responses
|
|
|
+ socket.on('message', (msg: TransportMessage) => {
|
|
|
+ if (clientProfileInfo) {
|
|
|
+ console.log({ message: `Websocket Client Transport Receive Msg` })
|
|
|
+ // publish to event
|
|
|
+ eventNotification.next({
|
|
|
id: uuidv4(),
|
|
|
- dateCreated: new Date(),
|
|
|
+ type: `Transport Event`,
|
|
|
+ event: 'New Message',
|
|
|
+ date: new Date(),
|
|
|
+ data: msg,
|
|
|
transport: `Websocket`,
|
|
|
- target: receiverProfileInfo.clientId,
|
|
|
- payload: msg
|
|
|
- } as TransportMessage
|
|
|
- })
|
|
|
- } 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 }) => {
|
|
|
- if (data.name == 'New Profile') {
|
|
|
- console.log({ message: `Assigned client Name: ${data.message.id}` })
|
|
|
- // Update websocket instance record
|
|
|
- receiverProfileInfo = {
|
|
|
- clientId: data.message.id,
|
|
|
- dateCreated: new Date(),
|
|
|
- socketInstance: socket,
|
|
|
- connectionState: new BehaviorSubject<ConnectionState>(`ONLINE`),
|
|
|
- transport: 'Websocket',
|
|
|
- transportServiceId: transportServiceId
|
|
|
+ transportServiceId: transportServiceId
|
|
|
+ } as GeneralEvent<TransportMessage>)
|
|
|
+ } 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
|
|
|
}
|
|
|
- writeFile(data.message as ConnectedSocketServer, receiverProfileInfo.clientId).then(() => {
|
|
|
+ })
|
|
|
+
|
|
|
+ socket.on('handshaking', (data: { id: string, message?: any }) => {
|
|
|
+ // check if this is previously connected or not by checking if the clientId has been previously assigned
|
|
|
+ if (serversConnected.some(obj => obj.clientId === data.id)) {
|
|
|
+ console.log({ message: data.message ?? `No message from server...` })
|
|
|
+ // Update websocket instance record
|
|
|
+ let clientObj: ConnectedSocketServer | undefined = serversConnected.find(obj => obj.clientId === data.id)
|
|
|
+ if (clientObj) {
|
|
|
+ console.log({
|
|
|
+ message: `Just to make sure they are pointed accurately !! Id match? ${clientProfileInfo.clientId == clientObj.clientId ? true : false} && compare ${clientObj.clientId}`,
|
|
|
+ })
|
|
|
+ // broadcast event to allow retransmission to release buffer
|
|
|
+ eventNotification.next({
|
|
|
+ id: uuidv4(),
|
|
|
+ type: `Transport Event`,
|
|
|
+ event: 'Server Connected',
|
|
|
+ date: new Date(),
|
|
|
+ data: clientObj,
|
|
|
+ transport: 'Websocket',
|
|
|
+ transportServiceId: transportServiceId
|
|
|
+ } as GeneralEvent<ClientObject>)
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ console.log({ message: `Server acknowledged for new client connection for ${data.id}` })
|
|
|
+ clientProfileInfo = {
|
|
|
+ clientId: data.id,
|
|
|
+ dateCreated: new Date(),
|
|
|
+ socketInstance: socket
|
|
|
+ }
|
|
|
+ addClientToDB(clientProfileInfo as ConnectedSocketServer, 'servers')
|
|
|
/* Note that there are two separate events, because transmission must first be set up before releasing buffer. */
|
|
|
// broadcast event to allow transmission manager to instantiate transmission components
|
|
|
eventNotification.next({
|
|
|
@@ -133,115 +135,63 @@ export function handleClientSocketConnection(transportServiceId: string, socket:
|
|
|
type: 'Transport Event',
|
|
|
event: `New Server`,
|
|
|
date: new Date(),
|
|
|
- data: receiverProfileInfo,
|
|
|
- transport: 'Websocket'
|
|
|
- })
|
|
|
+ data: clientProfileInfo,
|
|
|
+ transport: 'Websocket',
|
|
|
+ transportServiceId: transportServiceId
|
|
|
+ } as GeneralEvent<ClientObject>)
|
|
|
// broadcast event to allow retransmission to relase buffered messages
|
|
|
eventNotification.next({
|
|
|
id: uuidv4(),
|
|
|
type: 'Transport Event',
|
|
|
event: `Server Connected`,
|
|
|
date: new Date(),
|
|
|
- data: receiverProfileInfo,
|
|
|
- transport: 'Websocket'
|
|
|
- })
|
|
|
- }).catch((error) => { }) // do nothing at the moment.
|
|
|
- serversConnected.push(receiverProfileInfo)
|
|
|
- }
|
|
|
- if (data.name == 'Adjusted Profile') {
|
|
|
- console.log({ message: `Adjusted client Name: ${data.message.id}` })
|
|
|
- // Update websocket instance record
|
|
|
- let clientObj: ConnectedSocketServer | undefined = serversConnected.find(obj => obj.clientId === data.message.id)
|
|
|
- if (clientObj) {
|
|
|
- clientObj.socketInstance = socket
|
|
|
- clientObj.connectionState.next('ONLINE')
|
|
|
- console.log({
|
|
|
- message: `Just to make sure they are pointed accurately: This should be ONLINE: ${receiverProfileInfo.connectionState.getValue()} !! Id match? ${receiverProfileInfo.clientId == clientObj.clientId ? true : false} && compare ${clientObj.clientId}`,
|
|
|
- })
|
|
|
- // broadcast event to allow retransmission to release buffer
|
|
|
+ data: clientProfileInfo,
|
|
|
+ transport: 'Websocket',
|
|
|
+ transportServiceId: transportServiceId
|
|
|
+ } as GeneralEvent<ClientObject>)
|
|
|
+ serversConnected.push(clientProfileInfo)
|
|
|
+ }
|
|
|
+ })
|
|
|
+
|
|
|
+ // Handle disconnection
|
|
|
+ socket.on('disconnect', () => {
|
|
|
+ console.log({ message: `Socket Server ${clientProfileInfo.clientId} Disconnected` })
|
|
|
+ if (clientProfileInfo) {
|
|
|
eventNotification.next({
|
|
|
id: uuidv4(),
|
|
|
type: `Transport Event`,
|
|
|
- event: 'Server Connected',
|
|
|
+ event: `Server Disconnected`,
|
|
|
date: new Date(),
|
|
|
- data: clientObj,
|
|
|
- transport: 'Websocket'
|
|
|
- })
|
|
|
+ data: clientProfileInfo,
|
|
|
+ transport: `Websocket`,
|
|
|
+ transportServiceId: transportServiceId
|
|
|
+ } as GeneralEvent<ClientObject>)
|
|
|
}
|
|
|
- }
|
|
|
- if (data.name == 'Error') {
|
|
|
- console.log({ message: `Server cannot find credentials`, details: data.message })
|
|
|
- // logic to request for new credentials
|
|
|
- setTimeout(() => {
|
|
|
- // for now, if clent cannot be recognize, then we will just proceed as a new client
|
|
|
- socket.emit('profile', {
|
|
|
- name: 'New Client',
|
|
|
- data: null
|
|
|
- })
|
|
|
- }, 2000)
|
|
|
- }
|
|
|
+ });
|
|
|
})
|
|
|
-
|
|
|
- // Handle disconnection
|
|
|
- socket.on('disconnect', () => {
|
|
|
- console.log({ message: `Socket Server ${receiverProfileInfo.clientId} Disconnected` })
|
|
|
- if (receiverProfileInfo) {
|
|
|
- eventNotification.next({
|
|
|
- id: uuidv4(),
|
|
|
- type: `Transport Event`,
|
|
|
- event: `Server Disconnected`,
|
|
|
- date: new Date(),
|
|
|
- data: receiverProfileInfo,
|
|
|
- transport: `Websocket`
|
|
|
- })
|
|
|
- receiverProfileInfo.connectionState.next(`OFFLINE`)
|
|
|
- }
|
|
|
- });
|
|
|
})
|
|
|
}
|
|
|
|
|
|
// For SERVER Usage: set up socket listeners to start listening for different events
|
|
|
export function handleNewSocketClient(transportServiceId: string, socket: SocketForConnectedClient, connectedClientSocket: ConnectedSocketClient[]): Observable<GeneralEvent<any>> {
|
|
|
return new Observable((event: Observer<GeneralEvent<any>>) => {
|
|
|
+ let selfId!: string
|
|
|
+ checkOwnClientInfo(`serverprofile`).then((profile: { id: string }) => {
|
|
|
+ selfId = profile.id
|
|
|
+ })
|
|
|
console.log({ message: `Socket client connected. 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: ConnectedSocketClient = {
|
|
|
- clientId: uuidv4(), // client should only be assigned at this level. And is passed around for reference pointing
|
|
|
- dateCreated: new Date(),
|
|
|
- socketInstance: socket,
|
|
|
- connectionState: new BehaviorSubject<ConnectionState>(`OFFLINE`),
|
|
|
- transport: 'Websocket',
|
|
|
- transportServiceId: transportServiceId
|
|
|
- }
|
|
|
- // send to receiver for reference
|
|
|
- socket.emit('profile', {
|
|
|
- name: `New Profile`, message: { id: clientInstance.clientId }
|
|
|
- })
|
|
|
- // publish first event notification
|
|
|
- event.next({
|
|
|
- id: uuidv4(),
|
|
|
- type: 'Transport Event',
|
|
|
- event: `New Client`,
|
|
|
- date: new Date(),
|
|
|
- data: clientInstance,
|
|
|
- transport: 'Websocket'
|
|
|
- })
|
|
|
- // Update connected clientInstance info to adapter
|
|
|
- connectedClientSocket.push(clientInstance)
|
|
|
- addClientToDB(clientInstance)
|
|
|
- startListening(socket, clientInstance, event)
|
|
|
- } else {
|
|
|
+ socket.on(`handshaking`, (data: { id: string, message?: any }) => {
|
|
|
+ // if it's a previously connected client
|
|
|
+ if (connectedClientSocket.some(obj => obj.clientId == data.id)) {
|
|
|
// update first
|
|
|
let clientInstance: ConnectedSocketClient | undefined
|
|
|
if (connectedClientSocket.length > 0) {
|
|
|
- clientInstance = connectedClientSocket.find(obj => obj.clientId === message.data.id)
|
|
|
+ clientInstance = connectedClientSocket.find(obj => obj.clientId === data.id)
|
|
|
handleFoundClient(clientInstance)
|
|
|
} else {
|
|
|
- // for the case server itself got shit down or something
|
|
|
- checkIfClientExists(message.data.id).then((client: ConnectedSocketClient) => {
|
|
|
+ // for the case server itself got shut down or something. This one will check at persisted local storage
|
|
|
+ checkIfClientExists(data.id).then((client: ConnectedSocketClient) => {
|
|
|
clientInstance = client
|
|
|
handleFoundClient(clientInstance)
|
|
|
}).catch(error => {
|
|
|
@@ -251,57 +201,80 @@ export function handleNewSocketClient(transportServiceId: string, socket: Socket
|
|
|
function handleFoundClient(clientInstance: ConnectedSocketClient | undefined) {
|
|
|
if (clientInstance) {
|
|
|
console.log({ message: `Socket Client ${clientInstance.clientId} Found. This is a previously connected client.` })
|
|
|
- socket.emit('profile', { name: 'Adjusted Profile', message: { id: clientInstance.clientId } })
|
|
|
+ // publish own id to client
|
|
|
+ if (selfId) {
|
|
|
+ socket.emit('handshaking', { id: selfId })
|
|
|
+ } else {
|
|
|
+ console.error({ message: 'Self Id not established' })
|
|
|
+ }
|
|
|
// 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, true)
|
|
|
+ startListening(socket, clientInstance, event, transportServiceId, true)
|
|
|
} else {
|
|
|
console.log({ message: `Profile Not Found` })
|
|
|
- socket.emit('profile', { name: 'Error', message: 'Receiver Profile Not found' })
|
|
|
+ socket.emit('handshaking', { id: selfId, message: 'Receiver Profile Not found' })
|
|
|
}
|
|
|
}
|
|
|
+ } else {
|
|
|
+ console.log({ message: `New Client Connected. ClientID: ${data.id}` })
|
|
|
+ let clientInstance: ConnectedSocketClient = {
|
|
|
+ clientId: data.id,
|
|
|
+ dateCreated: new Date(),
|
|
|
+ socketInstance: socket,
|
|
|
+ }
|
|
|
+ // send to receiver for reference
|
|
|
+ if (selfId) {
|
|
|
+ socket.emit('handshaking', { id: selfId })
|
|
|
+ } else {
|
|
|
+ console.error({ message: 'Self Id not established' })
|
|
|
+ }
|
|
|
+ // publish first event notification
|
|
|
+ event.next({
|
|
|
+ id: uuidv4(),
|
|
|
+ type: 'Transport Event',
|
|
|
+ event: `New Client`,
|
|
|
+ date: new Date(),
|
|
|
+ data: clientInstance,
|
|
|
+ transport: 'Websocket',
|
|
|
+ transportServiceId: transportServiceId
|
|
|
+ } as GeneralEvent<ClientObject>)
|
|
|
+ // Update connected clientInstance info to adapter
|
|
|
+ connectedClientSocket.push(clientInstance)
|
|
|
+ addClientToDB(clientInstance, `clients`)
|
|
|
+ startListening(socket, clientInstance, event, transportServiceId)
|
|
|
}
|
|
|
+
|
|
|
})
|
|
|
})
|
|
|
}
|
|
|
|
|
|
|
|
|
// this is for server usage only
|
|
|
-export function startListening(socket: SocketForConnectedClient, client: ConnectedSocketClient, eventListener: Observer<GeneralEvent<any>>, oldClient?: boolean): void {
|
|
|
+export function startListening(socket: SocketForConnectedClient, client: ConnectedSocketClient, eventListener: Observer<GeneralEvent<any>>, transportServiceid: string, oldClient?: boolean): void {
|
|
|
// notify it's associated retransmission to start releaseing buffer
|
|
|
eventListener.next({
|
|
|
id: uuidv4(),
|
|
|
type: 'Transport Event',
|
|
|
- event: oldClient ? 'Client Re-connected' : `Client Connected`,
|
|
|
+ event: oldClient ? `Client Re-connected` : `Client Connected`,
|
|
|
date: new Date(),
|
|
|
data: client,
|
|
|
- transport: 'Websocket'
|
|
|
- })
|
|
|
- // Resume operation
|
|
|
- // some explanation here. For the case where the server reads from the DB, no need to terminate subject, since all instances would be destroyed alongside the server shut down. This case is specificd only when there's a need to read from local file
|
|
|
- if (!client.connectionState) {
|
|
|
- client.connectionState = new BehaviorSubject<ConnectionState>(`ONLINE`)
|
|
|
- } else {
|
|
|
- client.connectionState.next(`ONLINE`)
|
|
|
- }
|
|
|
+ transport: 'Websocket',
|
|
|
+ transportServiceid: transportServiceid
|
|
|
+ } as unknown as GeneralEvent<ClientObject>)
|
|
|
|
|
|
/* Generally, we don't need this unless in the case of being the receiver */
|
|
|
- socket.on('message', (message: any) => {
|
|
|
+ socket.on('message', (message: TransportMessage) => {
|
|
|
console.log({ message: `Message from client ${client.clientId}`, details: message })
|
|
|
eventListener.next({
|
|
|
id: uuidv4(),
|
|
|
type: 'Transport Event',
|
|
|
event: 'New Message',
|
|
|
date: new Date(),
|
|
|
- data: {
|
|
|
- id: uuidv4(),
|
|
|
- dateCreated: new Date(),
|
|
|
- transport: `Websocket`,
|
|
|
- target: client.clientId, // this ref to be associated with the client/channel
|
|
|
- payload: message
|
|
|
- } as TransportMessage
|
|
|
- })
|
|
|
+ data: message,
|
|
|
+ transport: `Websocket`,
|
|
|
+ transportServiceId: transportServiceid
|
|
|
+ } as GeneralEvent<TransportMessage>)
|
|
|
})
|
|
|
|
|
|
socket.on('disconnect', () => {
|
|
|
@@ -311,8 +284,9 @@ export function startListening(socket: SocketForConnectedClient, client: Connect
|
|
|
event: 'Client Disconnected',
|
|
|
date: new Date(),
|
|
|
data: client,
|
|
|
- transport: 'Websocket'
|
|
|
- })
|
|
|
+ transport: 'Websocket',
|
|
|
+ transportServiceId: transportServiceid
|
|
|
+ } as GeneralEvent<ClientObject>)
|
|
|
eventListener.error(`Client ${client.clientId} disconnected. Terminating this observable event for this client socket...`)
|
|
|
eventListener.complete()
|
|
|
})
|
