added client call request validation

services/grpc.service.method.ts

@@ -1,6 +1,6 @@
 import * as grpc from '@grpc/grpc-js';
-import { Subject, Subscription } from "rxjs";
-import { Message, ConnectionAttribute, GrpcConnectionType, ConnectionState, MessageLog, ConnectionStatus } from "../interfaces/general.interface";
+import { Observable, Subject, Subscription } from "rxjs";
+import { Message, ConnectionAttribute, GrpcConnectionType, ConnectionState, MessageLog, ConnectionStatus, StreamAttribute } from "../interfaces/general.interface";
 import { Status } from '@grpc/grpc-js/build/src/constants';
 import { message_proto } from './protos/server.proto'
 import * as _ from 'lodash'
@@ -27,24 +27,6 @@ export class GrpcServiceMethod {
             resolve('Just putting it here for now....')
         })
     }
-
-    private async generateAdditionalAttributes(connectionAttribute: ConnectionAttribute, clientInfo?: any, localInfo?: any) {
-        // if (clientInfo) {
-        //     connectionAttribute.inComing.StreamID = clientInfo.StreamID
-        //     connectionAttribute.inComing.PublisherID = clientInfo.PublisherID
-        //     connectionAttribute.inComing.SubscriberID = clientInfo.SubscriberID
-        // }
-        // if (localInfo) {
-        //     connectionAttribute.outGoing.StreamID = localInfo.StreamID
-        //     connectionAttribute.outGoing.PublisherID = localInfo.PublisherID
-        //     connectionAttribute.outGoing.SubscriberID = localInfo.SubscriberID
-        // }
-        // if (connectionAttribute.outGoing.StreamID && connectionAttribute.inComing.StreamID) {
-        //     connectionAttribute.ConnectionID.local = connectionAttribute.outGoing.StreamID + connectionAttribute.inComing.StreamID
-        //     connectionAttribute.ConnectionID.remote = connectionAttribute.inComing.StreamID + connectionAttribute.outGoing.StreamID
-        // }
-    }
-
     private async createGrpcInstance(grpcType: GrpcConnectionType, connectionAttribute: ConnectionAttribute) {
         // Reconnection Logic here
         while (true) {
@@ -78,45 +60,55 @@ export class GrpcServiceMethod {
                     this.server = new grpc.Server()
                     this.localServerStatus.next({
                         connectionStatus: 'ON',
-                        serverName: connectionAttribute.outGoing.PublisherID,
+                        connectionIDlocal: connectionAttribute.ConnectionID.local,
                         message: `${connectionAttribute.outGoing.serverUrl} started.`
                     })
                 } else {
                     console.log(`Grpc server alrady started.`)
+                    this.localServerStatus.next({
+                        connectionStatus: 'ON',
+                        connectionIDlocal: connectionAttribute.ConnectionID.local,
+                        message: `${connectionAttribute.outGoing.serverUrl} already started.`
+                    })
                 }
 
+
                 this.server.addService(message_proto.Message.service, {
                     HandleMessage: (call) => {
+                        /// add a checking for standard message request
                         let clientInfo: ConnectionAttribute = JSON.parse(call.request.message)
-                        this.clientRequest.next(clientInfo)
-
-                        console.log(`Initializing stream. Opening Channel... Confirmation from ${call.request.id}`)
-
-                        if (connectionAttribute.outGoing.MessageToBePublished) {
-                            let subscription: Subscription = connectionAttribute.outGoing.MessageToBePublished.subscribe({
-                                next: (response: Message) => {
-                                    console.log(`Sending from GRPC server: ${(response.message as MessageLog).appData.msgId} `)
-                                    let message = {
-                                        id: response.id,
-                                        message: JSON.stringify(response.message)
+                        if (this.isConnectionAttribute(clientInfo)) {
+                            this.clientRequest.next(clientInfo)
+                            console.log(`Initializing stream. Opening Channel... Confirmation from ${call.request.id}`)
+
+                            if (connectionAttribute.outGoing.MessageToBePublished) {
+                                let subscription: Subscription = connectionAttribute.outGoing.MessageToBePublished.subscribe({
+                                    next: (response: Message) => {
+                                        console.log(`Sending from GRPC server: ${(response.message as MessageLog).appData.msgId} `)
+                                        let message = {
+                                            id: response.id,
+                                            message: JSON.stringify(response.message)
+                                        }
+                                        call.write(message)
+                                    },
+                                    error: err => {
+                                        console.error(err)
+                                        subscription.unsubscribe()
+                                        resolve('')
+                                    },
+                                    complete: () => {
+                                        console.log(`Stream response completed for ${call.request.id}`)
+                                        subscription.unsubscribe()
+                                        resolve('')
                                     }
-                                    call.write(message)
-                                },
-                                error: err => {
-                                    console.error(err)
-                                    subscription.unsubscribe()
-                                    resolve('')
-                                },
-                                complete: () => {
-                                    console.log(`Stream response completed for ${call.request.id}`)
-                                    subscription.unsubscribe()
-                                    resolve('')
+                                })
+                                let report: ConnectionState = {
+                                    status: 'DIRECT_PUBLISH'
                                 }
-                            })
-                            let report: ConnectionState = {
-                                status: 'DIRECT_PUBLISH'
+                                connectionAttribute.connectionStatus!.next(report)
                             }
-                            connectionAttribute.connectionStatus!.next(report)
+                        } else {
+                            console.error(`INVALID REQUEST`)
                         }
                     },
                     Check: (_, callback) => {
@@ -213,6 +205,38 @@ export class GrpcServiceMethod {
             })
         })
     }
+
+
+    private isConnectionAttribute(obj: any): obj is ConnectionAttribute {
+        const isMatch = (
+            typeof obj.ConnectionID === 'object' && // Further checks can be added based on the structure of ConnectionID
+            isStreamAttribute(obj.outGoing) &&
+            isStreamAttribute(obj.inComing) &&
+            (obj.connectionStatus === null || obj.connectionStatus instanceof Subject)
+        );
+
+        if (isMatch) {
+            console.log('gRPC client call matches ConnectionAttribute type');
+        } else {
+            console.log('gRPC client call does not match ConnectionAttribute type');
+        }
+
+        return isMatch;
+        function isStreamAttribute(obj: any): obj is StreamAttribute {
+            return (
+                (typeof obj.StreamID === 'string' || obj.StreamID === undefined) &&
+                (typeof obj.PublisherID === 'string' || obj.PublisherID === undefined) &&
+                (typeof obj.SubscriberID === 'string' || obj.SubscriberID === undefined) &&
+                // Check other properties like PublisherInstance and SubscriberInstance based on their expected types
+                (typeof obj.serverUrl === 'string' || obj.serverUrl === undefined) &&
+                // Check connectionState based on its type, assuming it's an enum or similar
+                (obj.MessageToBePublished === null || obj.MessageToBePublished instanceof Observable) &&
+                (obj.MessageToBeReceived === null || obj.MessageToBeReceived instanceof Subject)
+            );
+        }
+    }
+
+
 }
 
 

services/server-client.service.ts

@@ -23,30 +23,6 @@ export class ServerClientManager {
     }
 
     constructor() {
-        this.grpcService.getLocalServerStatus().subscribe({
-            next: (notification: any) => {
-                console.log(`Received a update from Grpc Server methods`)
-                if (notification.connectionStatus === `ON`) {
-                    let connectionAttribute = this.connectionAttributes.find(connection => connection.outGoing.PublisherID == notification.serverName)
-                    connectionAttribute!.outGoing.connectionState = 'ON'
-                    console.log(notification.message)
-                }
-            },
-            error: err => console.error(err),
-            complete: () => { }
-        })
-        this.grpcService.getClientRequest().subscribe({
-            next: (clientConnectionAttribute: ConnectionAttribute) => {
-                console.log(`Received a request from ${clientConnectionAttribute.outGoing.serverUrl}`)
-                let connectionAttribute = this.connectionAttributes.find(connection => connection.ConnectionID.remote == clientConnectionAttribute.ConnectionID.local)
-                connectionAttribute!.inComing.connectionState = 'ON' 
-                // console.log(clientConnectionAttribute)
-                console.log(`Updated!!! look down...`)
-                console.log(this.connectionAttributes.find(connection => connection.ConnectionID.remote == clientConnectionAttribute.ConnectionID.local))
-            },
-            error: err => console.error(err),
-            complete: () => { }
-        })
     }
 
     public async generateConnection(request: ConnectionRequest): Promise<any> {
@@ -82,7 +58,6 @@ export class ServerClientManager {
             if (statusChain == 1) {
                 // Connection Type checking
                 if (request.server!.connectionType != request.client!.connectionType) {
-                    // console.log(`Connection Type DOES NOT MATCH!`)
                     statusChain = -1
                     errorString = "Connection Type DOES NOT MATCH!"
                 } else {
@@ -138,6 +113,8 @@ export class ServerClientManager {
                 }
                 // For each connection type:
                 if (request.client!.connectionType == 'GRPC') {
+                    this.grpcServerStatusHandler()
+                    this.grpcClientStatusHandler()
                     this.grpcService.create(connectionAttribute!).then(() => {
                         // logic here
                     }).catch(() => {
@@ -170,11 +147,46 @@ export class ServerClientManager {
             let result: boolean = this.connectionAttributes.some(connection =>
                 connection.ConnectionID.local === connectionAttribute.ConnectionID.local
             );
-            // console.log(`Checking ${connectionAttribute.ConnectionID.local} and returns ${result}`);
             resolve(result);
         });
     }
 
+    private grpcServerStatusHandler() {
+        this.grpcService.getLocalServerStatus().subscribe({
+            next: (notification: any) => {
+                if (notification.connectionStatus === `ON`) {
+                    let connectionAttribute = this.connectionAttributes.find(connection => connection.ConnectionID.local == notification.connectionIDlocal)
+                    if (connectionAttribute) {
+                        connectionAttribute.outGoing.connectionState = 'ON'
+                        console.log(`Connection ${notification.connectionIDlocal} updated`)
+                    } else {
+                        console.log(`Connection ${notification.connectionIDlocal} attribute is not found.`)
+                    }
+                }
+            },
+            error: err => console.error(err),
+            complete: () => { }
+        })
+    }
 
+    private grpcClientStatusHandler() {
+        this.grpcService.getClientRequest().subscribe({
+            next: (clientConnectionAttribute: ConnectionAttribute) => {
+                console.log(`Received a request from ${clientConnectionAttribute.outGoing.serverUrl}`)
+                let connectionAttribute = this.connectionAttributes.find(connection => connection.ConnectionID.remote == clientConnectionAttribute.ConnectionID.local)
+                if (connectionAttribute) {
+                    console.log(`Connection ${clientConnectionAttribute.ConnectionID.local} updated`)
+                    connectionAttribute.inComing.connectionState = 'ON'
+                } else {
+                    console.log(`Connection Attribut ${clientConnectionAttribute.inComing.PublisherID} is not found`)
+                }
+            },
+            error: err => console.error(err),
+            complete: () => { }
+        })
+    }
 }
 
+
+
+

test/test2.ts

@@ -1,5 +1,5 @@
 /* call server client service instance 3 times.  */
-
+// Testing to isolate server client servicce
 import { Subject } from "rxjs";
 import { ConnectionRequest, Message } from "../interfaces/general.interface";
 import { ServerClientManager } from "../services/server-client.service";