feat: streaming response support + CI tests

Signed-off-by: Xin Liu <[email protected]>

Author: apepkuss

.github/workflows/test.yml

@@ -17,6 +17,7 @@ on:
       - '**/*.rs'
       - '**/*.hurl'
       - 'run_tests.sh'
+      - 'test_streaming.sh'
       - 'tests/**'
   pull_request:
     branches: [ main ]
@@ -252,3 +253,79 @@ jobs:
           docker rm json-api || true
           docker stop ws-echo || true
           docker rm ws-echo || true
+
+  streaming-test:
+    name: Streaming Response Test
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        rust: [1.90.0]
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v5
+
+      - name: Install Rust-stable
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          toolchain: ${{ matrix.rust }}
+
+      - name: Cache Rust dependencies
+        uses: Swatinem/rust-cache@v2
+        with:
+          cache-on-failure: true
+
+      - name: Install Python dependencies
+        run: |
+          python3 --version
+          pip3 --version
+
+      - name: Install SQLite
+        run: sudo apt-get update && sudo apt-get install -y sqlite3
+
+      - name: Install jq (for JSON parsing in tests)
+        run: sudo apt-get install -y jq
+
+      - name: Verify tools
+        run: |
+          echo "Rust version:"
+          cargo --version
+          rustc --version
+          echo ""
+          echo "Python version:"
+          python3 --version
+          echo ""
+          echo "SQLite version:"
+          sqlite3 --version
+          echo ""
+          echo "jq version:"
+          jq --version
+
+      - name: Run streaming tests
+        env:
+          CI: true
+          TEST_PROXY_PORT: 8080
+          TEST_MOCK_PORT: 10086
+        run: |
+          chmod +x test_streaming.sh
+          ./test_streaming.sh
+
+      - name: Upload test logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: streaming-test-logs
+          path: |
+            /tmp/proxy.log
+            /tmp/mock-server.log
+            test_streaming.db
+          retention-days: 7
+
+      - name: Display logs on failure
+        if: failure()
+        run: |
+          echo "=== Proxy Logs ==="
+          cat /tmp/proxy.log || echo "No proxy log found"
+          echo ""
+          echo "=== Mock Server Logs ==="
+          cat /tmp/mock-server.log || echo "No mock server log found"

Cargo.lock

@@ -15,7 +15,7 @@ clap = { version = "4.5", features = ["derive", "env"] }
 futures-util = "0.3"
 
 # HTTP client - Using rustls instead of native-tls/openssl
-reqwest = { version = "0.12", features = ["json", "rustls-tls"], default-features = false }
+reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream"], default-features = false }
 
 # Serialization
 serde = { version = "1.0", features = ["derive"] }

Cargo.toml

@@ -36,6 +36,7 @@ A high-performance proxy server built with Rust, supporting HTTP/HTTPS and WebSo
       - [Option 1: With Docker Services (Recommended)](#option-1-with-docker-services-recommended)
       - [Option 2: Manual Service Management](#option-2-manual-service-management)
       - [Option 3: Individual Test Suites](#option-3-individual-test-suites)
+      - [Streaming Tests](#streaming-tests)
     - [Test Services](#test-services)
     - [Benefits of Docker-Based Testing](#benefits-of-docker-based-testing)
     - [CI/CD](#cicd)
@@ -56,6 +57,7 @@ A high-performance proxy server built with Rust, supporting HTTP/HTTPS and WebSo
 
 - 🚀 **High-Performance Async Proxy**: Built on Tokio and Axum
 - 🔄 **Protocol Support**: Supports HTTP/HTTPS and WebSocket proxying
+- 🌊 **Streaming Support**: Native support for streaming responses (SSE, LLM APIs, chunked encoding)
 - 💾 **Session Management**: Uses SQLite database to store session information
 - 🎯 **Dynamic Routing**: Dynamically forwards to different downstream servers based on session_id
 - ⚡ **Connection Pooling**: Built-in database connection pool and HTTP client connection pool
@@ -401,10 +403,33 @@ cargo test --test integration
 
 # Hurl HTTP API tests only (requires services running)
 hurl --test --variable port=8080 tests/http.hurl
+
+# Streaming response tests (独立测试)
+./test_streaming.sh
 ```
 
 **Note:** WebSocket tests are only available in Rust integration tests (`tests/integration.rs`), as Hurl doesn't support WebSocket message protocol.
 
+#### Streaming Tests
+
+流式传输测试验证 ss-proxy 对流式响应的支持，包括 LLM API（如 OpenAI）的流式输出：
+
+```bash
+# 运行完整的流式传输测试套件
+./test_streaming.sh
+
+# 使用自定义端口
+TEST_PROXY_PORT=9090 TEST_MOCK_PORT=10087 ./test_streaming.sh
+```
+
+测试覆盖：
+- ✅ 非流式请求转发 (`stream=false`)
+- ✅ 流式请求转发 (`stream=true`)
+- ✅ SSE (Server-Sent Events) 格式验证
+- ✅ 首字节延迟 (TTFB) 性能测试
+
+详见：[流式传输测试文档](tests/STREAMING_TEST_README.md)
+
 ### Test Services
 
 The test suite uses the following Docker services (all run locally):
@@ -427,10 +452,17 @@ All services are automatically managed by `run_tests.sh` when `USE_DOCKER_SERVIC
 
 ### CI/CD
 
-The project includes a GitHub Actions workflow (`.github/workflows/test.yml`) that automatically:
+The project includes GitHub Actions workflows that automatically:
+
+**Test Workflow** (`.github/workflows/test.yml`):
 - Runs all tests on push/PR
+- Two independent test jobs:
+  - `test`: HTTP/HTTPS and WebSocket proxy tests
+  - `streaming-test`: Streaming response tests (新增)
 - Uses service containers for test dependencies
 - Caches Rust dependencies for faster builds
+
+**Build Workflow** (`.github/workflows/build.yml`):
 - Runs linting and formatting checks
 - Builds binaries for multiple platforms
 

README.md

@@ -1,3 +1,4 @@
+use futures_util::StreamExt;
 use reqwest::{Client, Method, Request, Url};
 use std::time::Duration;
 use tracing::{error, info};
@@ -78,15 +79,18 @@ impl HttpProxy {
             builder = builder.header(key, value);
         }
 
-        // Get response body
-        let body_bytes = response.bytes().await.map_err(|e| {
-            error!("Failed to read response body: {}", e);
-            ProxyError::ResponseReadFailed(e.to_string())
-        })?;
+        // Convert reqwest response stream to axum body for streaming support
+        let stream = response.bytes_stream();
+        let body_stream = stream.map(|result| {
+            result.map_err(|e| {
+                error!("Error reading response stream: {}", e);
+                std::io::Error::new(std::io::ErrorKind::Other, e)
+            })
+        });
 
-        // Build final response
+        // Build final response with streaming body
         let final_response = builder
-            .body(axum::body::Body::from(body_bytes))
+            .body(axum::body::Body::from_stream(body_stream))
             .map_err(|e| {
                 error!("Failed to build response: {}", e);
                 ProxyError::ResponseBuildFailed(e.to_string())
@@ -105,9 +109,6 @@ pub enum ProxyError {
     #[error("Request failed: {0}")]
     RequestFailed(String),
 
-    #[error("Failed to read response: {0}")]
-    ResponseReadFailed(String),
-
     #[error("Failed to build response: {0}")]
     ResponseBuildFailed(String),
 }