1. import json
  2. import logging
  4. # Configure logging
  5. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  7. def convert_message(message, input_format, output_format):
  8. """
  9. Converts a message from one queue format to another.
  11. Args:
  12. message (str): The message as a JSON string.
  13. input_format (str): The format of the input message ('raw', 'json', 'kafka').
  14. output_format (str): The desired format of the output message ('raw', 'json', 'kafka').
  16. Returns:
  17. str: The converted message as a JSON string, or None if conversion fails.
  18. """
  20. try:
  21. if input_format == 'json':
  22. data = json.loads(message)
  23. elif input_format == 'kafka':
  24. # Simulate Kafka message format (replace with actual Kafka parsing if needed)
  25. data = json.loads(message) # Assuming Kafka messages are JSON
  26. elif input_format == 'raw':
  27. data = message # treat raw input as a string
  28. else:
  29. logging.error(f"Invalid input format: {input_format}")
  30. return None
  32. if output_format == 'json':
  33. return json.dumps(data)
  34. elif output_format == 'kafka':
  35. # Simulate Kafka message format (replace with actual Kafka serialization if needed)
  36. return json.dumps(data)
  37. elif output_format == 'raw':
  38. return str(data)
  39. else:
  40. logging.error(f"Invalid output format: {output_format}")
  41. return None
  43. except json.JSONDecodeError as e:
  44. logging.error(f"JSON decoding error: {e}")
  45. return None
  46. except Exception as e:
  47. logging.error(f"Conversion error: {e}")
  48. return None
  50. if __name__ == '__main__':
  51. # Example Usage and Edge Case Handling
  53. # Test case 1: JSON to Kafka
  54. message1 = '{"name": "Alice", "age": 30}'
  55. converted_message1 = convert_message(message1, 'json', 'kafka')
  56. if converted_message1:
  57. print(f"Test 1 (JSON to Kafka): {converted_message1}")
  58. else:
  59. print("Test 1 failed.")
  61. # Test case 2: Kafka to JSON
  62. message2 = '{"product": "Laptop", "price": 1200}'
  63. converted_message2 = convert_message(message2, 'kafka', 'json')
  64. if converted_message2:
  65. print(f"Test 2 (Kafka to JSON): {converted_message2}")
  66. else:
  67. print("Test 2 failed.")
  69. # Test case 3: Raw to JSON
  70. message3 = "Name: Bob, Age: 25"
  71. converted_message3 = convert_message(message3, 'raw', 'json')
  72. if converted_message3:
  73. print(f"Test 3 (Raw to JSON): {converted_message3}")
  74. else:
  75. print("Test 3 failed.")
  77. # Test case 4: Invalid format
  78. message4 = '{"city": "New York"}'
  79. converted_message4 = convert_message(message4, 'invalid_format', 'json')
  80. if converted_message4:
  81. print(f"Test 4 (Invalid format): {converted_message4}")
  82. else:
  83. print("Test 4 failed (as expected).")
  85. #Test case 5: Invalid JSON
  86. message5 = '{"city": "New York"' #missing closing brace
  87. converted_message5 = convert_message(message5, 'json', 'json')
  88. if converted_message5:
  89. print(f"Test 5 (Invalid JSON): {converted_message5}")
  90. else:
  91. print("Test 5 failed (as expected).")

Add your comment