• 嵌入式
• 物联网
• 硬件开发
• 驱动开发

本文围绕 FastLED 仓库中的 docs/remote_stream_investigation.md 调查文档展开，剖析fl::RemoteJSON-RPC 远程控制在 Serial / Network / HTTP Streaming 等流式场景下 I/O 处理的痛点，对比三种候选设计方案，并给出推荐方案（Option B：分离 pull/push + 抽象 RpcStream）的完整 API 与实现细节。同时结合仓库中 Server/Remote 的实际落地实现、Remote.ino 示例 与单元测试，印证设计目标如何被回调式 Server 的pull()/push()/update()分相架构所实现。读完本文，你将理解 FastLED 远程控制的 I/O 协调层应如何设计，以及如何在保持向后兼容的前提下为使用者提供"一行 update()"与"显式 pull/tick/push"两档 API。

一、调查背景：fl::Remote远程控制面临的三个核心问题

FastLED 的fl::Remote是库内的 JSON-RPC 服务端组件（定义于 src/fl/remote/remote.h），允许主机通过 Serial、网络等流式通道向固件发送 JSON 命令，实现远程操控 LED。它支持即时执行与带timestamp字段的定时调度执行两种模式，内部由三部分组成（见 remote.h 注释中的架构说明）：

• Server：JSON-RPC 的 I/O 协调层（pull/push）；
• Rpc：方法注册表与 JSON-RPC 执行器；
• RpcScheduler：基于时间的任务调度器；
• Remote：组合以上三者的门面协调器。

调查文档指出，在这一架构成型之前，使用者面临的 I/O 处理方式是高度手工化的，具体体现为三个问题：

1. 手工 I/O 样板代码泛滥

用户必须手动完成全部七个步骤：

1. 从流（Serial、Network 等）读取数据；
2. 解析 JSON；
3. 调用processRpc()；
4. 手工构建响应 JSON；
5. 把响应写回流；
6. 单独调用tick()处理调度任务；
7. 调用getResults()并手工打印结果。

2. 没有统一的流处理层

调查文档给出了当时每个示例都在重复实现的模式（约 40 行以上的手工解析/响应代码）：

if (Serial.available()) { fl::string jsonRpc = readSerialJson(); // 40+ lines of manual parsing/response handling } remote.tick(millis()); auto results = remote.getResults(); for (const auto& r : results) { fl::Remote::printJson(r.to_json()); }

3. 输出行为不一致

"即时响应"（同步调用直接返回）与"调度结果"（定时任务执行完后再回传）走了两条不同的输出路径，用户需要自己拼接两种结果，容易遗漏或重复。

调查时的典型用法模式

文档摘录了当时Remote.ino示例（113~183 行）的loop()结构，完整呈现了上述样板：

void loop() { remote.tick(millis()); // 1. Process scheduled calls auto results = remote.getResults(); // 2. Get scheduled results for (const auto& r : results) { fl::Remote::printJson(r.to_json()); // 3. Print scheduled results } if (Serial.available()) { // 4. Read input fl::string jsonRpc = readSerialJson(); fl::json doc = fl::json::parse(jsonRpc); auto response = remote.processRpc(fl::Remote::RpcRequest{...}); fl::json response = fl::json::object(); // 5. Build response if (response.ok()) { response.set("status", "ok"); // ... 40 lines of response building } fl::Remote::printJson(response); // 6. Print immediate response } }

这段代码中，调度、输入读取、解析、响应构建、输出被混在一个loop()里，既难以测试，也难以复用到 Serial 之外的传输上。

二、三种候选设计方案的对比分析

调查文档系统性地提出了三种候选方案，并逐一列出优缺点。

Option A：单一update()方法

// Setup remote.setStream(&Serial); // Loop remote.update(millis()); // Does everything: tick + pull + push

Option B：分离 pull/push 方法（⭐ 推荐）

// Setup (optional - can also use manual processRpc) remote.attachStream(&Serial); // Loop remote.pull(); // Read available input, process RPCs, queue responses remote.tick(millis()); // Process scheduled calls remote.push(); // Write queued responses and scheduled results

Option C：回调式

remote.onRequest([](Stream& stream) { if (stream.available()) { return stream.readStringUntil('\n'); } return fl::string(); }); remote.onResponse([](const fl::json& response) { Serial.println(response.to_string()); });

一个值得注意的印证

将 Option C 的回调式接口与仓库当前的落地实现对照可以发现：仓库最终在 Server 基类 中选择了"回调式"作为传输抽象（RequestSource/ResponseSink），但在阶段划分上完全采纳了 Option B 的 pull/push/update 分相架构。也就是说，调查文档推荐的"分相"原则落地了，而流的接入方式由抽象RpcStream收敛为两个可注入的fl::function回调。这一点在"仓库实际落地"一节会详细展开。

三、推荐方案详解：Option B + 抽象 RpcStream

调查文档的核心产出，是一套以 Option B 为基础的完整设计：流抽象 + Remote 类 API 扩展 + 三段式实现。

3.1 流抽象：RpcStream与ArduinoStreamAdapter

为了让Remote不依赖具体传输（Serial、Network、HTTP Streaming……），设计引入一个与 ArduinoStream语义兼容的纯虚接口：

// Abstract stream interface (compatible with Arduino Stream) class RpcStream { public: virtual ~RpcStream() = default; virtual int available() = 0; virtual int read() = 0; virtual size_t write(const uint8_t* buffer, size_t size) = 0; virtual size_t print(const char* str) = 0; virtual size_t println(const char* str) = 0; };

配套提供一个把任意 ArduinoStream子类（如Serial、SoftwareSerial、网络流）包装成RpcStream的适配器模板：

// Adapter for Arduino Serial/Stream template<typename TStream> class ArduinoStreamAdapter : public RpcStream { TStream* mStream; public: ArduinoStreamAdapter(TStream* stream) : mStream(stream) {} int available() override { return mStream->available(); } int read() override { return mStream->read(); } size_t write(const uint8_t* buffer, size_t size) override { return mStream->write(buffer, size); } size_t print(const char* str) override { return mStream->print(str); } size_t println(const char* str) override { return mStream->println(str); } };

该设计的核心价值：Remote只面向 5 个方法的抽象接口编程，Serial 与网络流可以互换；测试时也能注入 mock 流。

3.2Remote类 API 扩展

在现有Remote之上新增三个流管理方法与三个 I/O 方法，并保持手动processRpc()的兼容：

class Remote { public: // Stream attachment (optional - maintains backward compatibility) void attachStream(RpcStream* stream); void detachStream(); bool hasStream() const; // Stream-based I/O (only work if stream attached) size_t pull(); // Read available input, process RPCs, queue responses size_t push(); // Write queued responses and scheduled results // Convenience method (combines pull + tick + push) size_t update(u32 currentTimeMs); private: RpcStream* mStream = nullptr; fl::vector<fl::json> mOutgoingQueue; // Queued responses to send fl::string mInputBuffer; // Partial line buffer for reading };

设计要点：mOutgoingQueue用于缓存"即时响应"，mInputBuffer用于缓存跨多次read()的不完整行（半行缓冲）；update()只是一个组合便利方法，并不替代显式调用。

3.3pull()：读取并处理输入

pull()从流中逐字节读取，按行分隔符（\n/\r）切分完整 JSON 请求，解析后调用processRpc()，并把构建好的响应 JSON 压入发送队列：

size_t Remote::pull() { if (!mStream) return 0; size_t processed = 0; // Read available data into buffer while (mStream->available()) { int c = mStream->read(); if (c < 0) break; if (c == '\n' || c == '\r') { if (!mInputBuffer.empty()) { // Process complete line fl::json doc = fl::json::parse(mInputBuffer); auto response = processRpc(fl::Remote::RpcRequest{ doc["function"] | fl::string(""), doc["args"], static_cast<u32>(doc["timestamp"] | 0) }); // Build response JSON and queue it fl::json responseJson = buildResponseJson(response); mOutgoingQueue.push_back(responseJson); mInputBuffer.clear(); processed++; } } else { mInputBuffer += static_cast<char>(c); } } return processed; }

注意这里的两个细节：请求字段是function/args/timestamp（对应仓库中 JSON-RPC 请求的既有约定，见 Remote.ino 顶部注释中的命令示例）；返回值processed表示本轮处理的请求数，便于调用方监控负载。

3.4push()：写出排队响应

push()分两段输出：先清空mOutgoingQueue中的即时响应，再输出getResults()拿到的调度结果，两者统一走既有的printJson()辅助函数：

size_t Remote::push() { if (!mStream) return 0; size_t written = 0; // Write queued immediate responses while (!mOutgoingQueue.empty()) { const fl::json& response = mOutgoingQueue[0]; printJson(response); // Uses existing printJson helper mOutgoingQueue.erase(mOutgoingQueue.begin()); written++; } // Write scheduled results auto results = getResults(); for (const auto& r : results) { printJson(r.to_json()); written++; } return written; }

这一步直接解决了"输出不一致"问题——即时响应与调度结果在push()内部统一出口、统一格式。

3.5update()：一行搞定一切

size_t Remote::update(u32 currentTimeMs) { size_t processed = pull(); size_t executed = tick(currentTimeMs); size_t written = push(); return processed + executed + written; }

返回值把"处理了几条请求 + 执行了几个调度任务 + 写了几条响应"汇总成一个计数，方便用户诊断单次循环的工作量。

3.6 更新后的示例用法

设计文档给出了改造后的完整用法，这也是调查目标"简单场景一行调用、进阶场景显式控制"的直接体现：

void setup() { Serial.begin(115200); FastLED.addLeds<WS2812, DATA_PIN>(leds, NUM_LEDS); // Register functions remote.bind("setLed", [](int i, int r, int g, int b) { leds[i] = CRGB(r, g, b); }); // Attach stream (optional - can still use manual processRpc) remote.attachStream(&Serial); } void loop() { // Option 1: Simple (single call) remote.update(millis()); // Option 2: Explicit (more control) // remote.pull(); // Read input, queue responses // remote.tick(millis()); // Process scheduled calls // remote.push(); // Write all output FastLED.show(); delay(10); }

两种用法在同一个loop()内可以按需切换，且FastLED.show()与delay(10)的动画主循环节奏完全不受影响。

四、向后兼容性保证

调查文档把"不破坏既有用法"列为设计的硬约束，明确四点：

• 手动processRpc()仍然可用；
• 流附着（attachStream）是可选能力；
• 既有示例无需修改即可继续工作；
• 新示例可以使用更简单的流式 API。

这与仓库的实际实现理念完全一致：在 remote.h 中，Server与Remote的构造函数都接受回调式的RequestSource/ResponseSink，同时保留processRpc()、bind()、get()等手工调用入口；attachStream式的新接口即使加入，也只是叠加层而非替换层。

五、替代方案：模板化流适配器（避免虚函数）

调查文档还评估了一个"零虚函数开销"的变体——把Remote本身做成模板类，直接持有具体流类型：

template<typename TStream> class Remote { public: void attachStream(TStream* stream) { mStream = stream; } size_t pull() { if (!mStream) return 0; // Read from mStream directly (no virtual calls) while (mStream->available()) { int c = mStream->read(); // ... } } private: TStream* mStream = nullptr; }; // Usage fl::Remote<decltype(Serial)> remote; remote.attachStream(&Serial);

文档明确否决了这条路径：它带来的"破坏性变更"与向后兼容硬约束冲突，因此推荐保留抽象RpcStream（ArduinoStreamAdapter<T>模板足以覆盖零开销诉求——适配器是模板化的，只有Remote面向抽象接口）。

六、仓库实际落地：回调式 Server 如何实现分相 I/O

调查文档提出的pull()/push()/update()分相设计，在仓库中确实以 Server 基类 的形式落地了——只不过传输抽象从RpcStream*换成了两个fl::function回调。这是理解该调查文档价值的关键对照点。

6.1 Server 的接口与三个阶段方法

class Server { public: using RequestSource = fl::function<fl::optional<fl::json>()>; using ResponseSink = fl::function<void(const fl::json&)>; using RequestHandler = fl::function<fl::json(const fl::json&)>; Server() FL_NO_EXCEPT; // no-op callbacks Server(RequestSource source, ResponseSink sink); void setRequestHandler(RequestHandler handler); void setRequestSource(RequestSource source); void setResponseSink(ResponseSink sink); size_t update(); // pull + push size_t pull(); // Pull requests from source, process, queue responses size_t push(); // Push queued responses to sink protected: RequestSource mRequestSource; ResponseSink mResponseSink; RequestHandler mRequestHandler; fl::vector<fl::json> mOutgoingQueue; };

与调查文档的 Option B 设计一一对应：

• mOutgoingQueue正是文档中"需要维护的响应队列"（见 server.h 与 server.cpp.hpp）；
• pull()对应"读取可用输入、处理 RPC、排队响应"；
• push()对应"把排队的响应写到 sink"；
• update()即pull() + push()的组合，与文档中update() = pull + tick + push的骨架一致。

6.2pull()/push()的实际实现

server.cpp.hpp 中的pull()从mRequestSource循环拉取请求，交给mRequestHandler处理后，通过信封标记过滤掉"调度确认"（scheduled:true）与"异步跳过"（noEnqueue/__skip，见注释中的 #3228 兼容说明），其余响应压入mOutgoingQueue：

size_t Server::pull() { if (!mRequestSource || !mRequestHandler) { return 0; } size_t processed = 0; while (auto optRequest = mRequestSource()) { fl::json request = fl::move(*optRequest); fl::json response = mRequestHandler(request); bool isScheduledAck = response.contains("scheduled") && response["scheduled"].as_bool().value_or(false); bool isAsyncSkip = (response.contains("noEnqueue") && response["noEnqueue"].as_bool().value_or(false)) || (response.contains("__skip") && response["__skip"].as_bool().value_or(false)); if (!response.is_null() && !isScheduledAck && !isAsyncSkip) { mOutgoingQueue.push_back(fl::move(response)); } processed++; } return processed; } size_t Server::push() { if (!mResponseSink) { return 0; } size_t sent = 0; while (!mOutgoingQueue.empty()) { mResponseSink(mOutgoingQueue[0]); mOutgoingQueue.erase(mOutgoingQueue.begin()); sent++; } return sent; }

这段实现印证了调查文档的三个设计判断：显式分相（pull/push 独立）、响应队列（mOutgoingQueue缓存）、阶段可测（source/handler/sink 全部可注入）。

6.3Remote::update()的组合语义

remote.cpp.hpp 中Remote::update(u32 currentTimeMs)在 Server 的 pull/push 之上叠加了tick()调度阶段，完整实现调查文档"pull + tick + push"的组合：

size_t Remote::update(u32 currentTimeMs) { size_t processed = Server::pull(); // Pull requests from Server size_t executed = tick(currentTimeMs); // Process scheduled tasks // Push scheduled results as JSON-RPC responses for (const auto& r : mResults) { fl::json response = fl::json::object(); response.set("result", r.result); mOutgoingQueue.push_back(response); } size_t sent = Server::push(); // Push responses from Server return processed + executed + sent; }

调度结果（mResults中的RpcResult）在这里被统一包装成{"result": ...}响应并送入mOutgoingQueue，与push()阶段一起输出——正是调查文档针对"即时响应 vs 调度结果不一致"问题给出的收敛方案。

6.4RpcResult：调度结果的元数据结构

调查文档中的getResults()对应仓库的RpcResult结构（src/fl/remote/types.h），携带完整的执行时序信息，to_json()序列化为紧凑单行格式：

struct RpcResult { fl::string functionName; // Name of function that executed fl::json result; // Return value (null if no return) u32 scheduledAt; // Timestamp when scheduled (0 for immediate) u32 receivedAt; // Timestamp when RPC request received u32 executedAt; // Timestamp when function executed bool wasScheduled; // true if scheduled, false if immediate fl::json to_json() const; // Serialize result to JSON object };

6.5 实际示例：Remote.ino 的回调式用法

当前仓库的 examples/Remote/Remote.ino 正是调查文档目标状态的"简洁版"——借助回调式 I/O 与remote.update(millis())，主循环已经从"40 行手工样板"收敛为极简结构：

void loop() { // Check for incoming JSON RPC via serial and queue it if (Serial.available()) { fl::string jsonRpc = readSerialJson(); fl::json doc = fl::json::parse(jsonRpc); requestQueue.push_back(doc); } // Process all queued requests: pull + tick + push remote.update(millis()); // Drain response queue and output to serial while (!responseQueue.empty()) { const auto& response = responseQueue[0]; Serial.println(response.to_string().c_str()); responseQueue.erase(responseQueue.begin()); } FastLED.show(); delay(10); }

其中remote的构造即回调式传输注入（等价于调查文档 Option C 的形态，但阶段划分走 Option B）：

fl::Remote remote( // RequestSource: pull from queue []() -> fl::optional<fl::json> { if (requestQueue.empty()) { return fl::nullopt; } auto req = fl::move(requestQueue[0]); requestQueue.erase(requestQueue.begin()); return req; }, // ResponseSink: push to queue [](const fl::json& response) { responseQueue.push_back(response); } );

示例同时展示了文档未展开的能力：bind()注册即时命令（setLed、fill、setBrightness）与带返回值的查询（millis、micros、getStatus、getLed），以及rpc.discover获取扁平 schema（flat tuple 格式，为低内存设备优化）。

6.6 测试验证：分相与可注入性

调查文档强调的"可测试性（可注入 mock 流）"在仓库测试中得到直接验证。tests/fl/remote/remote.cpp 定义了TestIO夹具——用两个fl::vector<fl::json>队列模拟请求源与响应汇：

struct TestIO { fl::vector<fl::json> requests; fl::vector<fl::json> responses; size_t requestIndex = 0; fl::optional<fl::json> pullRequest() { if (requestIndex >= requests.size()) { return fl::nullopt; } return requests[requestIndex++]; } void pushResponse(const fl::json& response) { responses.push_back(response); } // ... };

基于该夹具的测试覆盖了构造、方法注册（void 返回 / 带返回值 / 带 Config / 解绑 / 按签名获取get<int(int,int)>("add")）、即时执行、调度执行等场景——这正是"pull/push 各阶段可独立测试"的设计收益。

此外，tests/fl/remote/loopback.cpp 展示了该分相设计在网络流上的完整闭环：HttpStreamServer作为传输，服务端server_remote.update(now)在后台线程中运行，客户端通过 TCP 发送{"jsonrpc":"2.0","method":"add","params":[5,7],"id":1}并等待result返回。这表明"流抽象 + pull/push"的 I/O 协调层对 Serial 与 Network 两类传输同样成立。

七、低内存目标上的工程约束

仓库实现中有一处调查文档未涉及、但对嵌入式场景至关重要的工程细节：调度与结果追踪路径受FL_PLATFORM_HAS_LARGE_MEMORY宏门控（见 remote.h 与 remote.cpp.hpp 中的 #3224 Tier 1B 注释）。在低内存设备（如 LPC8xx 整型 RPC 契约）上：

• mAsyncRequests、mScheduler、mResults等字段与调度逻辑整体被裁剪，可回收每个Remote实例数百字节的 .bss；
• sendAsyncResponse()等异步方法退化为带FL_WARN提示的 no-op，保持源码兼容；
• processRpc()仅走即时执行路径。

这说明：调查文档设计的"调度结果输出"能力是有内存成本上限的，在设计流式 I/O API 时需为低内存 tier 保留退化路径——这也是pull/push/update分相设计的又一优势：各阶段独立，编译器可对未使用路径做链接级 DCE。

八、审查问题清单与设计结论

调查文档在结尾列出了六个待评审问题，这些问题的答案决定了最终 API 的形态：

1. pull()应自动解析 JSON，还是只读行、把解析交给用户？（仓库实践：Server::pull()直接消费fl::json请求，解析在 source 回调侧完成）
2. push()应自动调用printJson()，还是让用户控制格式？（仓库实践：sink 回调拥有传输级 framing，如前缀与换行）
3. 是否支持多流（如 Serial + Network）？（回调式 Server 天然支持——每个 Remote 实例绑定一组 source/sink）
4. 响应构建是否可定制（用户定义 formatter）？（可通过setRequestHandler()与自定义 sink 实现）
5. 应缓冲不完整 JSON 行，还是要求每次读取都是完整行？（仓库的 Serial 示例通过readSerialJson()在应用层保证完整行；调查文档 Option B 则用mInputBuffer做半行缓冲）
6. update()应存在，还是强制用户显式调用 pull/tick/push？（仓库两者并存，update()作为便利方法保留）

最终推荐结论

调查文档给出的推荐实施清单如下：

1. 新增RpcStream抽象接口；
2. 新增ArduinoStreamAdapter<T>模板，兼容 ArduinoStream；
3. 为Remote增加attachStream/detachStream/hasStream；
4. 为Remote增加pull/push/update方法；
5. 与手动 API 保持完整向后兼容。

预期收益（文档原文要点）：

• ✅ 简单用例有简洁 API（update()）；
• ✅ 进阶用例有显式控制（pull/tick/push）；
• ✅ 可测试性（可注入 mock 流）；
• ✅ 向后兼容（可选特性）；
• ✅ 非流式用法无虚函数调用开销；
• ✅ 兼容任何 Arduino Stream 兼容类。

对照仓库现状可以看到：其中第 1、2 项的"流抽象"最终以回调式RequestSource/ResponseSink形式落地（server.h），第 4、5 项的"分相与兼容"已完全实现（server.cpp.hpp、remote.cpp.hpp），第 3 项的attachStream便利接口则可由回调构造在应用层等价实现（见 Remote.ino 的requestQueue/responseQueue模式）。对于希望进一步演进该模块的开发者，这份调查文档连同上述源码、示例与测试，构成了从"设计动机 → 方案权衡 → 落地对照"的完整参考链路。

• 嵌入式
• 物联网
• 硬件开发
• 驱动开发