This commit is contained in:
yumoqing 2024-04-26 12:55:16 +08:00
parent 8e04b83ba3
commit d90cd47b66
2 changed files with 69 additions and 56 deletions

View File

@ -395,26 +395,35 @@ class ProcessorResource(AppLogger, StaticResource,Url2File):
return None return None
def websocketUrl(self, request, url): def websocketUrl(self, request, url):
url = entireUrl(request, url) url = entireUrl(request, url)
if url.startswith('https'): if url.startswith('https'):
return 'wss' + url[5:] return 'wss' + url[5:]
return 'ws' + url[4:] return 'ws' + url[4:]
def urlWebsocketify(self, url):
if url.endswith('.ws') or url.endswith('.wss'):
if url.startswith('https'):
return 'wss' + url[5:]
return 'ws' + url[4:]
return url
def entireUrl(self, request, url): def entireUrl(self, request, url):
ret_url = ''
if url.startswith('http://') or \ if url.startswith('http://') or \
url.startswith('https://') or \ url.startswith('https://') or \
url.startswith('ws://') or \ url.startswith('ws://') or \
url.startswith('wss://'): url.startswith('wss://'):
return url ret_url = url
if url.startswith('/'): elif url.startswith('/'):
u = f'{self._preurl}{url}' u = f'{self._preurl}{url}'
# print(f'entireUrl(), {u=}, {url=}, {self._preurl=}') # print(f'entireUrl(), {u=}, {url=}, {self._preurl=}')
return u ret_url = u
path = request.path else:
p = self.relatedurl(path,url) path = request.path
u = f'{self._preurl}{p}' p = self.relatedurl(path,url)
# print(f'entireUrl(), {u=}, {url=}, {self._preurl=}') u = f'{self._preurl}{p}'
return u ret_url = u
return self.urlWebsocketify(ret_url)
def url2path(self, url): def url2path(self, url):
if url.startswith(self._preurl): if url.startswith(self._preurl):

View File

@ -78,90 +78,94 @@ class XtermProcessor(BaseProcessor):
self.sshnode.close() self.sshnode.close()
self.p_obj.close() self.p_obj.close()
async def ws_send(ws:web.WebSocketResponse, data):
data = {
"type":1,
"data":data
}
return await ws.send_str(json.dumps(data))
class WsPool: class WsPool:
def __init__(self, ws_path, app): def __init__(self, ws, ws_path, app):
self.app = app self.app = app
self.id = None self.id = None
self.ws = ws
print(f'-----------{self.id=}-----------')
self.ws_path = ws_path self.ws_path = ws_path
self.data = self.get_data()
if self.data is None:
self.data = {}
self.set_data()
def get_data(self): def get_data(self):
self.data = self.app.get_data(self.ws_path) return self.app.get_data(self.ws_path)
def set_data(self): def set_data(self, data):
self.app.set_data(self.ws_path, self.data) self.app.set_data(self.ws_path, data)
def set_id(id): def is_online(self, userid):
data = self.get_data()
if data.get(userid):
return True
else:
return False
def register(self, id):
self.id = id self.id = id
self.add_mine()
def get_ws(self, id): def add_mine(self):
self.get_data() data = self.get_data()
return self.data.get(id) if data is None:
data = {}
data.update({self.id:self.ws})
self.set_data(data)
def add_ws(self, ws): def delete_mine(self):
if not self.id: data = self.get_data()
if not data.get(self.id):
return return
data = {k:v for k,v in self.data.items() if k != self.id }
self.set_data(data)
self.data = self.get_data() async def sendto(self, data, id=None):
if (self.data.get(self.id)): if id is None:
return return await ws_send(self.ws, data)
self.data.update({self.id:ws}) d = self.get_data()
self.set_data() ws = d.get(id)
return await ws_send(ws, data)
def delete_ws(self):
if not self.id:
return
self.data = self.get_data()
if (not self.data.get(self.id)):
return
self.data = {k:v for k,v in self.data.items() if k != self.id }
self.set_data()
class WebsocketProcessor(PythonScriptProcessor): class WebsocketProcessor(PythonScriptProcessor):
@classmethod @classmethod
def isMe(self,name): def isMe(self,name):
return name=='ws' return name=='ws'
async def ws_sendstr(self, ws:web.WebSocketResponse, s:str):
data = {
"type":1,
"data":s
}
await ws.send_str(json.dumps(data))
def received_data_convert(self, data):
return data
async def path_call(self, request,params={}): async def path_call(self, request,params={}):
await self.set_run_env(request) await self.set_run_env(request)
lenv = self.run_ns.copy() lenv = self.run_ns.copy()
lenv.update(params) lenv.update(params)
params_kw = lenv.params_kw
userid = lenv.params_kw.userid or await lenv.get_user()
print(f'{params_kw=}, {userid=} ============')
print(f'{request.path=}') print(f'{request.path=}')
del lenv['request'] del lenv['request']
txt = await self.loadScript(self.real_path) txt = await self.loadScript(self.real_path)
ws = web.WebSocketResponse() ws = web.WebSocketResponse()
await ws.prepare(request) await ws.prepare(request)
ws_pool = WsPool(request.path, request.app) ws_pool = WsPool(ws, request.path, request.app)
await self.ws_sendstr(ws, 'Welcome to websock') # await self.ws_send(ws, 'Welcome to websock')
async for msg in ws: async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT: if msg.type == aiohttp.WSMsgType.TEXT:
print('msg=:', msg.data) if msg.data == 'exit':
lenv['ws_data'] = self.received_data_convert(msg.data) break
lenv['ws_data'] = msg.data
lenv['ws_pool'] = ws_pool lenv['ws_pool'] = ws_pool
exec(txt,lenv,lenv) exec(txt,lenv,lenv)
func = lenv['myfunc'] func = lenv['myfunc']
resp = await func(request,**lenv) resp = await func(request,**lenv)
await self.ws_sendstr(ws, resp) # await self.ws_sendstr(ws, resp)
elif msg.type == aiohttp.WSMsgType.ERROR: elif msg.type == aiohttp.WSMsgType.ERROR:
# print('ws connection closed with exception %s' % ws.exception()) # print('ws connection closed with exception %s' % ws.exception())
pass break
else: else:
print('datatype error', msg.type) print('datatype error', msg.type)
# print('5----------------------------------') ws_pool.delete_mine()
self.retResponse = ws self.retResponse = ws
await ws.close() await ws.close()
return ws return ws