(self, event: events.Event)
| 129 | yield from () # pragma: no cover |
| 130 | |
| 131 | def handle_event(self, event: events.Event) -> CommandGenerator[None]: |
| 132 | if self._paused: |
| 133 | # did we just receive the reply we were waiting for? |
| 134 | pause_finished = ( |
| 135 | isinstance(event, events.CommandCompleted) |
| 136 | and event.command is self._paused.command |
| 137 | ) |
| 138 | if self.debug is not None: |
| 139 | yield self.__debug(f"{'>>' if pause_finished else '>!'} {event}") |
| 140 | if pause_finished: |
| 141 | assert isinstance(event, events.CommandCompleted) |
| 142 | yield from self.__continue(event) |
| 143 | else: |
| 144 | self._paused_event_queue.append(event) |
| 145 | else: |
| 146 | if self.debug is not None: |
| 147 | yield self.__debug(f">> {event}") |
| 148 | command_generator = self._handle_event(event) |
| 149 | send = None |
| 150 | |
| 151 | # inlined copy of __process to reduce call stack. |
| 152 | # <✂✂✂> |
| 153 | try: |
| 154 | # Run ._handle_event to the next yield statement. |
| 155 | # If you are not familiar with generators and their .send() method, |
| 156 | # https://stackoverflow.com/a/12638313/934719 has a good explanation. |
| 157 | command = command_generator.send(send) |
| 158 | except StopIteration: |
| 159 | return |
| 160 | |
| 161 | while True: |
| 162 | if self.debug is not None: |
| 163 | if not isinstance(command, commands.Log): |
| 164 | yield self.__debug(f"<< {command}") |
| 165 | if command.blocking is True: |
| 166 | # We only want this layer to block, the outer layers should not block. |
| 167 | # For example, take an HTTP/2 connection: If we intercept one particular request, |
| 168 | # we don't want all other requests in the connection to be blocked a well. |
| 169 | # We signal to outer layers that this command is already handled by assigning our layer to |
| 170 | # `.blocking` here (upper layers explicitly check for `is True`). |
| 171 | command.blocking = self |
| 172 | self._paused = Paused( |
| 173 | command, |
| 174 | command_generator, |
| 175 | ) |
| 176 | yield command |
| 177 | return |
| 178 | else: |
| 179 | yield command |
| 180 | try: |
| 181 | command = next(command_generator) |
| 182 | except StopIteration: |
| 183 | return |
| 184 | # </✂✂✂> |
| 185 | |
| 186 | def __process(self, command_generator: CommandGenerator, send=None): |
| 187 | """ |
no test coverage detected