Check if we can write to a non-idle stream.
(self, stream_id: int)
| 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.""" |
| 110 | stream = self.h2_conn.streams.get(stream_id, None) |
| 111 | if ( |
| 112 | stream is not None |
| 113 | and stream.state_machine.state |
| 114 | is not h2.stream.StreamState.HALF_CLOSED_LOCAL |
| 115 | and stream.state_machine.state is not h2.stream.StreamState.CLOSED |
| 116 | and self.h2_conn.state_machine.state |
| 117 | is not h2.connection.ConnectionState.CLOSED |
| 118 | ): |
| 119 | return True |
| 120 | else: |
| 121 | return False |
| 122 | |
| 123 | def _handle_event(self, event: Event) -> CommandGenerator[None]: |
| 124 | if isinstance(event, Start): |
no test coverage detected