如何在Ballerina的HTTP资源函数中返回CSV读取的Record Stream？

核心思路 #

Ballerina的HTTP模块原生支持流式响应，无需将Stream转换为数组即可直接返回。关键是通过分块传输编码（Chunked Transfer Encoding）实现数据的流式输出，同时根据需求将Record类型的Stream转换为对应媒体格式的Stream（如JSON或CSV）。

实现步骤与代码示例 #

1. 基础准备：定义Record类型与导入依赖

首先定义与CSV结构匹配的Record类型，并导入必要的模块：

import ballerina/http;
import ballerina/csv;
import ballerina/json;

// 与CSV文件字段对应的Record类型
type Employee record {
    string id;
    string name;
    decimal salary;
};

2. 返回JSON格式的流式响应

将CSV读取的Record Stream转换为JSON字符串Stream，再通过HTTP响应返回：

service /employees on new http:Listener(9090) {
    resource function get json-stream() returns http:Response|error {
        // 读取CSV文件为Record类型的Stream
        stream<Employee, error?> employeeStream = csv:read("employees.csv", Employee);

        // 将每个Record转换为JSON字符串
        stream<string, error?> jsonStream = employeeStream.map(function(Employee emp) returns string {
            return check json:toString(emp);
        });

        // 构建流式响应
        http:Response response = new;
        response.setPayload(jsonStream);
        response.setHeader("Content-Type", "application/json");
        // 启用分块传输，适配流式输出
        response.setHeader("Transfer-Encoding", "chunked");

        return response;
    }
}

3. 返回CSV格式的流式响应

如果需要直接返回CSV格式的流式数据，可利用csv:writeStream将Record Stream转换回CSV行的Stream：

service /employees on new http:Listener(9090) {
    resource function get csv-stream() returns http:Response|error {
        stream<Employee, error?> employeeStream = csv:read("employees.csv", Employee);

        // 将Record Stream转换为CSV行的Stream
        stream<string, error?> csvStream = csv:writeStream(employeeStream, Employee);

        http:Response response = new;
        response.setPayload(csvStream);
        response.setHeader("Content-Type", "text/csv");
        response.setHeader("Transfer-Encoding", "chunked");

        return response;
    }
}

关键注意事项 #

• 分块传输编码：必须设置Transfer-Encoding: chunked头部，确保服务器逐块发送数据，避免一次性加载所有数据到内存，实现真正的流式处理。
• 错误处理：Stream的处理过程（如map或writeStream）可能产生错误，需通过check或自定义错误逻辑处理异常。
• 自动序列化：若直接将Stream<Employee, error?>作为响应payload，HTTP模块会自动将每个Record序列化为JSON对象，无需手动转换为字符串Stream。

内容的提问来源于stack exchange，提问作者Lakshan Weerasinghe