Client¶

class lsst.ts.tcpip.Client(host: str | None, port: int | None, log: Logger, connect_callback: Callable[[BaseClientOrServer], Awaitable[None]] | None = None, monitor_connection_interval: float = 0.1, name: str = '', encoding: str = 'utf-8', terminator: bytes = b'\r\n', run_heartbeat_send_task: bool = False, heartbeat_send_sleep: float = 1.0, **kwargs: Any)¶

Bases: BaseClientOrServer

A TCP/IP socket client.

A thin wrapper around asyncio.open_connection. Like that function, it can only be used to connect once. Construct a new instance each time you wish to make a new connection.

Parameters:

hoststr | None

IP address. If blank (“”) then create a Client that is already closed.

portint | None

IP port.

loglogging.Logger

Logger.

connect_callbackcallable or None, optional

Asynchronous function to call when the connection state changes. If the other end (server) closes the connection, it may take monitor_connection_interval seconds or longer to notice. The function receives one argument: this Client.

monitor_connection_intervalfloat, optional

Interval between checking if the connection is still alive (seconds). Defaults to DEFAULT_MONITOR_CONNECTION_INTERVAL. If ≤ 0 then do not monitor the connection at all. Monitoring is only useful if you do not regularly read from the reader using the read methods of this class (or copying what they do to detect and report hangups).

namestr

Optional name used for log messages.

encodingstr

The encoding used by read_str and write_str, read_json,: and write_json.

terminatorbytes

The terminator used by read_str and write_str, read_json,: and write_json.

run_heartbeat_send_taskbool

Should the task to send heartbeats be started or not? The default is False because the OneClientReadLoopServer delegates reading data from the socket to subclasses and by default sending heartbeats would be a breaking change.

heartbeat_send_sleepfloat

The time [sec] between heartbeats. The default is 1.0 sec.

**kwargsdict [str, typing.Any]

Additional keyword arguments for asyncio.open_connection, beyond host and port.

Attributes:

hoststr | None: IP address; the host constructor argument.
portint | None: IP port; the port constructor argument.
plus…: Attributes provided by parent class BaseClientOrServer.

Notes

See tests/test_example.py for an example.

Creating an already-closed Client by specifying host=”” allows you to initialize a client attribute in a CSC or other class, in such a way that it is safe to access connected and should_be_connected (both of which will be False), and call close (which will be a no-op). You can also await start_task and done_task if you wish, but both will be done when constructed.

Always wait for start_task after constructing an instance before using the instance. This indicates the client has connected and is ready to read and write data. The sole exception is specifying host=”” to create a Client that is already closed, since you will never use those to read or write data and both start_task and done_task are already done when created.

This class provides high-level read and write methods that monitor the connection (to call connect_callback as needed) and reject any attempt to read or write if not connected. Please use them.

This class can be used as an async context manager, which may be useful for unit tests.

Attributes Summary

connected

Return True if self._reader and self._writer are connected.

Methods Summary

`basic_close`()	Close the connected client socket, if any, and set done_task done.
`call_connect_callback`()	Call self.__connect_callback.
`close`()	Close the connected client socket, if any, and set done_task done.
`read`(n)	Read up to n bytes.
`read_into`(struct)	Read binary data from a stream reader into a `ctypes.Structure`.
`read_json`()	Read JSON data.
`read_str`()	Read and decode a terminated str; strip the terminator.
`readexactly`(n)	Read exactly n bytes.
`readline`()	Read a sequence of bytes ending with `\n`.
`readuntil`([separator])	Read one line, where “line” is a sequence of bytes ending with `separator`.
`start`(**kwargs)	Connect to the TCP/IP server.
`write`(data)	Write data and call `drain`.
`write_from`(*structs)	Write binary data from one or more `ctypes.Structure`s.
`write_json`(data)	Write data in JSON format.
`write_str`(line)	Encode, terminate, and write a str.
`writelines`(lines)	Write an iterable of bytes and call `drain`.

Attributes Documentation

connected¶

Return True if self._reader and self._writer are connected.

Note: if the other end drops the connection and if you are not trying to read data (e.g. in a background loop), then it takes the operating system awhile to realize the connection is lost. So this can return true for some unknown time after the connection has been dropped.

Methods Documentation

async basic_close() → None¶

Close the connected client socket, if any, and set done_task done.

Like close except does not clear self.should_be_connected, nor cancel self._monitor_connection_task.

async call_connect_callback() → None¶

Call self.__connect_callback.

This is always safe to call. It only calls the callback function if that function is not None and if the connection state has changed since the last time this method was called.

async close() → None¶

Close the connected client socket, if any, and set done_task done.

Call connect_callback if a client was connected.

async read(n: int) → bytes¶

Read up to n bytes.

Parameters:

nint: The number of bytes to read. If -1 then block until the other end closes its writer, then return all data seen.

Raises:

ConnectionError: If the connection is lost before, or while, reading.

async read_into(struct: Structure) → None¶

Read binary data from a stream reader into a ctypes.Structure.

Parameters:

structctypes.Structure: Structure to set.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before n bytes can be read. Use the IncompleteReadError.partial attribute to get the partially read data.

async read_json() → Any¶

Read JSON data.

Read the data with read_str and return the json-decoded result.

Returns:

datatyping.Any: Data decoded from JSON.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before the complete separator is found and the internal buffer is reset.
LimitOverrunError: If the amount of data read exceeds the configured stream lmit. The data is left in the internal buffer and can be read again.
TypeError: If the data are of a type that cannot be decoded from JSON.
json.JSONDecodeError: If the data cannot be decoded from JSON.

async read_str() → str¶

Read and decode a terminated str; strip the terminator.

Read until self.terminator, strip the terminator, and decode the data as self.encoding with strict error handling.

Returns:

linestr: Line of data, as a str with the terminator stripped.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before the complete separator is found and the internal buffer is reset.
LimitOverrunError: If the amount of data read exceeds the configured stream lmit. The data is left in the internal buffer and can be read again.
UnicodeError: If decoding fails.

async readexactly(n: int) → bytes¶

Read exactly n bytes.

Parameters:

nint: The number of bytes to read.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before n bytes can be read. Use the IncompleteReadError.partial attribute to get the partially read data.

async readline() → bytes¶

Read a sequence of bytes ending with \n.

If EOF is received and \n was not found, the method returns partially read data.

Raises:

ConnectionError: If the connection is lost before, or while, reading.

async readuntil(separator: bytes = b'\n') → bytes¶

Read one line, where “line” is a sequence of bytes ending with separator.

Read data from the stream until separator is found.

On success, the data and separator will be removed from the internal buffer (consumed). Returned data will include the separator at the end.

See also read_str, which is more convenient for most use cases.

Parameters:

separatorbytes: The desired separator. The default matches the standard library, rather than using terminator.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
asyncio.IncompleteReadError: If EOF is reached before the complete separator is found and the internal buffer is reset.
LimitOverrunError: If the amount of data read exceeds the configured stream lmit. The data is left in the internal buffer and can be read again.

async start(**kwargs: Any) → None¶

Connect to the TCP/IP server.

This is called automatically by the constructor, and is not intended to be called by the user. It is a public method so that subclasses can override it.

Parameters:

**kwargsdict [str, typing.Any]: Additional keyword arguments for asyncio.open_connection, beyond host and port.

Raises:

RuntimeError: If start already called.

async write(data: bytes) → None¶

Write data and call drain.

Parameters:

databytes: The data to write.

Raises:

ConnectionError: If self.connected false before writing begins.

async write_from(*structs: Structure) → None¶

Write binary data from one or more ctypes.Structures.

Parameters:

structslist [ctypes.Structure]: Structures to write.

Raises:

ConnectionError: If self.connected false before writing begins.

async write_json(data: Any) → None¶

Write data in JSON format.

Encode the data as json and write the result with write_str.

Parameters:

dataany: The data to be written, typically a dict, but any json-encodable data is acceptable.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
UnicodeError: If encoding fails.
json.JSONEncodeError: If the data cannot be json-encoded.

async write_str(line: str) → None¶

Encode, terminate, and write a str.

Encode the str as self.encoding with strict error handling, and append self.terminator.

Parameters:

linestr: The line of data to be written.

Raises:

ConnectionError: If the connection is lost before, or while, reading.
UnicodeError: If encoding fails.

async writelines(lines: Iterable) → None¶

Write an iterable of bytes and call drain.

Parameters:

linescollections.abc.Iterable [bytes]: The data to write, as an iterable collection of bytes.

Raises:

ConnectionError: If self.connected false before writing begins.

Navigation

Client¶