(self, port=None)
| 906 | return self.reader_thread.get_next_message(context_message, timeout=timeout) |
| 907 | |
| 908 | def start_socket(self, port=None): |
| 909 | assert not hasattr(self, "port"), "Socket already initialized." |
| 910 | from _pydev_bundle.pydev_localhost import get_socket_name |
| 911 | |
| 912 | if SHOW_WRITES_AND_READS: |
| 913 | print("start_socket") |
| 914 | |
| 915 | self._sequence = -1 |
| 916 | if port is None: |
| 917 | socket_name = get_socket_name(close=True) |
| 918 | else: |
| 919 | socket_name = (pydev_localhost.get_localhost(), port) |
| 920 | server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 921 | server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 922 | server_socket.bind(socket_name) |
| 923 | self.port = socket_name[1] |
| 924 | server_socket.listen(1) |
| 925 | if SHOW_WRITES_AND_READS: |
| 926 | print("Waiting in socket.accept()") |
| 927 | self.server_socket = server_socket |
| 928 | new_socket, addr = server_socket.accept() |
| 929 | if SHOW_WRITES_AND_READS: |
| 930 | print("Test Writer Thread Socket:", new_socket, addr) |
| 931 | |
| 932 | self._set_socket(new_socket) |
| 933 | |
| 934 | def _set_socket(self, new_socket): |
| 935 | curr_socket = getattr(self, "sock", None) |
no test coverage detected