index.js

const { mqtt } = require('aws-iot-device-sdk-v2');
const { send } = require('process');
const { TextDecoder } = require('util');
const path = require("path");
const fork = require("child_process").fork;

const yargs = require('yargs');

const common_args = require('./cli_args');

// Add command line arguments for direct connection establishment and topic message
// arguments to yargs
yargs.command('*', false, (yargs) => {
    common_args.add_direct_connection_establishment_arguments(yargs);
    common_args.add_topic_message_arguments(yargs);
}, main).parse();

let fusionData = {};

// Execute a session with the given connection and arguments
async function monitorCycle(connection, argv) {
    return new Promise(async (resolve, reject) => {
        try {
            let published = false;
            let subscribed = false;
            const decoder = new TextDecoder('utf8');

            // Define a function to handle incoming published messages
            const onAWSmessage = async (topic, payload, dup, qos, retain) => {
                const messageData = decoder.decode(payload);
                console.log(`[AWS] Message received from AWS. topic:"${topic}" dup:${dup} qos:${qos} retain:${retain}`);
                console.log(`[AWS] Payload: ${messageData}`);
                try {
                    const message = JSON.parse(messageData);
                    if (message.sequence == argv.count) {
                        subscribed = true;
                        if (subscribed && published) {
                            resolve();
                        }
                    }
                }
                catch (error) {
                    console.log("[AWS] Warning: Could not parse message as JSON...");
                }
            }

            // Subscribe to the specified topic with the given QoS and message handler
            console.log("[AWS] Subscribing to: " + argv.topic)
            await connection.subscribe(argv.topic, mqtt.QoS.AtLeastOnce, onAWSmessage);
            console.log("[AWS] Subscribed to: " + argv.topic)

            let published_counts = 0;

            // Publish the specified number of messages to the topic
            const sendMessage = async (data) => {
                const json = JSON.stringify(data);
                console.log("[AWS] Sending message")

                // Publish the message to the topic with the given QoS
                connection.publish(argv.topic, json, mqtt.QoS.AtLeastOnce).then(() => {
                    console.log("[AWS] Message Sent")
                    ++published_counts;
                    if (published_counts == argv.count) {
                        published = true;
                        if (subscribed && published) {
                            resolve();
                        }
                    }
                })
            }

            console.log("[SRV] Starting Fusion sniffer");
            const sniffaSrc = path.resolve(__dirname, 'fusionSniffer.js');
            const options = {
                stdio: ['pipe', 'pipe', 'pipe', 'ipc'],
                silent: true
            };

            const sniffa = fork(sniffaSrc, [], options);

            sniffa.on('message', fusionData => {
                console.log("[Server] Update from Sniffa")
                console.log(fusionData)
                newmsg = {
                    "state": {
                      "desired": {
                        "siteID": "[AA00000]",
                        "siteName": "SSG - Four Bed Bay",
                        "connectedNodes": 0,
                        "errors": {},
                        "connectedSystems": {},
                        "currentEvents": {},
                        "eventHistory": [],
                        "systemLog": [],
                        "metadata": {}
                      },
                      "reported": {
                        "siteID": "[AA00000]",
                        "siteName": "SSG - Four Bed Bay",
                        "connectedNodes": 0,
                        "errors": {
                            "testError": 1
                        },
                        "connectedSystems": fusionData.devices,
                        "currentEvents": fusionData.liveevents,
                        "eventHistory": fusionData.eventhistory,
                        "systemLog": fusionData.log.fusion,
                        "metadata": fusionData.meta
                      }
                    }
                  }
                sendMessage(newmsg);
            });

            sniffa.stdout.on('data', (data) => {
                console.log(`[Sniffa]${data}`);
            });
            sniffa.stderr.on('data', (data) => {
                console.log(`[ERROR][Sniffa]${data}`);
            });

            const payload = {
                "state": {
                    "desired": {
                        "welcome": "aws-iot",
                        "connectedNodes": 0,
                        "errors": {},
                        "connectedSystems": {},
                        "currentEvent": {},
                        "lastEvent": {}
                    },
                    "reported": {
                        "welcome": "aws-iot",
                        "connectedNodes": 0,
                        "errors": {},
                        "connectedSystems": {},
                        "currentEvent": {},
                        "lastEvent": {}
                    }
                }
            };
            sendMessage(payload)
        }
        catch (error) {
            console.log("[AWS] " + error)
            reject(error);
        }
    });
}

// Main function to execute the program
async function main(argv) {
    common_args.apply_sample_arguments(argv);

    // Build a connection from the command line arguments
    const connection = common_args.build_connection_from_cli_args(argv);

    console.log("[AWS] Connecting to AWS network...")

    // Connect to the AWS network
    await connection.connect()

    console.log("[AWS] Connected to AWS!");

    // Execute a session with the built connection and command line arguments
    await monitorCycle(connection, argv)
}