8 from processor import Session, Dispatcher
9 from utils import print_log
12 class TcpSession(Session):
14 def __init__(self, connection, address, use_ssl, ssl_certfile, ssl_keyfile):
15 Session.__init__(self)
16 self.use_ssl = use_ssl
19 self._connection = ssl.wrap_socket(
22 certfile=ssl_certfile,
24 ssl_version=ssl.PROTOCOL_SSLv23,
25 do_handshake_on_connect=False)
27 self._connection = connection
29 self.address = address[0]
30 self.name = "TCP " if not use_ssl else "SSL "
31 self.response_queue = queue.Queue()
33 def do_handshake(self):
35 self._connection.do_handshake()
39 raise Exception("Session was stopped")
41 return self._connection
48 self._connection.shutdown(socket.SHUT_RDWR)
50 # print_log("problem shutting down", self.address)
51 # traceback.print_exc(file=sys.stdout)
54 self._connection.close()
58 def send_response(self, response):
59 self.response_queue.put(response)
62 class TcpClientResponder(threading.Thread):
64 def __init__(self, session):
65 self.session = session
66 threading.Thread.__init__(self)
69 while not self.session.stopped():
70 response = self.session.response_queue.get()
71 data = json.dumps(response) + "\n"
74 l = self.session.connection().send(data)
81 class TcpClientRequestor(threading.Thread):
83 def __init__(self, dispatcher, session):
84 self.shared = dispatcher.shared
85 self.dispatcher = dispatcher
87 self.session = session
88 threading.Thread.__init__(self)
92 self.session.do_handshake()
96 while not self.shared.stopped():
100 self.session.time = time.time()
106 data = self.receive()
117 return self.session.connection().recv(2048)
122 raw_buffer = self.message.find('\n')
126 raw_command = self.message[0:raw_buffer].strip()
127 self.message = self.message[raw_buffer + 1:]
128 if raw_command == 'quit':
133 command = json.loads(raw_command)
135 self.dispatcher.push_response({"error": "bad JSON", "request": raw_command})
139 # Try to load vital fields, and return an error if
141 message_id = command['id']
142 method = command['method']
144 # Return an error JSON in response.
145 self.dispatcher.push_response({"error": "syntax error", "request": raw_command})
147 self.dispatcher.push_request(self.session, command)
152 class TcpServer(threading.Thread):
154 def __init__(self, dispatcher, host, port, use_ssl, ssl_certfile, ssl_keyfile):
155 self.shared = dispatcher.shared
156 self.dispatcher = dispatcher.request_dispatcher
157 threading.Thread.__init__(self)
161 self.lock = threading.Lock()
162 self.use_ssl = use_ssl
163 self.ssl_keyfile = ssl_keyfile
164 self.ssl_certfile = ssl_certfile
167 print_log( ("SSL" if self.use_ssl else "TCP") + " server started on port %d"%self.port)
168 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
169 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
170 sock.bind((self.host, self.port))
173 while not self.shared.stopped():
175 #if self.use_ssl: print_log("SSL: socket listening")
177 connection, address = sock.accept()
179 traceback.print_exc(file=sys.stdout)
183 #if self.use_ssl: print_log("SSL: new session", address)
185 session = TcpSession(connection, address, use_ssl=self.use_ssl, ssl_certfile=self.ssl_certfile, ssl_keyfile=self.ssl_keyfile)
186 except BaseException, e:
188 print_log("cannot start TCP session", error, address)
193 self.dispatcher.add_session(session)
194 self.dispatcher.collect_garbage()
195 client_req = TcpClientRequestor(self.dispatcher, session)
197 responder = TcpClientResponder(session)