Fix asynchronous console and shell contracts

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):

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

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}')

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):

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')