Add bytes to internal bytes buffer and if echo is True, echo contents to inner stream.
(self, b: bytes)
| 499 | self.std_sim_instance = std_sim_instance |
| 500 | |
| 501 | def write(self, b: bytes) -> None: |
| 502 | """Add bytes to internal bytes buffer and if echo is True, echo contents to inner stream.""" |
| 503 | if not isinstance(b, bytes): |
| 504 | raise TypeError(f"a bytes-like object is required, not {type(b)}") |
| 505 | if not self.std_sim_instance.pause_storage: |
| 506 | self.byte_buf += b |
| 507 | if self.std_sim_instance.echo: |
| 508 | self.std_sim_instance.inner_stream.buffer.write(b) |
| 509 | |
| 510 | # Since StdSim wraps TextIO streams, we will flush the stream if line buffering is on |
| 511 | # and the bytes being written contain a new line character. This is helpful when StdSim |
| 512 | # is being used to capture output of a shell command because it causes the output to print |
| 513 | # to the screen more often than if we waited for the stream to flush its buffer. |
| 514 | if self.std_sim_instance.line_buffering and any(newline in b for newline in ByteBuf.NEWLINES): |
| 515 | self.std_sim_instance.flush() |
| 516 | |
| 517 | |
| 518 | class ProcReader: |