(b: Union[bytes, bytearray])
| 365 | if request.streaming_callback: |
| 366 | |
| 367 | def write_function(b: Union[bytes, bytearray]) -> int: |
| 368 | assert request.streaming_callback is not None |
| 369 | self.io_loop.add_callback(request.streaming_callback, b) |
| 370 | return len(b) |
| 371 | |
| 372 | else: |
| 373 | write_function = buffer.write # type: ignore |
nothing calls this directly
no test coverage detected