使用Flask和WebSocket-for-Python(ws4py)的IPython Notebook的WebSockets代理

Question

灵感来自ipython-notebook-proxy,基于ipydra,并扩展后者以支持更复杂的用户身份验证以及代理,因为在我的用例中,只有端口80可以暴露.

我正在为工作者使用flask-socketgunicorn,但我遇到代理WebSockets的麻烦.IPython的使用三种不同的WebSockets连接/shell,/stdin以及/iopub,但我只能够拿到101 Switching Protocols的第2位.并在创建后尽快/stdin收到Connection Close Frame.

这是有问题的摘录代码:

# Flask imports...
from werkzeug import LocalProxy
from ws4py.client.geventclient import WebSocketClient

# I use my own LocalProxy because flask-sockets does not support Werkzeug Rules
websocket = LocalProxy(lambda: request.environ.get('wsgi.websocket', None))
websockets = {}

PROXY_DOMAIN = "127.0.0.1:8888"  # IPython host and port
methods = ["GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS", "PATCH",
           "CONNECT"]


@app.route('/', defaults={'url': ''}, methods=methods)
@app.route('/<path:url>', methods=methods)
def proxy(url):
    with app.test_request_context():
        if websocket:
            while True:
                data = websocket.receive()
                websocket_url = 'ws://{}/{}'.format(PROXY_DOMAIN, url)
                if websocket_url not in websockets:
                    client = WebSocketClient(websocket_url,
                                             protocols=['http-only', 'chat'])
                    websockets[websocket_url] = client
                else:
                    client = websockets[websocket_url]
                client.connect()
                if data:
                    client.send(data)
                client_data = client.receive()
                if client_data:
                    websocket.send(client_data)
            return Response()

我也尝试创建自己的WebSocket代理类,但它也不起作用.

class WebSocketProxy(WebSocketClient):
    def __init__(self, to, *args, **kwargs):
        self.to = to
        print(("Proxy to", self.to))
        super(WebSocketProxy, self).__init__(*args, **kwargs)

    def opened(self):
        m = self.to.receive()
        print("<= %d %s" % (len(m), str(m)))
        self.send(m)

    def closed(self, code, reason):
        print(("Closed down", code, reason))

    def received_message(self, m):
        print("=> %d %s" % (len(m), str(m)))
        self.to.send(m)

常规请求 - 响应周期就像魅力一样,所以我删除了该代码.如果有兴趣,完整的代码托管在hidra中.

我运行服务器

$ gunicorn -k flask_sockets.worker hidra:app

Answer 1

这是我的解决方案（左右）。它很粗糙，但应该作为构建 websocket 代理的起点。完整的代码可在未发布的项目中找到，pyramid_notebook。

• 这使用 ws4py 和 uWSGI 而不是gunicorn

• 我们使用uWSGI的内部机制来接收下游websocket消息循环。Python 世界中还没有像 WSGI 这样的用于 websockets 的东西（还没有？），但看起来每个 Web 服务器都实现了自己的机制。

• 创建一个自定义的 ws4py ProxyConnection ，它可以将 ws4py 事件循环与 uWSGI 事件循环结合起来

• 事情开始了，消息开始四处传播

• 这使用 Pyramid request（基于 WebOb），但这确实不重要，并且代码应该适合任何 Python WSGI 应用程序，只需进行少量修改

• 正如您所看到的，这并没有真正利用异步性，而只是在套接字没有任何内容传入时使用 sleep()

代码在这里：

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

import uwsgi
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):
        raise RuntimeError("NO!")
        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()

    # Happens only if exceptions fly around
    return ""