Test websocket IMC

5 files changed, 172 insertions, 15 deletions

diff --git a/src/py/backend_server.py b/src/py/backend_server.py
index e32bc12..7f4ce41 100644
--- a/src/py/backend_server.py
+++ b/src/py/backend_server.py
@@ -3,25 +3,42 @@
 import socket
 import json
 import datetime
+import time
 
 import tornado.iostream
 import tornado.ioloop
+import tornado.httpserver
+import tornado.websocket
 
 import netio
 import imc.nonblock
 from imc.proxy import Proxy,Connection,imc_call,imc_register_call
 
 class BackendWorker():
-    def __init__(self,center_addr):
+    def __init__(self,center_addr,ws_port):
         self.ioloop = tornado.ioloop.IOLoop.current()
         self.center_addr = center_addr
-
         self.linkclass = 'backend'
         self.linkid = None
+        self.ws_port = ws_port
+
+        self._client_linkidmap = {}
 
     def start(self):
         self._conn_center()
 
+    def add_client(self,linkid,handler):
+        self._client_linkidmap[linkid] = {}
+
+        conn = netio.WebSocketConnection(linkid,handler)
+        conn.add_close_callback(lambda conn : self.del_client(conn.linkid))
+        Proxy.instance.add_conn(conn)
+
+        return conn
+
+    def del_client(self,linkid):
+        del self._client_linkidmap[linkid]
+
     def _conn_center(self):
         def __retry():
             print('retry connect center')
@@ -46,7 +63,7 @@ class BackendWorker():
 
             netio.send_pack(stream,bytes(json.dumps({
                 'linkclass':self.linkclass,
-                'ws_addr':('210.70.137.215',81)
+                'ws_addr':('210.70.137.215',self.ws_port)
             }),'utf-8'))
             netio.recv_pack(stream,___recv_info_cb)
 
@@ -56,15 +73,45 @@ class BackendWorker():
 
     @imc.nonblock.func
     def _test_call(self,param):
+        print(time.perf_counter())
         ret = (yield imc_call(self.linkid,'/backend/' + self.center_conn.linkid,'test_dst','Hello'))
+        print(time.perf_counter())
         print(ret)
 
     @imc.nonblock.func
     def _test_dst(self,param):
-        return 'Hello Too'
-       
-if __name__ == '__main__':
-    backend_worker = BackendWorker(('localhost',5730))
+        return param + ' Too'
+
+class WebSocketConnHandler(tornado.websocket.WebSocketHandler):
+    def open(self):
+        pass
+
+    def on_message(self,msg):
+        global backend_worker
+
+        if hasattr(self,'worker_conn'):
+            self.worker_conn.recv_msg(msg)
+        
+        else:
+            info = json.loads(msg)
+            self.worker_conn = backend_worker.add_client(info['client_linkid'],self)
+
+    def on_close(self):
+        if hasattr(self,'worker_conn'):
+            self.worker_conn.close()
+
+def start_backend_worker(ws_port):
+    global backend_worker
+
+    http_serv = tornado.httpserver.HTTPServer(tornado.web.Application([
+        ('/conn',WebSocketConnHandler)
+    ]))
+    http_serv.listen(ws_port)
+
+    backend_worker = BackendWorker(('localhost',5730),ws_port)
     backend_worker.start()
 
+if __name__ == '__main__':
+    start_backend_worker(82)
+
     tornado.ioloop.IOLoop.instance().start()
diff --git a/src/py/center_server.py b/src/py/center_server.py
index 81512b5..803137f 100644
--- a/src/py/center_server.py
+++ b/src/py/center_server.py
@@ -43,7 +43,7 @@ class BackendWorker(Worker):
 
         center_serv.add_backend_worker(self)
 
-    def add_client(self):
+    def add_client(self,client_linkid):
         return self.ws_addr
 
     def close(self):
@@ -91,7 +91,11 @@ class CenterServer(tornado.tcpserver.TCPServer):
         if size == 0:
             return None
 
-        return self.backend_workerlist[random.randrange(size)].add_client()
+        linkid = self._create_linkid()
+        worker = self.backend_workerlist[random.randrange(size)]
+        ws_ip,ws_port = worker.add_client(linkid)
+
+        return (linkid,worker.linkid,ws_ip,ws_port)
 
     def _create_linkid(self):
         linkid = uuid.uuid4()
@@ -113,16 +117,22 @@ class CenterServer(tornado.tcpserver.TCPServer):
         return param + ' World'
 
 class WebConnHandler(tornado.web.RequestHandler):
-    def get(self):
+    def set_default_headers(self):
+        self.set_header('Access-Control-Allow-Origin','*')
+
+    def post(self):
         global center_serv
 
-        addr = center_serv.add_client()
-        if addr == None:
+        data = center_serv.add_client()
+        if data == None:
             self.write('Eno_backend')
         else:
-            ip,port = addr
+            client_linkid,worker_linkid,ip,port = data
             self.write(json.dumps({
-                'ip':ip,'port':port
+                'client_linkid':client_linkid,
+                'worker_linkid':worker_linkid,
+                'ip':ip,
+                'port':port
             }))
 
 if __name__ == '__main__':
diff --git a/src/py/imc/proxy.py b/src/py/imc/proxy.py
index a428ac3..125238a 100755
--- a/src/py/imc/proxy.py
+++ b/src/py/imc/proxy.py
@@ -99,7 +99,7 @@ class Proxy:
                 stat,data = self._call_pathmap[''.join([path,'/',func_name])](param)
 
             except KeyError:
-                print('Enot_exist')
+                raise
 
             if stat == True:
                 if caller_linkid == self._linkid:
diff --git a/src/py/netio.py b/src/py/netio.py
index b7454b5..e8cb30a 100644
--- a/src/py/netio.py
+++ b/src/py/netio.py
@@ -65,3 +65,19 @@ class SocketConnection(Connection):
         self._ping_timer = tornado.ioloop.PeriodicCallback(__check,1000)
         self._ping_timer.start()
         self._ping_delay = 0
+
+class WebSocketConnection(Connection):
+    def __init__(self,linkid,handler):
+        super().__init__(linkid)
+
+        self.ioloop = tornado.ioloop.IOLoop.current()
+        self.handler = handler
+
+    def send_msg(self,data):
+        self.handler.write_message(data,True)
+
+    def recv_msg(self,data):
+        self._recvloop_callback(self,data)
+
+    def start_recvloop(self,recvloop_callback):
+        self._recvloop_callback = tornado.stack_context.wrap(recvloop_callback)
diff --git a/src/test/wstest.html b/src/test/wstest.html
new file mode 100644
index 0000000..be144b4
--- /dev/null
+++ b/src/test/wstest.html
@@ -0,0 +1,84 @@
+<!DOCTYPE HTML>
+<html>
+<head>
+<script type="text/javascript" src="/jquery-1.9.1.min.js"></script>
+<script type="text/javascript">
+var count = 0;
+var last = 0;
+var data = new ArrayBuffer(1024);
+
+function test(){
+    $.post('http://toj.tfcis.org:83/conn',{},function(res){
+        var reto;
+        
+        if(res[0] != 'E'){
+            reto = JSON.parse(res)
+            conn(reto.client_linkid,reto.worker_linkid,reto.ip,reto.port);
+        }else{
+
+        }
+    });
+}
+
+function conn(client_linkid,worker_linkid,ip,port){
+    console.log(client_linkid);
+    console.log(worker_linkid);
+
+    var ws;
+    var reader = new FileReader;
+    var imc_call = function(iden,dst,func_name,param){
+        call = {
+            'type':'call',
+            'caller_genid':13,
+            'timeout':60000,
+            'iden':iden,
+            'dst':dst,
+            'func_name':func_name,
+            'param':param
+        }
+
+        data = new Blob([JSON.stringify(call)],{'type':'application/octet-stream'})
+        ws.send(data);
+    };
+    
+    ws = new WebSocket('ws://' + ip + ':' + port + '/conn');
+    ws.onopen = function(){
+        var i;
+
+        console.log('open');
+        ws.send(JSON.stringify({
+            'client_linkid':client_linkid
+        }))
+
+        imc_call(client_linkid,'/backend/' + worker_linkid,'test_dst','Hello');
+    };
+    ws.onmessage = function(e){
+        reader.onload = function(e){
+            var res = e.target.result;    
+            //console.log(JSON.parse(res));
+
+            imc_call(client_linkid,'/backend/' + worker_linkid,'test_dst','Hello');
+        };
+        reader.readAsText(e.data);
+        count++;
+    };
+
+    perf();
+}
+
+function perf(){
+    $('#speed').text((count - last) + '/s');
+    last = count;
+    setTimeout(perf,1000);
+}
+
+$(document).ready(function(){
+    test();
+});
+
+</script>
+</head>
+<body>
+<h1 id = "speed">0</h1>
+</body>
+</html>