用於WebSocket通信的Jupyter

Question

我正在研究一個需要WebSockets API的應用程序，並且還將Jupyter（前IPython）筆記本作爲相對較小的功能進行集成。由於Jupyter已經使用WebSockets進行通信，因此將它作爲一個通用的庫來集成其他WebSockets API除了它自己之外還有多難？或者我最好使用另一個庫，如aiohttp？我正在尋找任何建議和暗示爲此的最佳做法。謝謝！

Answer 1

您可以將Web應用程序從主應用程序代理到Jupyter。

你用什麼技術爲WebSockets服務並不重要，代理循環將非常相似（等待消息，推送消息前進）。但是，它將會依賴於Web服務器，因爲Python對於類似WSGI的WebSockets沒有標準。

我在pyramid_notebook project做了一個。至少在編寫代碼時，將Jupyter直接嵌入到應用程序中是不可行的，因此在其自己的進程中運行Jupyter是必須的。我不確定如果最新版本改變了這一點。 Jupyter本身正在使用Tornado。

"""UWSGI websocket proxy.""" 
from urllib.parse import urlparse, urlunparse 
import logging 
import time 

import uwsgi 
from pyramid import httpexceptions 
from ws4py import WS_VERSION 
from ws4py.client import WebSocketBaseClient 


#: HTTP headers we need to proxy to upstream websocket server when the Connect: upgrade is performed 
CAPTURE_CONNECT_HEADERS = ["sec-websocket-extensions", "sec-websocket-key", "origin"] 


logger = logging.getLogger(__name__) 


class ProxyClient(WebSocketBaseClient): 
    """Proxy between upstream WebSocket server and downstream UWSGI.""" 

    @property 
    def handshake_headers(self): 
     """ 
     List of headers appropriate for the upgrade 
     handshake. 
     """ 
     headers = [ 
      ('Host', self.host), 
      ('Connection', 'Upgrade'), 
      ('Upgrade', 'WebSocket'), 
      ('Sec-WebSocket-Key', self.key.decode('utf-8')), 
      # Origin is proxyed from the downstream server, don't set it twice 
      # ('Origin', self.url), 
      ('Sec-WebSocket-Version', str(max(WS_VERSION))) 
      ] 

     if self.protocols: 
      headers.append(('Sec-WebSocket-Protocol', ','.join(self.protocols))) 

     if self.extra_headers: 
      headers.extend(self.extra_headers) 

     logger.info("Handshake headers: %s", headers) 
     return headers 

    def received_message(self, m): 
     """Push upstream messages to downstream.""" 

     # TODO: No support for binary messages 
     m = str(m) 
     logger.debug("Incoming upstream WS: %s", m) 
     uwsgi.websocket_send(m) 
     logger.debug("Send ok") 

    def handshake_ok(self): 
     """ 
     Called when the upgrade handshake has completed 
     successfully. 

     Starts the client's thread. 
     """ 
     self.run() 

    def terminate(self): 
     super(ProxyClient, self).terminate() 

    def run(self): 
     """Combine async uwsgi message loop with ws4py message loop. 

     TODO: This could do some serious optimizations and behave asynchronously correct instead of just sleep(). 
     """ 

     self.sock.setblocking(False) 
     try: 
      while not self.terminated: 
       logger.debug("Doing nothing") 
       time.sleep(0.050) 

       logger.debug("Asking for downstream msg") 
       msg = uwsgi.websocket_recv_nb() 
       if msg: 
        logger.debug("Incoming downstream WS: %s", msg) 
        self.send(msg) 

       s = self.stream 

       self.opened() 

       logger.debug("Asking for upstream msg") 
       try: 
        bytes = self.sock.recv(self.reading_buffer_size) 
        if bytes: 
         self.process(bytes) 
       except BlockingIOError: 
        pass 

     except Exception as e: 
      logger.exception(e) 
     finally: 
      logger.info("Terminating WS proxy loop") 
      self.terminate() 


def serve_websocket(request, port): 
    """Start UWSGI websocket loop and proxy.""" 
    env = request.environ 

    # Send HTTP response 101 Switch Protocol downstream 
    uwsgi.websocket_handshake(env['HTTP_SEC_WEBSOCKET_KEY'], env.get('HTTP_ORIGIN', '')) 

    # Map the websocket URL to the upstream localhost:4000x Notebook instance 
    parts = urlparse(request.url) 
    parts = parts._replace(scheme="ws", netloc="localhost:{}".format(port)) 
    url = urlunparse(parts) 

    # Proxy initial connection headers 
    headers = [(header, value) for header, value in request.headers.items() if header.lower() in CAPTURE_CONNECT_HEADERS] 

    logger.info("Connecting to upstream websockets: %s, headers: %s", url, headers) 

    ws = ProxyClient(url, headers=headers) 
    ws.connect() 

    # TODO: Will complain loudly about already send headers - how to abort? 
    return httpexceptions.HTTPOk()