blob: 85e780a388607cb60b08c4f8499cfdb689b9b575 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
#!/usr/bin/env python
#
# Server that will accept connections from a Vim channel.
# Used by test_channel.vim.
#
# This requires Python 2.6 or later.
from __future__ import print_function
from test_channel import ThreadedTCPServer, TestingRequestHandler, \
writePortInFile
import socket
import threading
import os
try:
FileNotFoundError
except NameError:
# Python 2
FileNotFoundError = (IOError, OSError)
if not hasattr(socket, "AF_UNIX"):
raise NotImplementedError("Unix sockets are not supported on this platform")
class ThreadedUnixServer(ThreadedTCPServer):
address_family = socket.AF_UNIX
class ThreadedUnixRequestHandler(TestingRequestHandler):
pass
def main(path):
server = ThreadedUnixServer(path, ThreadedUnixRequestHandler)
# Start a thread with the server. That thread will then start a new thread
# for each connection.
server_thread = threading.Thread(target=server.serve_forever)
server_thread.start()
# Signal the test harness we're ready, the port value has no meaning.
writePortInFile(1234)
print("Listening on {0}".format(server.server_address))
# Main thread terminates, but the server continues running
# until server.shutdown() is called.
try:
while server_thread.is_alive():
server_thread.join(1)
except (KeyboardInterrupt, SystemExit):
server.shutdown()
if __name__ == "__main__":
try:
os.remove("Xtestsocket")
except FileNotFoundError:
pass
main("Xtestsocket")
|
