1. /**
  2. * Exports message queue results for manual execution.
  3. * Handles edge cases like empty queues and invalid data.
  4. *
  5. * @param {object} queue - The message queue object (e.g., RabbitMQ, Kafka).
  6. * @param {string} filename - The name of the file to export to.
  7. * @param {function} processMessage - Function to execute on each message.
  8. * @returns {Promise<void>} - A promise that resolves when the export is complete.
  9. */
  10. async function exportQueueResults(queue, filename, processMessage) {
  11. try {
  12. // Check if queue is valid
  13. if (!queue || typeof queue.consume !== 'function') {
  14. console.error("Invalid queue object provided.");
  15. return Promise.reject(new Error("Invalid queue object"));
  16. }
  18. // Check if filename is valid
  19. if (!filename || typeof filename !== 'string' || filename.trim() === '') {
  20. console.error("Invalid filename provided.");
  21. return Promise.reject(new Error("Invalid filename"));
  22. }
  24. const results = [];
  25. let message;
  27. // Consume messages until the queue is empty or an error occurs
  28. while ((message = queue.consume()) !== null) {
  29. try {
  30. // Process the message
  31. const result = processMessage(message); //Execute the process message function
  32. results.push(result);
  33. } catch (error) {
  34. console.error("Error processing message:", error);
  35. //Handle the error. You can log it or take other actions.
  36. }
  37. }
  39. // Close the consumer
  40. queue.close();
  42. // Write results to file
  43. const fs = require('fs'); //CommonJS module for file system operations.
  44. try {
  45. fs.writeFileSync(filename, JSON.stringify(results, null, 2)); //Write to file with pretty printing
  46. console.log(`Results exported to ${filename}`);
  47. } catch (err) {
  48. console.error("Error writing to file:", err);
  49. return Promise.reject(err);
  50. }
  52. } catch (error) {
  53. console.error("An error occurred:", error);
  54. return Promise.reject(error); //Reject the promise on any error.
  55. }
  56. }

Add your comment