(self, pre_ack_response)
| 60 | return self._compose_pre_ack_process_response(trigger_instance, message) |
| 61 | |
| 62 | def process(self, pre_ack_response): |
| 63 | trigger_instance, message = self._decompose_pre_ack_process_response( |
| 64 | pre_ack_response |
| 65 | ) |
| 66 | if not trigger_instance: |
| 67 | raise ValueError("No trigger_instance provided for processing.") |
| 68 | |
| 69 | get_driver().inc_counter("trigger.%s.processed" % (trigger_instance.trigger)) |
| 70 | |
| 71 | try: |
| 72 | # Use trace_context from the message and if not found create a new context |
| 73 | # and use the trigger_instance.id as trace_tag. |
| 74 | trace_context = message.get(TRACE_CONTEXT, None) |
| 75 | if not trace_context: |
| 76 | trace_context = { |
| 77 | TRACE_ID: "trigger_instance-%s" % str(trigger_instance.id) |
| 78 | } |
| 79 | # add a trace or update an existing trace with trigger_instance |
| 80 | trace_service.add_or_update_given_trace_context( |
| 81 | trace_context=trace_context, |
| 82 | trigger_instances=[ |
| 83 | trace_service.get_trace_component_for_trigger_instance( |
| 84 | trigger_instance |
| 85 | ) |
| 86 | ], |
| 87 | ) |
| 88 | |
| 89 | container_utils.update_trigger_instance_status( |
| 90 | trigger_instance, trigger_constants.TRIGGER_INSTANCE_PROCESSING |
| 91 | ) |
| 92 | |
| 93 | with CounterWithTimer(key="rule.processed"): |
| 94 | with Timer(key="trigger.%s.processed" % (trigger_instance.trigger)): |
| 95 | self.rules_engine.handle_trigger_instance(trigger_instance) |
| 96 | |
| 97 | container_utils.update_trigger_instance_status( |
| 98 | trigger_instance, trigger_constants.TRIGGER_INSTANCE_PROCESSED |
| 99 | ) |
| 100 | except: |
| 101 | # TODO : Capture the reason for failure. |
| 102 | container_utils.update_trigger_instance_status( |
| 103 | trigger_instance, trigger_constants.TRIGGER_INSTANCE_PROCESSING_FAILED |
| 104 | ) |
| 105 | # This could be a large message but at least in case of an exception |
| 106 | # we get to see more context. |
| 107 | # Beyond this point code cannot really handle the exception anyway so |
| 108 | # eating up the exception. |
| 109 | LOG.exception("Failed to handle trigger_instance %s.", trigger_instance) |
| 110 | return |
| 111 | |
| 112 | @staticmethod |
| 113 | def _compose_pre_ack_process_response(trigger_instance, message): |
nothing calls this directly
no test coverage detected