1
# ❌ 错误方案
2
async def get_device_version(id):
3
    await device.websocket.send_json(CMD_VERSION)
4
    response = await device.websocket.receive_json()  # 阻塞等待
5
    return response

1
# ❌ 低效方案
2
device.latest_version = None
3
await device.websocket.send_json(CMD_VERSION)
4
for _ in range(50):
5
    if device.latest_version is not None:
6
        return device.latest_version
7
    await asyncio.sleep(0.1)
8
return "timeout"

1
# ⚠️ 尝试用回调实现
2
class Esp32:
3
    def __init__(self):
4
        self.callbacks = {}  # key -> callback function
5

6
    def register_callback(self, key: str, callback):
7
        """注册：等收到这个 key 的响应时，调用 callback"""
8
        self.callbacks[key] = callback
9

10
# 消息循环中触发回调
11
async def _on_device_text(device, raw):
12
    json_data = json.loads(raw)
13
    key = json_data.get("key")
14
    if key in device.callbacks:
15
        device.callbacks.pop(key)(json_data)  # 调用回调
16
        return
17
    # ...正常转发...

1
# ❌ 回调的死胡同
2
@router.get("/{id}/version")
3
async def get_device_version(id):
4
    result = None
5

6
    def on_version(data):
7
        nonlocal result
8
        result = data   # 回调把结果写入 result
9

10
    device.register_callback("version", on_version)
11
    await device.websocket.send_json(CMD_VERSION)
12

13
    # 但是...现在 result 还是 None！
14
    # 回调还没被调用（设备还没响应）
15
    # 而 HTTP 函数必须现在就 return 一个值
16
    return CommonOut(data=result)  # ← None，错误！

1
时间线：
2
  T0  HTTP handler: 注册回调, 发送指令
3
  T1  HTTP handler: 函数执行到末尾, 必须 return  ← 此时设备还没响应！
4
  T2  WS 消息循环: 收到设备响应, 触发回调
5
  T3  回调执行: result = data  ← 但 HTTP 响应早就在 T1 返回了，来不及了

1
# ✅ 回调 + Future = 最终方案
2
async def get_device_version(id):
3
    future = loop.create_future()                       # 创建占位符
4
    device.register_callback("version", future.set_result)  # 回调 = 填充 Future
5
    await device.websocket.send_json(CMD_VERSION)
6
    result = await future   # 挂起，直到回调执行 set_result()
7
    return CommonOut(data=result)

1
协程A (HTTP handler)          共享的 Future           协程B (WS 消息循环)
2
    │                              │                       │
3
    ├─ create future ────────────► │                       │
4
    ├─ send command to device      │                       │
5
    ├─ await future (挂起) ◄───── pending                  │
6
    │                              │     收到设备响应 ────── │
7
    │                              │ ◄── set_result(data)   │
8
    ├─ await 返回, 拿到 data ◄── filled                    │
9
    └─ return response             │                       │

1
# 最朴素的回调模式
2
def 点餐(菜名, 做好了之后):
3
    厨房.开始做(菜名)
4
    厨房.做好时调用(做好了之后)  # 注册回调
5

6
def 吃饭(菜):
7
    print(f"开吃: {菜}")
8

9
点餐("红烧肉", 吃饭)   # 传入回调函数
10
# 这里程序继续往下跑，不会等在这里

1
# ========= 纯回调实现（在 FastAPI 中行不通的版本）=========
2

3
# 1. 设备对象维护回调注册表
4
class Esp32:
5
    def __init__(self):
6
        self.callbacks: dict[str, callable] = {}
7

8
    def on_response(self, key, callback):
9
        self.callbacks[key] = callback
10

11
# 2. 消息循环：收到消息后查找并触发回调
12
async def _on_device_text(device, raw):
13
    json_data = json.loads(raw)
14
    key = json_data.get("key")
15
    cb = device.callbacks.pop(key, None)
16
    if cb:
17
        cb(json_data)  # 触发回调！
18
        return
19
    # ...正常转发...
20

21
# 3. HTTP 接口中使用
22
@router.get("/{id}/version")
23
async def get_device_version(id):
24
    device = esp32IdDict[id]
25

26
    # 问题在这里 ↓
27
    received = {}
28
    def my_callback(data):
29
        received["result"] = data   # 回调把数据存到外部字典
30

31
    device.on_response("version", my_callback)
32
    await device.websocket.send_json(CMD_VERSION)
33

34
    # ❌ 此刻 received 是空的！因为设备还没响应
35
    # ❌ 但 HTTP 函数必须返回值
36
    return received.get("result")  # → None

1
HTTP handler 执行流         回调函数           WS 消息循环
2
     │                        │                    │
3
  T0 ├── on_response(cb) ────►│                    │
4
  T1 ├── send_json(CMD) ──────┼───────────────────►│
5
  T2 ├── return received ──►  │                    │  ← HTTP 响应已发出（None）!
6
     │   （函数结束了）          │                    │
7
  T3 │                        │    receive(data) ──┤
8
  T4 │  （已经不存在了）   ◄── cb(data)             │  ← 回调终于执行了
9
     │                        │                    │     但 HTTP 响应已经返回了

1
// Node.js Express — 回调是可以的
2
app.get('/version', (req, res) => {
3
    device.send({ key: 'version' });
4
    device.once('message:version', (data) => {
5
        res.json(data);   // ← 在回调里发送 HTTP 响应
6
    });
7
    // 注意：这里没有 return 响应内容
8
    // Express 不要求函数返回值，它靠 res.json() 来发响应
9
});

1
# FastAPI：响应 = 函数返回值
2
@router.get("/version")
3
async def handler():
4
    return CommonOut(data=...)  # ← 只能在这里决定返回什么
5

6
# Express：响应 = 手动调用 res
7
app.get('/version', (req, res) => {
8
    // 可以在任何时候、任何回调里调用 res.json()
9
    setTimeout(() => res.json({...}), 1000);
10
});

1
# 回调                              Future
2
# ────                              ──────
3

4
def on_version(data):               future = loop.create_future()
5
    # data 在这里，但没法
6
    # 传回 HTTP handler              pending["version"] = future
7
    ???                              await ws.send_json(CMD_VERSION)
8

9
                                     data = await future  # 暂停在这里
10
                                     # ↑ 等价于：注册了一个回调说
11
                                     # "有人 set_result 时唤醒我"
12

13
# 消息循环触发回调：                    # 消息循环填充 Future：
14
cb(json_data)                        future.set_result(json_data)
15
# ↓                                  # ↓
16
# 回调执行了，但 HTTP handler          # await 被唤醒，data 有值了
17
# 已经 return 了                      # HTTP handler 继续执行到 return

1
回调:    "嘿，做完了调用这个函数"    → 通知能力 ✓，挂起等待 ✗
2
Future:  "嘿，做完了往这个盒子里放结果，我在盒子前 await 着"  → 通知能力 ✓，挂起等待 ✓

1
while True:
2
    data = await websocket.receive()  # 独占读取
3
    if "text" in data:
4
        await _on_device_text(device, data["text"])
5
    if "bytes" in data:
6
        await _on_device_bytes(device, websocket, data["bytes"])

1
import uuid
2

3
async def send_and_wait(self, command, key, timeout=5.0):
4
    request_id = f"{key}:{uuid.uuid4().hex[:8]}"
5
    command["_req_id"] = request_id  # 需要设备原样返回此字段
6
    future = loop.create_future()
7
    self.pending_responses[request_id] = future
8
    # ...