smartinspectpython.sipipestream

Module: sipipestream.py

Revision History

Date Version Description
2023/05/30 3.0.0.0 Initial Version.

@export
class SIPipeStream(_io.BytesIO):

Stream class used to read from / write to a named pipe via Win32API calls.

This class utilizes the Windows win32file support from the pywin32 module. More information can be found here: http://timgolden.me.uk/pywin32-docs/win32file__ReadFile_meth.html

SIPipeStream(pipeHandle: int, initialSize: int = 8192)

Initializes a new instance of the class.

Arguments:
  • pipeHandle (int): Handle to the named pipe that is open for reading / writing.
  • initialSize (int): The initial size of the buffer to allocate for this stream.
Raises:
  • SIArgumentNullException: Thrown if the pipeHandle argument is null or an empty string.
def flush(self) -> None:

Overridden. Force bytes held in the buffer into the raw stream. Prior to flushing, buffered data will be written to the named pipe.

def tell(self) -> int:

Overridden. Calls the same method on the destination stream specified at initialization.

def writable(self) -> bool:

Overridden. Calls the same method on the destination stream specified at initialization.

def write(self, data) -> int:

Overridden. Writes data to the internal buffer.

Arguments:
  • data (Any): Bytes of data to write.

Buffered data will only be written to the named pipe once a "flush()" is called.

def close(self) -> None:

Overridden. Closes the stream, as well as the named pipe.

def read(self, size: int) -> bytes:

Overridden. Read data from the named pipe, and moves it to our buffer.

Arguments:
  • size (int): Number of bytes to read. This argument is ignored, as ALL available data is read from the named-pipe.
Raises:
  • IOError: Thrown if the read failed for any reason.
Returns:

A byte array of data read.

def readable(self) -> bool:

Overridden. Calls the same method on the destination stream specified at initialization.

name: str

Overridden. Calls the same method on the destination stream specified at initialization.

def detach(self) -> io.RawIOBase:

Overridden. Calls the same method on the destination stream specified at initialization.

def fileno(self) -> int:

Overridden. Calls the same method on the destination stream specified at initialization.

def isatty(self) -> bool:

Overridden. Calls the same method on the destination stream specified at initialization.

def seek(self, offset: int, whence: int = 0) -> int:

Overridden. Calls the same method on the destination stream specified at initialization.