| 142 | elif envelope.kind == pb.Envelope.KIND_RESPONSE: |
| 143 | self._pending.pop(envelope.id, None) |
| 144 | if not pending.future.done(): |
| 145 | pending.future.set_result(envelope) |
| 146 | return |
| 147 | if envelope.kind == pb.Envelope.KIND_EVENT: |
| 148 | try: |
| 149 | pending.queue.put_nowait(envelope) |
| 150 | except asyncio.QueueFull: |
| 151 | self._pending.pop(envelope.id, None) |
| 152 | _fail_stream( |
| 153 | pending.queue, |
| 154 | CmdopConnectionError( |
| 155 | "agent stream consumer is too slow", code="transport" |
| 156 | ), |
| 157 | ) |
| 158 | elif envelope.kind == pb.Envelope.KIND_DONE: |
| 159 | self._pending.pop(envelope.id, None) |
| 160 | if pending.queue.qsize() > pending.queue.maxsize - 2: |
| 161 | _fail_stream( |