1. import functools
  2. import inspect
  3. import importlib
  4. import sys
  5. import types
  7. def instrument_queue_code(queue_name, module_path):
  8. """
  9. Instruments message queue code for backward compatibility without async logic.
  11. Args:
  12. queue_name (str): The name of the queue to instrument (e.g., "my_queue").
  13. module_path (str): The path to the module containing the queue implementation.
  14. """
  16. try:
  17. module = importlib.import_module(module_path)
  18. except ImportError as e:
  19. print(f"Error importing module {module_path}: {e}")
  20. return
  22. queue_class = None
  23. for name, obj in inspect.getmembers(module):
  24. if name == queue_name and inspect.isclass(obj):
  25. queue_class = obj
  26. break
  28. if queue_class is None:
  29. print(f"Queue class '{queue_name}' not found in module {module_path}")
  30. return
  32. original_put = None
  33. original_get = None
  34. original_size = None
  36. # Store original methods for later restoration
  37. try:
  38. original_put = queue_class.put
  39. except AttributeError:
  40. pass
  42. try:
  43. original_get = queue_class.get
  44. except AttributeError:
  45. pass
  47. try:
  48. original_size = queue_class.size
  49. except AttributeError:
  50. pass
  53. def wrapped_put(item):
  54. """
  55. Wraps the put method to add instrumentation.
  56. """
  57. print(f"Putting item: {item} into queue: {queue_name}")
  58. if original_put:
  59. original_put(item)
  60. return
  62. def wrapped_get():
  63. """
  64. Wraps the get method to add instrumentation.
  65. """
  66. print(f"Getting item from queue: {queue_name}")
  67. if original_get:
  68. return original_get()
  69. return None
  71. def wrapped_size():
  72. """
  73. Wraps the size method to add instrumentation.
  74. """
  75. print(f"Queue {queue_name} size: {len(queue_class._queue)}")
  76. if original_size:
  77. return original_size()
  78. return len(queue_class._queue) #Default return
  80. # Replace the original methods with the wrapped versions
  81. queue_class.put = wrapped_put
  82. queue_class.get = wrapped_get
  83. queue_class.size = wrapped_size
  85. print(f"Instrumented queue '{queue_name}' in module '{module_path}'")
  87. return queue_class #Return the modified class for potential use
  89. if __name__ == '__main__':
  90. # Example usage (replace with your actual queue implementation)
  92. # Create a dummy queue class for testing
  93. class DummyQueue:
  94. def __init__(self):
  95. self._queue = []
  97. def put(self, item):
  98. self._queue.append(item)
  100. def get(self):
  101. if self._queue:
  102. return self._queue.pop(0)
  103. else:
  104. return None
  106. def size(self):
  107. return len(self._queue)
  109. # Save the dummy queue class to a file
  110. with open("dummy_queue.py", "w") as f:
  111. f.write("""
  112. class DummyQueue:
  113. def __init__(self):
  114. self._queue = []
  116. def put(self, item):
  117. self._queue.append(item)
  119. def get(self):
  120. if self._queue:
  121. return self._queue.pop(0)
  122. else:
  123. return None
  125. def size(self):
  126. return len(self._queue)
  127. """)
  129. #Instrument the dummy queue
  130. instrument_queue_code("DummyQueue", "dummy_queue")
  132. #Test the instrumented queue
  133. import dummy_queue
  134. q = dummy_queue.DummyQueue()
  135. q.put("item1")
  136. q.put("item2")
  137. print(q.get())
  138. print(q.size())

Add your comment