(self, request)
| 160 | return request.ip |
| 161 | |
| 162 | def on_open(self, request): |
| 163 | s = settings() |
| 164 | loggedUsername = s.get(["cloudSlicer", "loggedUser"]) |
| 165 | |
| 166 | if loggedUsername: |
| 167 | token = request.arguments.get("token") |
| 168 | token = token[0] if token else None |
| 169 | tokenContents = octoprint.server.read_ws_token(token) |
| 170 | if not tokenContents or tokenContents['public_key'] != self._userManager.findUser(loggedUsername).publicKey: |
| 171 | return False |
| 172 | |
| 173 | remoteAddress = self._getRemoteAddress(request) |
| 174 | self._logger.info("New connection from client [IP address: %s, Session id: %s]", remoteAddress, self.session.session_id) |
| 175 | |
| 176 | # connected => update the API key, might be necessary if the client was left open while the server restarted |
| 177 | self._emit("connected", {"apikey": octoprint.server.UI_API_KEY, "version": octoprint.server.VERSION, "sessionId": self.session.session_id}) |
| 178 | self.sendEvent(Events.ASTROPRINT_STATUS, boxrouterManager().status) |
| 179 | |
| 180 | printer = printerManager() |
| 181 | |
| 182 | printer.registerCallback(self) |
| 183 | printer.fileManager.registerCallback(self) |
| 184 | |
| 185 | self._eventManager.fire(Events.CLIENT_OPENED, {"remoteAddress": remoteAddress}) |
| 186 | for event in PrinterStateConnection.EVENTS: |
| 187 | self._eventManager.subscribe(event, self._onEvent) |
| 188 | |
| 189 | if printer.promptManager.hasPrompt: |
| 190 | self.sendEvent(Events.PRINTER_PROMPT, { |
| 191 | 'type': 'show', |
| 192 | 'prompt': printer.promptManager.prompt |
| 193 | }) |
| 194 | |
| 195 | def on_close(self): |
| 196 | self._logger.info("Client connection closed [Session id: %s]", self.session.session_id) |
nothing calls this directly
no test coverage detected