Rust 的 axum streams

库为 axum 网络框架 提供 HTTP 响应流支持

• JSON 数组流格式
  • 当您需要在某些对象（例如第一个级别）内部包含数组时，支持简单包装结构
• JSON 行流格式
• CSV 流
• Protobuf 长度前缀流格式
• Apache Arrow IPC 流格式
• 文本流

此类响应在您需要从某些来源（如数据库、文件等）读取大量对象流时非常有用，并且希望避免大量内存分配。

快速入门

Cargo.toml

[dependencies]
axum-streams = { version = "0.18", features=["json", "csv", "protobuf", "text"] }

兼容性矩阵

示例代码

#[derive(Debug, Clone, Deserialize, Serialize)]
struct MyTestStructure {
  some_test_field: String
}

fn my_source_stream() -> impl Stream<Item=MyTestStructure> {
  // Simulating a stream with a plain vector and throttling to show how it works
  stream::iter(vec![
    MyTestStructure {
      some_test_field: "test1".to_string()
    }; 1000
  ]).throttle(std::time::Duration::from_millis(50))
}

async fn test_json_array_stream() -> impl IntoResponse {
  StreamBodyAs::json_array(source_test_stream())
}

async fn test_json_nl_stream() -> impl IntoResponse {
  StreamBodyAs::json_nl(source_test_stream())
}

async fn test_csv_stream() -> impl IntoResponse {
  StreamBodyAs::csv(source_test_stream())
}

async fn test_text_stream() -> impl IntoResponse {
  StreamBodyAs::text(source_test_stream())
}

所有示例均可在 examples 目录 中找到。

要运行示例，请使用

# cargo run --example json-example --features json

需要客户端支持？

对于以下功能，也存在相同的功能

• reqwest-streams.

帧大小的配置

默认情况下，库为流中的每个项目生成一个 HTTP 帧。您可以使用 StreamAsOptions 来更改此设置

    StreamBodyAsOptions::new().buffering_ready_items(1000)
        .json_array(source_test_stream())

错误处理

库提供了一种在流中传播错误的方式

struct MyError {
    message: String,
}

impl Into<axum::Error> for MyError {
    fn into(self) -> axum::Error {
        axum::Error::new(self.message)
    }
}

fn my_source_stream() -> impl Stream<Item=Result<MyTestStructure, MyError>> {
  // Simulating a stream with a plain vector and throttling to show how it works
  stream::iter(vec![
    Ok(MyTestStructure {
      some_test_field: "test1".to_string()
    }); 1000
  ])
}

async fn test_json_array_stream() -> impl IntoResponse {
  // Use _with_errors functions or directly `StreamBodyAs::with_options` 
  // to produce a stream with errors
  StreamBodyAs::json_array_with_errors(source_test_stream())
}

另一个对象内的 JSON 数组

有时您需要在某些对象内包含您的数组，例如

{
  "some_status_field": "ok",
  "data": [
    {
      "some_test_field": "test1"
    },
    {
      "some_test_field": "test2"
    }
  ]
}

包含 data 字段的包装对象称为信封。

您需要定义这两个结构：信封和记录

#[derive(Debug, Clone, Deserialize, Serialize)]
struct MyEnvelopeStructure {
    something_else: String,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    data: Vec<MyItem>
}

#[derive(Debug, Clone, Deserialize, Serialize)]
struct MyItem {
  some_test_field: String
}

并使用 json_array_with_envelope 代替 json_array。有关详细信息，请参阅 json-array-complex-structure.rs。

支持有限

• 仅支持第一层嵌套以避免具有性能影响的复杂实现。
• 您需要从 envelope 结构中移除目标数组字段，或者使用这个 Serde 技巧来避免 JSON 序列化问题。

    #[serde(skip_serializing_if = "Vec::is_empty")]

许可证

Apache 软件许可证 (ASL)

作者

Abdulla Abdurakhmanov