Method resolveRpcStream

src/fl/net/http/stream_transport.cpp.hpp:158–207

}

bool HttpStreamTransport::resolveRpcStream(const fl::json& msg, const fl::string& idKey) {
    auto it = mPendingStreams.find(idKey);
    if (it == mPendingStreams.end()) {
        return false;
    }
    PendingStream* pending = &it->second;

    // Check for error
    if (msg.contains("error")) {
        fl::task::Error err(msg["error"]["message"].as_string().value());
        pending->promise.complete_with_error(err);
        mPendingStreams.erase(idKey);
        return true;
    }

    // Check for ACK
    if (msg.contains("result") && msg["result"].contains("acknowledged")) {
        if (msg["result"]["acknowledged"].as_bool() == true) {
            pending->ackReceived = true;
            return true;
        }
    }

    // Check for stream update
    if (msg.contains("result") && msg["result"].contains("update")) {
        if (pending->updateCallback && *pending->updateCallback) {
            (*pending->updateCallback)(msg["result"]["update"]);
        }
        return true;
    }

    // Check for stream final (stop marker)
    if (msg.contains("result") && msg["result"].contains("stop")) {
        if (msg["result"]["stop"].as_bool() == true) {
            // Resolve with the value field if present, otherwise the full result
            if (msg["result"].contains("value")) {
                pending->promise.complete_with_value(msg["result"]["value"]);
            } else {
                pending->promise.complete_with_value(msg["result"]);
            }
            mPendingStreams.erase(idKey);
            return true;
        }
    }

    // Unrecognized message shape - treat as final
    pending->promise.complete_with_value(msg);
    mPendingStreams.erase(idKey);
    return true;
}

fl::task::Promise<fl::json> HttpStreamTransport::rpc(const fl::string& method, const fl::json& params) {
    int id = mNextCallId++;