Fis-SearchUtility/test/consumer_1.ts

import { SearchService } from "../services/query.service"
import _ = require("lodash")
import { Subject, interval } from "rxjs";
import * as WebSocket from 'ws';

/* ---------------------- COMPLEX OPERATION ------------------------------ */
// Check Heap size
// const v8 = require('v8');
// const heapStats = v8.getHeapStatistics();
// const maxHeapSize = heapStats.heap_size_limit / (1024 * 1024); // Convert to MB
// console.log(`Current maximum heap size: ${maxHeapSize.toFixed(2)} MB`);
// const used = process.memoryUsage().heapUsed / 1024 / 1024;
// const total = process.memoryUsage().heapTotal / 1024 / 1024;
// console.log(`Heap memory usage: ${used.toFixed(2)} MB`);
// console.log(`Total heap size: ${total.toFixed(2)} MB`);

function checkHeapSize(): any {
    let currentHeapSize = process.memoryUsage().heapUsed / 1024 / 1024;
    let allocatedHeapSize = 512;
    let percentage = (currentHeapSize / allocatedHeapSize) * 100;
    console.log(`Consumer_! Heap currentHeapSize: ${currentHeapSize} MB. Percentage: ${percentage}`);

    return percentage
}

// Create new Subject to handle incoming data from remote subscription
let payload: Subject<any> = new Subject()
payload.subscribe((element) => {
    console.log(`Received message from server: ${element.appData.msgId}`);
    // tell traffic control to check heap size
    // trafficControl.next(checkHeapSize())
})

// Create new Subject to monitor and broadcast heap size 
let trafficControl: Subject<any> = new Subject()
let intervalChecking = interval(1000)
intervalChecking.subscribe(() => {
    trafficControl.next(checkHeapSize())
})

// Create a WebSocket server
function createWebsocketServer() {
    const wss = new WebSocket.Server({ port: 8081 });

    // Listen for connections to the WebSocket server
    wss.on('connection', (ws: WebSocket) => {
        console.log('Client connected');

        // Subscribe to the subject when a client connects
        const subscription = trafficControl.subscribe((element) => {
            // Stringify heap status and send data over to connected client
            const messageString = JSON.stringify(element);
            ws.send(messageString);
        });

        // Listen for messages from the client
        ws.on('message', (message: any) => {
            console.log(`Received message from client: ${message}`);
        });

        // Unsubscribe from the subject when the client disconnects
        ws.on('close', () => {
            console.log('Client disconnected');
            subscription.unsubscribe();
        });
    });
}


// Create a new WebSocket client
function connectWebSocket() {
    const ws = new WebSocket('ws://localhost:8080');

    // Listen for the WebSocket connection to open
    ws.on('open', () => {
        console.log('Connecting to Publisher WebSocket server');

        // Send a message to the server
        ws.send('Hello, publisher server!');
    })

    // Listen for messages from the server
    ws.on('message', (message: string) => {
        let msgObj: any[] = JSON.parse(message)
        msgObj.forEach(element => {
            payload.next(element)
        });
    });

    // Listen for WebSocket errors
    ws.on('error', (element) => {
        console.error('WebSocket error:', element);
    })

    // Listen for the WebSocket connection to close
    ws.on('close', () => {
        console.log('Disconnected from WebSocket server');
        setTimeout(connectWebSocket, 1000); // Attempt to reconnect after 1 second

    })

}

createWebsocketServer()
connectWebSocket();