def release_flow_control()

in src/dubbo/remoting/aio/http2/controllers.py [0:0]

• 9 lines of code
• 6 McCabe index (conditional complexity)

    def release_flow_control(self, frame: WindowUpdateFrame) -> None:
        stream_id = frame.stream_id
        if stream_id is None or stream_id == 0:
            # This is for the entire connection.
            for i in self._flow_controls:
                self._outbound_queue.put_nowait(i)
            self._flow_controls.clear()
        elif stream_id in self._flow_controls:
            # This is specific to a single stream.
            self._flow_controls.remove(stream_id)
            self._outbound_queue.put_nowait(stream_id)