(self)
| 207 | |
| 208 | # Message loop for connection |
| 209 | def messageLoop(self): |
| 210 | if not self.sock: |
| 211 | self.log("Socket error: No socket found") |
| 212 | return False |
| 213 | self.protocol = "v2" |
| 214 | self.updateName() |
| 215 | self.connected = True |
| 216 | buff_len = 0 |
| 217 | req_len = 0 |
| 218 | self.unpacker_bytes = 0 |
| 219 | |
| 220 | try: |
| 221 | while not self.closed: |
| 222 | buff = self.sock.recv(64 * 1024) |
| 223 | if not buff: |
| 224 | break # Connection closed |
| 225 | buff_len = len(buff) |
| 226 | |
| 227 | # Statistics |
| 228 | self.last_recv_time = time.time() |
| 229 | self.incomplete_buff_recv += 1 |
| 230 | self.bytes_recv += buff_len |
| 231 | self.server.bytes_recv += buff_len |
| 232 | req_len += buff_len |
| 233 | |
| 234 | if not self.unpacker: |
| 235 | self.unpacker = self.getMsgpackUnpacker() |
| 236 | self.unpacker_bytes = 0 |
| 237 | |
| 238 | self.unpacker.feed(buff) |
| 239 | self.unpacker_bytes += buff_len |
| 240 | |
| 241 | while True: |
| 242 | try: |
| 243 | message = next(self.unpacker) |
| 244 | except StopIteration: |
| 245 | break |
| 246 | if not type(message) is dict: |
| 247 | if config.debug_socket: |
| 248 | self.log("Invalid message type: %s, content: %r, buffer: %r" % (type(message), message, buff[0:16])) |
| 249 | raise Exception("Invalid message type: %s" % type(message)) |
| 250 | |
| 251 | # Stats |
| 252 | self.incomplete_buff_recv = 0 |
| 253 | stat_key = message.get("cmd", "unknown") |
| 254 | if stat_key == "response" and "to" in message: |
| 255 | cmd_sent = self.waiting_requests.get(message["to"], {"cmd": "unknown"})["cmd"] |
| 256 | stat_key = "response: %s" % cmd_sent |
| 257 | if stat_key == "update": |
| 258 | stat_key = "update: %s" % message["params"]["site"] |
| 259 | self.server.stat_recv[stat_key]["bytes"] += req_len |
| 260 | self.server.stat_recv[stat_key]["num"] += 1 |
| 261 | if "stream_bytes" in message: |
| 262 | self.server.stat_recv[stat_key]["bytes"] += message["stream_bytes"] |
| 263 | req_len = 0 |
| 264 | |
| 265 | # Handle message |
| 266 | if "stream_bytes" in message: |
no test coverage detected