| 504 | |
| 505 | # Send data to connection |
| 506 | def send(self, message, streaming=False): |
| 507 | self.last_send_time = time.time() |
| 508 | if config.debug_socket: |
| 509 | self.log("Send: %s, to: %s, streaming: %s, site: %s, inner_path: %s, req_id: %s" % ( |
| 510 | message.get("cmd"), message.get("to"), streaming, |
| 511 | message.get("params", {}).get("site"), message.get("params", {}).get("inner_path"), |
| 512 | message.get("req_id")) |
| 513 | ) |
| 514 | |
| 515 | if not self.sock: |
| 516 | self.log("Send error: missing socket") |
| 517 | return False |
| 518 | |
| 519 | if not self.connected and message.get("cmd") != "handshake": |
| 520 | self.log("Wait for handshake before send request") |
| 521 | self.event_connected.get() |
| 522 | |
| 523 | try: |
| 524 | stat_key = message.get("cmd", "unknown") |
| 525 | if stat_key == "response": |
| 526 | stat_key = "response: %s" % self.last_cmd_recv |
| 527 | else: |
| 528 | self.server.num_sent += 1 |
| 529 | |
| 530 | self.server.stat_sent[stat_key]["num"] += 1 |
| 531 | if streaming: |
| 532 | with self.send_lock: |
| 533 | bytes_sent = Msgpack.stream(message, self.sock.sendall) |
| 534 | self.bytes_sent += bytes_sent |
| 535 | self.server.bytes_sent += bytes_sent |
| 536 | self.server.stat_sent[stat_key]["bytes"] += bytes_sent |
| 537 | message = None |
| 538 | else: |
| 539 | data = Msgpack.pack(message) |
| 540 | self.bytes_sent += len(data) |
| 541 | self.server.bytes_sent += len(data) |
| 542 | self.server.stat_sent[stat_key]["bytes"] += len(data) |
| 543 | message = None |
| 544 | with self.send_lock: |
| 545 | self.sock.sendall(data) |
| 546 | except Exception as err: |
| 547 | self.close("Send error: %s (cmd: %s)" % (err, stat_key)) |
| 548 | return False |
| 549 | self.last_sent_time = time.time() |
| 550 | return True |
| 551 | |
| 552 | # Stream file to connection without msgpacking |
| 553 | def sendRawfile(self, file, read_bytes): |