diff --git a/confluent_server/confluent/asynchttp.py b/confluent_server/confluent/asynchttp.py index 25a89938..ff94029c 100644 --- a/confluent_server/confluent/asynchttp.py +++ b/confluent_server/confluent/asynchttp.py @@ -57,8 +57,8 @@ class AsyncTermRelation(object): self.asynchdl = asynchdl self.termid = termid - def got_data(self, data): - self.asynchdl.add(self.termid, data) + async def got_data(self, data): + await self.asynchdl.add(self.termid, data) class AsyncSession(object): diff --git a/confluent_server/confluent/consoleserver.py b/confluent_server/confluent/consoleserver.py index d9707a36..691afa74 100644 --- a/confluent_server/confluent/consoleserver.py +++ b/confluent_server/confluent/consoleserver.py @@ -451,8 +451,7 @@ class ConsoleHandler(object): await self._send_rcpts({'deleting': True}) await self._disconnect() if self._console: - - self._console.close() + await self._console.close() self._console = None if self.connectionthread: self.connectionthread.cancel() @@ -871,7 +870,7 @@ class ConsoleSession(object): await self.conshdl.attachsession(self) self.write = self.conshdl.write - def got_data(self, data): + async def got_data(self, data): """Receive data from console and buffer If the caller does not provide a callback and instead will be polling diff --git a/confluent_server/confluent/httpapi.py b/confluent_server/confluent/httpapi.py index 27ac3c98..c1c6ded3 100644 --- a/confluent_server/confluent/httpapi.py +++ b/confluent_server/confluent/httpapi.py @@ -585,21 +585,24 @@ async def wsock_handler(req): if asess: await asess.destroy() return rsp - if '/console/session' in ws.path or '/shell/sessions/' in ws.path: - def datacallback(data): - ws.send(websockify_data(data)) - geom = ws.wait() - geom = geom[1:] + path = req.rel_url.path + if '/console/session' in path or '/shell/sessions/' in path: + async def datacallback(data): + await rsp.send_str(websockify_data(data)) + geom = await rsp.receive() + if geom.type != WSMsgType.TEXT: + return rsp + geom = geom.data[1:] geom = json.loads(geom) width = geom['width'] height = geom['height'] skipreplay = geom.get('skipreplay', False) #hard bake JSON into this path, do not support other incarnations - if '/console/session' in ws.path: - prefix, _, _ = ws.path.partition('/console/session') + if '/console/session' in path: + prefix, _, _ = path.partition('/console/session') shellsession = False - elif '/shell/sessions/' in ws.path: - prefix, _, _ = ws.path.partition('/shell/sessions') + elif '/shell/sessions/' in path: + prefix, _, _ = path.partition('/shell/sessions') shellsession = True _, _, nodename = prefix.rpartition('/') @@ -618,11 +621,12 @@ async def wsock_handler(req): ) except exc.NotFoundException: return - clientmsg = ws.wait() + clientmsg = await rsp.receive() try: - while clientmsg is not None: + while clientmsg.type == WSMsgType.TEXT: + clientmsg = clientmsg.data if clientmsg[0] == ' ': - consession.write(clientmsg[1:]) + await consession.write(clientmsg[1:]) elif clientmsg[0] == '!': cmd = json.loads(clientmsg[1:]) action = cmd.get('action', None) @@ -632,10 +636,11 @@ async def wsock_handler(req): consession.resize( width=cmd['width'], height=cmd['height']) elif clientmsg[0] == '?': - ws.send(u'?') - clientmsg = ws.wait() + await rsp.send_str(u'?') + clientmsg = await rsp.receive() finally: - consession.destroy() + await consession.destroy() + return rsp async def resourcehandler(request): @@ -927,7 +932,7 @@ async def resourcehandler_backend(req, make_response): await rsp.write(json.dumps({'session': querydict['session']})) return rsp # client has requests to send or receive, not both... elif 'closesession' in querydict: - consolesessions[querydict['session']]['session'].destroy() + await consolesessions[querydict['session']]['session'].destroy() del consolesessions[querydict['session']] rsp = await make_response('application/json', 200) await rsp.write(b'{"sessionclosed": true}') diff --git a/confluent_server/confluent/shellmodule.py b/confluent_server/confluent/shellmodule.py index 89bfb7e7..34bc7e37 100644 --- a/confluent_server/confluent/shellmodule.py +++ b/confluent_server/confluent/shellmodule.py @@ -30,6 +30,7 @@ import fcntl import os import pty import random +import select import subprocess @@ -54,12 +55,12 @@ class ExecConsole(conapi.Console): try: somedata = os.read(self._master, 128) while somedata: - self._datacallback(somedata) + await self._datacallback(somedata) await asyncio.sleep(0) somedata = os.read(self._master, 128) except OSError as e: if e.errno == 5: - self._datacallback(conapi.ConsoleEvent.Disconnect) + await self._datacallback(conapi.ConsoleEvent.Disconnect) self.subproc = None return if e.errno != 11: @@ -68,7 +69,7 @@ class ExecConsole(conapi.Console): try: somedata = self.subproc.stderr.read() while somedata: - self._datacallback(somedata) + await self._datacallback(somedata) await asyncio.sleep(0) somedata = self.subproc.stderr.read() except IOError as e: @@ -76,10 +77,10 @@ class ExecConsole(conapi.Console): raise childstate = self.subproc.poll() if childstate is not None: - self._datacallback(conapi.ConsoleEvent.Disconnect) + await self._datacallback(conapi.ConsoleEvent.Disconnect) self.subproc = None - def connect(self, callback): + async def connect(self, callback): self._datacallback = callback master, slave = pty.openpty() self._master = master @@ -90,14 +91,16 @@ class ExecConsole(conapi.Console): stderr=subprocess.PIPE, close_fds=True) except OSError: print("Unable to execute " + self.executable + " (permissions?)") - self.close() + os.close(master) + os.close(slave) + self._master = None return os.close(slave) fcntl.fcntl(master, fcntl.F_SETFL, os.O_NONBLOCK) fcntl.fcntl(self.subproc.stderr.fileno(), fcntl.F_SETFL, os.O_NONBLOCK) self.readerthread = tasks.spawn(self.relaydata()) - def write(self, data): + async def write(self, data): os.write(self._master, data) async def close(self): diff --git a/confluent_server/confluent/shellserver.py b/confluent_server/confluent/shellserver.py index cdcdcfe2..3bb6a086 100644 --- a/confluent_server/confluent/shellserver.py +++ b/confluent_server/confluent/shellserver.py @@ -54,7 +54,7 @@ class _ShellHandler(consoleserver.ConsoleHandler): _reaper = tasks.spawn(reapsessions()) - def check_collective(self, attrvalue): + async def check_collective(self, attrvalue): return def log(self, *args, **kwargs): @@ -69,7 +69,7 @@ class _ShellHandler(consoleserver.ConsoleHandler): # #retdata, connstate = await super(_ShellHandler, self).get_recent() # return '', {} # connstate - def _got_disconnected(self): + async def _got_disconnected(self): self.connectstate = 'closed' tasks.spawn(self._bgdisconnect()) @@ -77,7 +77,7 @@ class _ShellHandler(consoleserver.ConsoleHandler): await self._send_rcpts({'connectstate': self.connectstate}) for session in list(self.livesessions): await session.destroy() - self.feedbuffer('\x1bc') + await self.feedbuffer('\x1bc')