| 44 | print("Running") |
| 45 | |
| 46 | class AsyncQueueResult: |
| 47 | def __init__(self, worker_thread, task_queue, result_list): |
| 48 | self._worker_thread = worker_thread |
| 49 | self._task_queue = task_queue |
| 50 | self.result_list = result_list |
| 51 | self._seen_items = set() |
| 52 | self._finished = False # Flag to track if get() was called |
| 53 | |
| 54 | def get_unique(self, items): |
| 55 | single_item = False |
| 56 | if not isinstance(items, list): |
| 57 | items = [items] |
| 58 | single_item = True |
| 59 | |
| 60 | new_items = [] |
| 61 | |
| 62 | for item in items: |
| 63 | if isinstance(item, dict): |
| 64 | item_repr = frozenset(item.items()) |
| 65 | elif isinstance(item, list): |
| 66 | item_repr = tuple(item) |
| 67 | elif isinstance(item, set): |
| 68 | item_repr = frozenset(item) |
| 69 | else: |
| 70 | item_repr = item |
| 71 | |
| 72 | if item_repr not in self._seen_items: |
| 73 | new_items.append(item) |
| 74 | self._seen_items.add(item_repr) |
| 75 | |
| 76 | return new_items[0] if single_item and new_items else new_items |
| 77 | |
| 78 | def put(self, *args, **kwargs): |
| 79 | if self._finished: |
| 80 | raise RuntimeError("Cannot put items after get() was called") |
| 81 | |
| 82 | if args: |
| 83 | unique_args = self.get_unique(args[0]) |
| 84 | args_to_put = (unique_args, *args[1:]) |
| 85 | else: |
| 86 | args_to_put = () |
| 87 | |
| 88 | self._task_queue.put([args_to_put, kwargs]) |
| 89 | |
| 90 | def get(self): |
| 91 | import sys |
| 92 | self._task_queue.put(None) |
| 93 | self._finished = True # Mark as finished |
| 94 | thread = self._worker_thread |
| 95 | try: |
| 96 | # Must see https://stackoverflow.com/questions/4136632/how-to-kill-a-child-thread-with-ctrlc |
| 97 | while thread.is_alive(): |
| 98 | thread.join(0.1) |
| 99 | except KeyboardInterrupt: |
| 100 | sys.exit(1) |
| 101 | self._task_queue.join() |
| 102 | |
| 103 | return flatten(self.result_list) |
no outgoing calls
no test coverage detected