(self, debug=False, **kwargs)
| 122 | SERVER_CLASS = None |
| 123 | |
| 124 | def __init__(self, debug=False, **kwargs): |
| 125 | self.name = f"EngineClient {self.__class__.__name__}" |
| 126 | super().__init__(debug=debug) |
| 127 | self.process = None |
| 128 | if self.SERVER_CLASS is None: |
| 129 | raise ValueError(f"Must set EngineClient SERVER_CLASS, {self.SERVER_CLASS}") |
| 130 | self.CMDS = dict(self.SERVER_CLASS.CMDS) |
| 131 | for k, v in list(self.CMDS.items()): |
| 132 | self.CMDS[v] = k |
| 133 | self.socket_address = f"zmq_{rand_string(8)}.sock" |
| 134 | self.socket_path = Path(tempfile.gettempdir()) / self.socket_address |
| 135 | self.server_kwargs = kwargs.pop("server_kwargs", {}) |
| 136 | self._server_process = None |
| 137 | self.context = zmq.asyncio.Context() |
| 138 | self.context.setsockopt(zmq.LINGER, 0) |
| 139 | self.sockets = set() |
| 140 | |
| 141 | def check_error(self, message): |
| 142 | if isinstance(message, dict) and len(message) == 1 and "_e" in message: |
nothing calls this directly
no test coverage detected