Fis-Retransmission/services/grpc.service.method.ts

import * as grpc from '@grpc/grpc-js';
import { Subject, Subscription } from "rxjs";
import { ReportStatus, ColorCode, Message, MessageLog, ConnectionAttribute, ConnectionRequest, GrpcConnectionType, ConnectionState } from "../interfaces/general.interface";
import { Status } from '@grpc/grpc-js/build/src/constants';
import { v4 as uuidv4 } from 'uuid'
import { message_proto } from './protos/server.proto'
import { ServerWritableStreamImpl } from '@grpc/grpc-js/build/src/server-call';
export class GrpcServiceMethod {
    private server: grpc.Server | any
    private messageToBeSendOver: Message | any
    private callRequestsFromRemote: ServerWritableStreamImpl<any, ResponseType>[] = []

    public async create(request: ConnectionRequest, connectionAttribute: ConnectionAttribute): Promise<any> {

        // Assuming currently only one client
        this.createGrpcInstance(request.server.serverUrl, { instanceType: 'server' }, connectionAttribute)
        this.createGrpcInstance(request.client.targetServer, { instanceType: 'client' }, connectionAttribute)
        // connectionAttribute.outGoing.MessageToBePublished?.subscribe(e => console.log((e.message as MessageLog).appData.msgId))
    }

    // For testing only
    public async shutDownServer(): Promise<string> {
        return new Promise((resolve, reject) => {
            console.log(`Shutting down servers...`)
            if (this.server) {
                this.callRequestsFromRemote[0].destroy()
                this.callRequestsFromRemote[0].end()
                this.server.forceShutdown()
                let message: string = `Server shut down successfully!`
                resolve(message)
            }
            if (!this.server) {
                let errorMsg: string = `There's no active server here`
                reject(errorMsg)
            }
        })
    }

    private async generateAdditionalAttributes(connectionAttribute: ConnectionAttribute, clientInfo?: any, localInfo?: any) {
        if (clientInfo) {
            connectionAttribute.inComing.StreamID = clientInfo.channelID
            connectionAttribute.inComing.PublisherID = clientInfo.publisherID
            connectionAttribute.inComing.SubscriberID = clientInfo.subscriberID
        }
        if (localInfo) {
            connectionAttribute.outGoing.StreamID = localInfo.channelID
            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
            // let report: ReportStatus = {
            //     code: ColorCode.GREEN,
            //     message: `ConnectionID acquired. Informing Restranmission to release Messages...`,
            // }

        }
    }

    // To be migrated into a service in the immediate future
    private async createGrpcInstance(
        serverUrl: string,
        grpcType: GrpcConnectionType,
        connectionAttribute: ConnectionAttribute,
    ) {
        let statusControl: Subject<ConnectionState> = connectionAttribute.connectionStatus
        let consecutiveResolutions = 0;
        let lastResolutionTime = Date.now();
        let yellowErrorEmission: boolean = false
        let redErrorEmission: boolean = false

        while (true) {
            try {
                let recreatePromise = new Promise((resolve) => {
                    if (grpcType.instanceType == 'server') {
                        this.createServerStreamingServer(serverUrl, connectionAttribute).then(() => {
                            resolve('recreate')
                        })
                    }
                    if (grpcType.instanceType == 'client') {
                        this.createServerStreamingClient(serverUrl, connectionAttribute).then(() => {
                            resolve('recreate')
                        })
                    }
                })
                await recreatePromise
                // If connection resolves (indicating failure), increment the count
                consecutiveResolutions++;
                // console.log(`Reconnection Attempt: ${consecutiveResolutions}`)
                if (redErrorEmission == false) {
                    redErrorEmission = true
                    console.error(`Connection failed ${consecutiveResolutions} times. Stopping connection attempts.`);
                    // let error: ReportStatus = {
                    //     code: ColorCode.YELLOW,
                    //     message: 'Server is not responding. Proceed to buffer.',
                    // }
                    let error: ConnectionState = {
                        status: 'BUFFER',
                        reason: `Server is not responding...`
                    }
                    statusControl.next(error)
                }

            } catch (error) {
                // Connection did not resolve, reset the count
                consecutiveResolutions = 0;
                console.error('Connection attempt failed:', error);
            }
            // Check for a pause of more than 2 seconds since the last resolution attempt
            const currentTime = Date.now();
            const timeSinceLastResolution = currentTime - lastResolutionTime;
            if (timeSinceLastResolution > 2000) {
                consecutiveResolutions = 0;
                yellowErrorEmission = false
                redErrorEmission = false
            }
            // Update the last resolution time
            lastResolutionTime = currentTime;
            await new Promise(resolve => setTimeout(resolve, 1000)); // Wait for 1 second before the next attempt
            // timeout generate message to trigger this reconnection
        }
    }

    // Create Server Instance to stream all application Outgoing messages
    public async createServerStreamingServer(
        serverUrl: string,
        connectionAttribute: ConnectionAttribute
    ): Promise<any> { // '0.0.0.0:3001'
        return new Promise((resolve, reject) => {
            try {
                if (!this.server) {
                    this.server = new grpc.Server();
                }

                this.server.addService(message_proto.Message.service, {
                    HandleMessage: (call) => {
                        this.callRequestsFromRemote.push(call)
                        let clientInfo = JSON.parse(call.request.message)
                        this.generateAdditionalAttributes(connectionAttribute, 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} `)
                                    this.messageToBeSendOver = response
                                    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('')
                                }
                            })
                        }

                        console.log(connectionAttribute)
                        let report: ConnectionState = {
                            status: 'DIRECT_PUBLISH'
                        }
                        connectionAttribute.connectionStatus.next(report)

                    },

                    Check: (_, callback) => {
                        // for now it is just sending the status message over to tell the client it is alive
                        // For simplicity, always return "SERVING" as status
                        callback(null, { status: 'SERVING' });
                    },
                });
                // Bind and start the server
                this.server.bindAsync(serverUrl, grpc.ServerCredentials.createInsecure(), () => {
                    console.log(`gRPC server is running on ${serverUrl}`);
                    this.server.start();
                });
            }
            catch (error) {
                resolve(error)
            }

        })
    }

    // Send a request over to the other server to open a channel for this server to emit/stream messages over
    public async createServerStreamingClient(
        server: string,
        connectionAttribute: ConnectionAttribute
    ): Promise<string> {
        return new Promise(async (resolve, reject) => {
            const client = new message_proto.Message(server, grpc.credentials.createInsecure());

            let outGoingInfo: any = {
                channelID: uuidv4(),
                publisherID: uuidv4(),
                subscriberID: uuidv4()
            }
            this.generateAdditionalAttributes(connectionAttribute, {}, outGoingInfo)

            // connectionAttribute.ConnectionID = connectionAttribute.outGoing.ChannelID + (connectionAttribute.inComing.ChannelID ?? 'undefined')
            let call = client.HandleMessage({ id: server, message: JSON.stringify(outGoingInfo) })
            console.log(`Sending request to ${server} to open response channel...`)

            call.on('status', (status: Status) => {
                if (status == grpc.status.OK) { // only returns a status when there's error. Otherwise it just waits
                    console.log(`Message trasmission operation is successful`)
                    // RPC completed successfully
                } if (status == grpc.status.UNAVAILABLE) {
                    // let report: ReportStatus = {
                    //     code: ColorCode.YELLOW,
                    //     message: `Server doesn't seem to be alive. Error returned.`,
                    //     payload: this.messageToBeSendOver ?? `There's no message at the moment...`
                    // }
                    let report: ConnectionState = {
                        status: 'BUFFER',
                        reason: `Server doesn't seem to be alive. Error returned.`,
                        payload: this.messageToBeSendOver ?? `There's no message at the moment...`
                    }
                    connectionAttribute.connectionStatus.next(report)
                    resolve('No connection established. Server is not responding..')
                }
            });

            call.on('data', (data: any) => {
                let response: Message = {
                    id: data.id,
                    message: JSON.parse(data.message)
                }
                if (connectionAttribute.inComing.MessageToBeReceived) {
                    connectionAttribute.inComing.MessageToBeReceived.next(response)
                }
                // console.log(`Received ${(response.message as MessageLog).appData.msgId}`)
            });

            call.on('error', (err) => {
                console.error(err)
                resolve('')
            });
        })
    }


    // THis is no longer necesarry after the introduction of connection Attribute. But it is still useful for checking for the other side's health
    public async checkConnectionHealth(client: any, statusControl: Subject<ReportStatus>, alreadyHealthCheck: boolean): Promise<boolean> {
        return new Promise((resolve, reject) => {
            client.Check({}, (error, response) => {
                if (response) {
                    console.log(`GRPC Health check status: ${response.status} Server Connected`);
                    // Intepret the response status and implement code logic or handler
                    resolve(response.status)
                } else {
                    if (alreadyHealthCheck == false) console.error(`Health check failed: ${error}`);
                    reject(false)
                }
            })
        })
    }
}



// https://github.com/grpc/proposal/blob/master/L5-node-client-interceptors.md