Check if a non-idle stream is closed
(self, stream_id: int)
| 93 | self.streams = {} |
| 94 | |
| 95 | def is_closed(self, stream_id: int) -> bool: |
| 96 | """Check if a non-idle stream is closed""" |
| 97 | stream = self.h2_conn.streams.get(stream_id, None) |
| 98 | if ( |
| 99 | stream is not None |
| 100 | and stream.state_machine.state is not h2.stream.StreamState.CLOSED |
| 101 | and self.h2_conn.state_machine.state |
| 102 | is not h2.connection.ConnectionState.CLOSED |
| 103 | ): |
| 104 | return False |
| 105 | else: |
| 106 | return True |
| 107 | |
| 108 | def is_open_for_us(self, stream_id: int) -> bool: |
| 109 | """Check if we can write to a non-idle stream.""" |
no test coverage detected