index.ts

/**
 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 * SPDX-License-Identifier: Apache-2.0.
 */

import{mqtt}from'aws-iot-device-sdk-v2';
import{TextDecoder}from'util';

typeArgs={[index: string]: any};

constyargs=require('yargs');

// The relative path is '../../util/cli_args' from here, but the compiled javascript file gets put one level
// deeper inside the 'dist' folder
constcommon_args=require('../../../util/cli_args');

yargs.command('*',false,(yargs: any)=>{
common_args.add_direct_connection_establishment_arguments(yargs);
common_args.add_topic_message_arguments(yargs);
},main).parse();

asyncfunctionexecute_session(connection: mqtt.MqttClientConnection,argv: Args){
returnnewPromise<void>(async(resolve,reject)=>{
try{
letpublished=false;
letsubscribed=false;
constdecoder=newTextDecoder('utf8');
conston_publish=async(topic: string,payload: ArrayBuffer,dup: boolean,qos: mqtt.QoS,retain: boolean)=>{
constjson=decoder.decode(payload);
console.log(`Publish received. topic:"${topic}" dup:${dup} qos:${qos} retain:${retain}`);
console.log(`Payload: ${json}`);
try{
constmessage=JSON.parse(json);
if(message.sequence==argv.count){
subscribed=true;
if(subscribed&&published){
resolve();
}
}
}
catch(error){
console.log("Warning: Could not parse message as JSON...");
}
}

awaitconnection.subscribe(argv.topic,mqtt.QoS.AtLeastOnce,on_publish);
letpublished_counts=0;
for(letop_idx=0;op_idx<argv.count;++op_idx){
constpublish=async()=>{
constmsg={
message: argv.message,
sequence: op_idx+1,
};
constjson=JSON.stringify(msg);
connection.publish(argv.topic,json,mqtt.QoS.AtLeastOnce).then(()=>{
++published_counts;
if(published_counts==argv.count){
published=true;
if(subscribed&&published){
resolve();
}
}
})
}
setTimeout(publish,op_idx*1000);
}
}
catch(error){
reject(error);
}
});
}

asyncfunctionmain(argv: Args){
common_args.apply_sample_arguments(argv);

constconnection=common_args.build_connection_from_cli_args(argv);

// force node to wait 60 seconds before killing itself, promises do not keep node alive
// ToDo: we can get rid of this but it requires a refactor of the native connection binding that includes
// pinning the libuv event loop while the connection is active or potentially active.
consttimer=setInterval(()=>{},60*1000);

awaitconnection.connect()
awaitexecute_session(connection,argv)
awaitconnection.disconnect()

// Allow node to die if the promise above resolved
clearTimeout(timer);
}