Use the following code to create a producer.js file. This file holds the actual application. Find descriptions of what it does as comments in the coding.
On a higher level you do the following:
-
Perform settings in respect to Node.js modules and to messaging.
-
Get the messaging options from the environment.
-
The messaging client is instantiated using the options.
-
Client and stream handler methods are defined.
As a result, the application sends messages to the defined topic and writes a quick info including a counter into the log file.
use strict;
//------------------------------------------------------------------------------------------------------------------
// Basic setup in respect to modules, messaging settings and getting messaging options
//------------------------------------------------------------------------------------------------------------------
const msg = require('@sap/xb-msg');
const env = require('@sap/xb-msg-env');
const xsenv = require('@sap/xsenv');
const service = '<REPLACE WITH YOUR MESSAGING SERVICE>';
const taskList = {
myOutA : { topic: '<REPLACE WITH YOUR TOPIC>' , timerMin: 1, timerMax: 11 },
myOutB : { topic: '<REPLACE WITH YOUR TOPIC>' , timerMin: 5, timerMax: 8 }
};
var counter = 1;
xsenv.loadEnv();
//------------------------------------------------------------------------------------------------------------------
// Start messaging client
//------------------------------------------------------------------------------------------------------------------
const client = new msg.Client(env.msgClientOptions(service, [], ['myOutA', 'myOutB']));
function setupOptions(tasks, options) {
Object.getOwnPropertyNames(tasks).forEach((id) => {
const task = tasks[id];
options.destinations[0].ostreams[id] = {
channel: 1,
exchange: 'amq.topic',
routingKey: task.topic
};
});
return options;
}
function getRandomInt(min, max) {
min = Math.ceil(min);
max = Math.floor(max);
return (Math.floor(Math.random() * (max - min + 1)) + min) * 1000;
}
function initTasks(tasks, client) {
Object.getOwnPropertyNames(tasks).forEach((id) => {
const task = tasks[id];
const stream = client.ostream(id);
const handler = () => {
console.log('publishing message number ' + counter + ' to topic ' + task.topic);
const message = {
target: { address: 'topic:' + task.topic },
payload: Buffer.from("Message Number " + counter)
};
if (!stream.write(message)) {
console.log('wait');
return;
}
setTimeout(handler, getRandomInt(task.timerMin, task.timerMax));
counter++;
};
stream.on('drain', () => {
setTimeout(handler, getRandomInt(task.timerMin, task.timerMax));
});
setTimeout(handler, getRandomInt(task.timerMin, task.timerMax));
});
}
//------------------------------------------------------------------------------------------------------------------
// Messaging client handler methods
//------------------------------------------------------------------------------------------------------------------
client
.on('connected', () => {
console.log('connected');
initTasks(taskList, client);
})
.on('drain', () => {
console.log('continue');
})
.on('error', (error) => {
console.log(error);
});
client.connect();
