无法在 Flutter 中使用 WebSocket 连接到 FastAPI。 403 禁止/代码 1006

Question

所以我一直在尝试在我的 flutter 应用程序和 FastAPI 之间建立一个 websocket 连接。我认为问题出在 Flutter 上。

到目前为止，我已经尝试了 flutter 包 socket_io_client、web_socket_channel 和 websocket_manager 都没有成功。

我怀疑这可能与应用程序架构有关，也许...有点不知所措。

这是抖动错误：

I/onListen(26110): arguments: null
I/EventStreamHandler(26110):  event sink
I/onListen(26110): arguments: null
I/EventStreamHandler(26110):  event sink
W/System.err(26110): java.net.ProtocolException: Expected HTTP 101 response but was '403 Forbidden'
W/System.err(26110):    at okhttp3.internal.ws.RealWebSocket.checkUpgradeSuccess$okhttp(RealWebSocket.kt:185)
W/System.err(26110):    at okhttp3.internal.ws.RealWebSocket$connect.onResponse(RealWebSocket.kt:156)
W/System.err(26110):    at okhttp3.RealCall$AsyncCall.run(RealCall.kt:140)
W/System.err(26110):    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167)
W/System.err(26110):    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641)
W/System.err(26110):    at java.lang.Thread.run(Thread.java:923)
I/EventStreamHandler(26110): ✅ sink is not null
I/flutter (26110): websocket closed

我知道它说 403 禁止来自我的 API，虽然我知道 websocket 连接是可能的，因为我用 javascript.

测试过它

这是来自 API 的日志：

DEBUG    | websockets.protocol:__init__:244 - server - state = CONNECTING
DEBUG    | websockets.protocol:connection_made:1340 - server - event = connection_made(<_SelectorSocketTransport fd=484 read=polling write=<idle, bufsize=0>>)
DEBUG    | websockets.protocol:data_received:1412 - server - event = data_received(<422 bytes>)
DEBUG    | websockets.server:read_http_request:237 - server < GET /11 HTTP/1.1
DEBUG    | websockets.server:read_http_request:238 - server < Headers([('authorization', 'Bearer *JWTTOKEN*'), ('upgrade', 'websocket'), ('connection', 'Upgrade'), ('sec-websocket-key', 'zytoCsWVlcmsKghL5XFEdA=='), ('sec-websocket-version', '13'), ('host', '10.0.2.2:8000'), ('accept-encoding', 'gzip'), ('user-agent', 'okhttp/4.3.1')])
INFO     | uvicorn.protocols.websockets.websockets_impl:asgi_send:198 - ('127.0.0.1', 50772) - "WebSocket /11" 403
DEBUG    | websockets.server:write_http_response:256 - server > HTTP/1.1 403 Forbidden
DEBUG    | websockets.server:write_http_response:257 - server > Headers([('Date', 'Fri, 09 Apr 2021 11:10:11 GMT'), ('Server', 'Python/3.7 websockets/8.1'), ('Content-Length', '0'), ('Content-Type', 'text/plain'), ('Connection', 'close')])
DEBUG    | websockets.server:write_http_response:267 - server > body (0 bytes)
DEBUG    | websockets.protocol:fail_connection:1261 - server ! failing CONNECTING WebSocket connection with code 1006
DEBUG    | websockets.protocol:connection_lost:1354 - server - event = connection_lost(None)
DEBUG    | websockets.protocol:connection_lost:1356 - server - state = CLOSED
DEBUG    | websockets.protocol:connection_lost:1365 - server x code = 1006, reason = [no reason]

我在 Class 中有所有 WebSocket 代码，它正在 'provided'，I.E WebSocketState:

 return runApp(
    MultiProvider(
      providers: [
        Provider<AuthenticationState>(
          create: (_) => new AuthenticationState(),
        ),
        Provider<WebSocketState>(
          create: (_) => new WebSocketState(),
        ),
      ],
      child: MyApp(),
    ),
  );

WebSocketState:

class WebSocketState {
  final _socketMessage = StreamController<Message>();
  Sink<Message> get getMessageSink => _socketMessage.sink;
  Stream<Message> get getMessageStream => _socketMessage.stream;

  WebsocketManager socket;

  bool isConnected() => true;

  void connectAndListen(int userId) async {
    var token = await secureStorage.read(key: 'token');
    socket = WebsocketManager(
        'ws://10.0.2.2:8000/$userId', {'Authorization': 'Bearer $token'});

    socket.onClose((dynamic message) {
      print('websocket closed');
    });

    // Listen to server messages
    socket.onMessage((dynamic message) {
      print("Message = " + message.toString());
    });
    // Connect to server
    socket.connect();
  }

  void dispose() {
    _socketMessage.close();
    socket.close();
  }
}

connectAndListen 方法在first/main 页面在用户通过身份验证后调用，然后在其他页面正在使用websocket。

@override
  void didChangeDependencies() {
    super.didChangeDependencies();
    Provider.of<WebSocketState>(context, listen: false).connectAndListen(
        Provider.of<AuthenticationState>(context, listen: false).id);
  }

API 网络套接字 'class':

websocket_notifier.py

from enum import Enum
import json
from typing import List

class SocketClient:
    def __init__(self, user_id: int, websocket: WebSocket):
        self.user_id = user_id
        self.websocket = websocket

class WSObjects(Enum):
    Message = 0

class Notifier:
    def __init__(self):
        self.connections: List[SocketClient] = []
        self.generator = self.get_notification_generator()

    async def get_notification_generator(self):
        while True:
            message = yield
            await self._notify(message)

    async def push(self, msg: str):
        await self.generator.asend(msg)

    async def connect(self, user_id: int, websocket: WebSocket):
        await websocket.accept()
        self.connections.append(SocketClient(user_id, websocket))

    def remove(self, websocket: WebSocket):
        client: SocketClient
        for x in self.connections:
            if x.websocket == websocket:
                client = x
        self.connections.remove(client)

    async def _notify(self, message: str):
        living_connections = []
        while len(self.connections) > 0:
            # Looping like this is necessary in case a disconnection is handled
            # during await websocket.send_text(message)
            client = self.connections.pop()
            await client.websocket.send_text(message)
            living_connections.append(client)
        self.connections = living_connections
    
    async def send(self, user_id: int, info: WSObjects, json_object: dict):
        print("WS send running")
        msg = {
            "info": info,
            "data": json_object
        }
        print("connections count: " + str(len(self.connections)))

        for client in self.connections:
            if client.user_id == user_id:
                print("WS sending msg to ${client.user_id}")
                await client.websocket.send_text(json.dumps(msg))
                break


notifier = Notifier()

API 主要:

from fastapi import FastAPI 

from websocket_notifier import notifier
from starlette.websockets import WebSocket, WebSocketDisconnect

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Root"}

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(user_id: int, websocket: WebSocket):
    await notifier.connect(user_id, websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        notifier.remove(websocket)

@app.on_event("startup")
async def startup():
    # Prime the push notification generator
    await notifier.generator.asend(None)

知道我哪里做错了吗？（我实际使用的其他 flutter websocket 包与我展示的一样）

Answer 1

通过大量测试，我终于找到了一种让 websockets 与我的 flutter 应用程序和 fastapi 一起工作的方法。 https://github.com/tiangolo/fastapi/issues/129

不得不从那个问题线程中尝试一些不同的东西。但最终使用 python-socketio。我必须使用较低版本的 python-socketio 才能与最新的 flutter socket_io_client 包兼容。

Answer 2

遇到同样问题的朋友，也请查看#2639。 APIRouter 的前缀在 websocket 装饰器中不起作用。

无法在 Flutter 中使用 WebSocket 连接到 FastAPI。 403 禁止/代码 1006

Cannot connect to to FastAPI with WebSocket in Flutter. 403 forbidden / code 1006

websocket

flutter