// 在 ngx_stream_block() 中初始化
static char *ngx_stream_block(ngx_conf_t *cf, ngx_command_t *cmd, void *conf) {
    // 1) 创建 Stream 配置上下文
    ctx = ngx_pcalloc(cf->pool, sizeof(ngx_stream_conf_ctx_t));
    
    // 2) 统计 Stream 模块数量
    ngx_stream_max_module = ngx_count_modules(cf->cycle, NGX_STREAM_MODULE);
    
    // 3) 创建 main_conf 和 srv_conf 数组
    ctx->main_conf = ngx_pcalloc(cf->pool, sizeof(void *) * ngx_stream_max_module);
    ctx->srv_conf = ngx_pcalloc(cf->pool, sizeof(void *) * ngx_stream_max_module);
    
    // 4) 调用各模块的 preconfiguration
    // 5) 解析 stream{} 配置块
    // 6) 初始化阶段处理器
    ngx_stream_init_phases(cf, cmcf);
    ngx_stream_init_phase_handlers(cf, cmcf);
    
    // 7) 优化监听端口和服务器配置
    return ngx_stream_optimize_servers(cf, cmcf, cmcf->ports);
}

void ngx_stream_init_connection(ngx_connection_t *c) {
    // 1) 创建 Stream 会话对象
    s = ngx_pcalloc(c->pool, sizeof(ngx_stream_session_t));
    s->signature = NGX_STREAM_MODULE;  // "STRM"
    s->connection = c;
    s->start_sec = ngx_time();
    s->start_msec = ngx_current_msec;
    
    // 2) 查找服务器配置
    cscf = ngx_stream_get_module_srv_conf(s, ngx_stream_core_module);
    
    // 3) 创建模块上下文数组
    s->ctx = ngx_pcalloc(c->pool, sizeof(void *) * ngx_stream_max_module);
    s->main_conf = addr_conf->ctx->main_conf;
    s->srv_conf = addr_conf->ctx->srv_conf;
    
    // 4) 初始化变量数组
    s->variables = ngx_pcalloc(c->pool, cmcf->variables.nelts

                                * sizeof(ngx_stream_variable_value_t));
    
    // 5) 设置日志处理器
    c->log->handler = s->log_handler;
    
    // 6) 设置读事件处理器，开始阶段处理
    c->read->handler = ngx_stream_session_handler;
    
    // 7) 启动阶段处理流程
    ngx_stream_session_handler(c->read);

}

void ngx_stream_core_run_phases(ngx_stream_session_t *s) {
    // 遍历阶段处理器数组
    ph = cmcf->phase_engine.handlers;
    
    while (ph[s->phase_handler].checker) {
        // 调用阶段检查器
        rc = ph[s->phase_handler].checker(s, &ph[s->phase_handler]);
        
        if (rc == NGX_OK) {
            // 继续下一阶段
            return;
        }
        
        // 错误或完成，结束会话
    }
}

// 7 个阶段
// 1. NGX_STREAM_POST_ACCEPT_PHASE    - 连接接受后处理
// 2. NGX_STREAM_PREACCESS_PHASE      - 预访问控制
// 3. NGX_STREAM_ACCESS_PHASE         - 访问控制
// 4. NGX_STREAM_SSL_PHASE            - SSL 握手
// 5. NGX_STREAM_PREREAD_PHASE        - 预读取（如 SSL Preread）
// 6. NGX_STREAM_CONTENT_PHASE        - 内容处理（代理转发）
// 7. NGX_STREAM_LOG_PHASE            - 日志记录

// ngx_stream_proxy_module.c
static void ngx_stream_proxy_init_upstream(ngx_stream_session_t *s) {
    // 1) 创建 Upstream 对象
    u = ngx_pcalloc(c->pool, sizeof(ngx_stream_upstream_t));
    s->upstream = u;
    u->peer.log = c->log;
    u->peer.log_error = NGX_ERROR_ERR;
    
    // 2) 初始化 Upstream 配置
    u->proxy_protocol = pscf->proxy_protocol;
    u->start_sec = ngx_time();
    
    // 3) 初始化 Peer 连接（选择后端服务器）
    rc = ngx_stream_upstream_init_round_robin_peer(s, us);
    
    // 4) 连接后端服务器
    ngx_stream_proxy_connect(s);
}

static void ngx_stream_proxy_connect(ngx_stream_session_t *s) {
    // 1) 发起连接
    rc = ngx_event_connect_peer(&u->peer);
    
    if (rc == NGX_ERROR || rc == NGX_BUSY || rc == NGX_DECLINED) {
        // 连接失败，切换到下一个后端
        ngx_stream_proxy_next_upstream(s);
        return;
    }
    
    // 2) 连接成功或进行中
    pc = u->peer.connection;
    pc->data = s;
    pc->read->handler = ngx_stream_proxy_upstream_handler;
    pc->write->handler = ngx_stream_proxy_upstream_handler;
    
    // 3) 客户端读写事件处理器
    c->read->handler = ngx_stream_proxy_downstream_handler;
    c->write->handler = ngx_stream_proxy_downstream_handler;
    
    // 4) 开始数据转发
    ngx_stream_proxy_process(s, 0, 1);
}

static void ngx_stream_proxy_process(ngx_stream_session_t *s,
    ngx_uint_t from_upstream, ngx_uint_t do_write) {
    
    // 双向数据转发
    // from_upstream = 0: 客户端 -> 后端
    // from_upstream = 1: 后端 -> 客户端
    
    if (from_upstream) {
        src = u->peer.connection;
        dst = c;
    } else {
        src = c;
        dst = u->peer.connection;
    }
    
    // 读取数据
    n = src->recv(src, b->last, b->end - b->last);
    
    if (n == NGX_AGAIN) {
        // 等待更多数据
        return;
    }
    
    if (n == NGX_ERROR || n == 0) {
        // 连接关闭或错误
        ngx_stream_finalize_session(s, 0);
        return;
    }
    
    b->last += n;
    
    // 写入数据
    if (do_write) {
        n = dst->send(dst, b->pos, b->last - b->pos);
        b->pos += n;
    }
}

void ngx_stream_finalize_session(ngx_stream_session_t *s, ngx_uint_t rc) {
    // 1) 记录日志
    ngx_log_debug2(NGX_LOG_DEBUG_STREAM, c->log, 0,
                   "finalize stream session: %ui, c:%ui", rc, c->fd);
    
    // 2) 设置状态码
    s->status = rc;
    
    // 3) 执行 LOG 阶段处理器
    s->phase_handler = cmcf->phase_engine.handlers;
    ngx_stream_core_run_phases(s);
    
    // 4) 关闭 Upstream 连接
    if (s->upstream) {
        u = s->upstream;
        if (u->peer.connection) {
            ngx_close_connection(u->peer.connection);
        }
    }
    
    // 5) 关闭客户端连接
    ngx_close_connection(c);
}

flowchart TB
    subgraph "Client Layer"
        C1[TCP Client]
        C2[UDP Client]
    end
    
    subgraph "Nginx Stream Module"
        direction TB
        IC[ngx_stream_init_connection<br/>连接初始化]
        
        subgraph "7-Phase Engine"
            P1[POST_ACCEPT<br/>连接接受]
            P2[PREACCESS<br/>预访问控制]
            P3[ACCESS<br/>访问控制]
            P4[SSL<br/>SSL握手]
            P5[PREREAD<br/>预读取]
            P6[CONTENT<br/>内容处理]
            P7[LOG<br/>日志记录]
            
            P1 --> P2 --> P3 --> P4 --> P5 --> P6 --> P7
        end
        
        subgraph "Content Handlers"
            PROXY[Proxy Module<br/>ngx_stream_proxy]
            PASS[Pass Module<br/>ngx_stream_pass]
            RETURN[Return Module<br/>ngx_stream_return]
        end
        
        subgraph "Upstream Layer"
            LB[Load Balancer<br/>负载均衡器]
            RR[Round Robin]
            LC[Least Conn]
            HASH[Hash]
            RAND[Random]
            
            LB --> RR & LC & HASH & RAND
        end
        
        subgraph "Modules"
            ACCESS_M[Access Module<br/>访问控制]
            LIMIT_M[Limit Conn Module<br/>连接限制]
            SSL_M[SSL Module<br/>TLS支持]
            PREREAD_M[SSL Preread<br/>SNI路由]
        end
        
        IC --> P1
        P6 --> PROXY & PASS & RETURN
        PROXY --> LB
        
        P2 --> ACCESS_M & LIMIT_M
        P4 --> SSL_M
        P5 --> PREREAD_M
    end
    
    subgraph "Backend Layer"
        direction LR
        B1[Backend Server 1]
        B2[Backend Server 2]
        B3[Backend Server 3]
    end
    
    C1 & C2 --> IC
    LB --> B1 & B2 & B3
    
    style IC fill:#e1f5ff
    style P6 fill:#fff3e0
    style PROXY fill:#f3e5f5
    style LB fill:#e8f5e9

struct ngx_stream_session_s {
    uint32_t                       signature;         // "STRM" 魔数
    
    ngx_connection_t              *connection;        // 客户端连接
    
    off_t                          received;          // 接收字节数
    time_t                         start_sec;         // 会话开始时间（秒）
    ngx_msec_t                     start_msec;        // 会话开始时间（毫秒）
    
    void                         **ctx;               // 模块上下文数组
    void                         **main_conf;         // main 配置指针
    void                         **srv_conf;          // srv 配置指针
    
    ngx_stream_upstream_t         *upstream;          // Upstream 对象
    ngx_array_t                   *upstream_states;   // Upstream 状态数组
    
    ngx_stream_variable_value_t   *variables;         // 变量数组
    
    ngx_int_t                      phase_handler;     // 当前阶段处理器索引
    ngx_uint_t                     status;            // 会话状态码
    
    unsigned                       ssl:1;             // 是否使用 SSL
    unsigned                       stat_processing:1; // 是否正在处理
    unsigned                       health_check:1;    // 是否健康检查
    unsigned                       limit_conn_status:2; // 连接限制状态
};

struct ngx_stream_phase_handler_s {
    ngx_stream_phase_handler_pt    checker;     // 阶段检查器函数
    ngx_stream_handler_pt          handler;     // 具体处理器函数
    ngx_uint_t                     next;        // 下一个阶段索引
};

typedef struct {
    ngx_peer_connection_t          peer;              // Peer 连接
    
    ngx_buf_t                     *downstream_buf;    // 下行缓冲区
    ngx_buf_t                     *upstream_buf;      // 上行缓冲区
    
    off_t                          received;          // 从后端接收字节数
    time_t                         start_sec;         // Upstream 开始时间
    ngx_msec_t                     start_msec;
    
    ngx_uint_t                     responses;         // 响应次数
    ngx_uint_t                     connect_timeout;   // 连接超时
    ngx_uint_t                     timeout;           // 读写超时
    
    ngx_stream_upstream_srv_conf_t  *upstream;       // Upstream 配置
    
    unsigned                       connected:1;       // 是否已连接
    unsigned                       proxy_protocol:1;  // 是否使用 PROXY Protocol
} ngx_stream_upstream_t;

stateDiagram-v2
    [*] --> POST_ACCEPT: 连接建立
    POST_ACCEPT --> PREACCESS: realip, set variables
    PREACCESS --> ACCESS: 访问控制检查
    ACCESS --> SSL: SSL/TLS 握手
    SSL --> PREREAD: 预读取数据
    PREREAD --> CONTENT: 内容处理
    CONTENT --> LOG: 会话结束
    LOG --> [*]
    
    ACCESS --> [*]: 拒绝访问
    SSL --> [*]: SSL 错误
    CONTENT --> [*]: 代理错误

// ngx_stream_upstream_round_robin.c
ngx_int_t ngx_stream_upstream_get_round_robin_peer(ngx_peer_connection_t *pc,
    void *data) {
    
    rrp = data;
    peers = rrp->peers;
    
    // 遍历所有后端服务器
    for (i = 0; i < peers->number; i++) {
        peer = &peers->peer[rrp->current];
        
        // 检查服务器是否可用
        if (!peer->down && peer->max_fails == 0
            || peer->fails < peer->max_fails) {
            // 选择此服务器
            pc->sockaddr = peer->sockaddr;
            pc->socklen = peer->socklen;
            pc->name = &peer->name;
            
            rrp->current = (rrp->current + 1) % peers->number;
            return NGX_OK;
        }
        
        rrp->current = (rrp->current + 1) % peers->number;
    }
    
    return NGX_BUSY;  // 所有服务器不可用
}

// ngx_stream_upstream_least_conn_module.c
ngx_int_t ngx_stream_upstream_get_least_conn_peer(ngx_peer_connection_t *pc,
    void *data) {
    
    best = NULL;
    best_conns = 0;
    
    // 找到连接数最少的服务器
    for (i = 0; i < peers->number; i++) {
        peer = &peers->peer[i];
        
        if (peer->down || peer->fails >= peer->max_fails) {
            continue;
        }
        
        // 计算有效连接数（考虑权重）
        conns = peer->conns * peers->total_weight / peer->weight;
        
        if (best == NULL || conns < best_conns) {
            best = peer;
            best_conns = conns;
        }
    }
    
    if (best) {
        pc->sockaddr = best->sockaddr;
        pc->socklen = best->socklen;
        pc->name = &best->name;
        best->conns++;  // 增加连接计数
        return NGX_OK;
    }
    
    return NGX_BUSY;
}

// ngx_stream_upstream_hash_module.c
ngx_int_t ngx_stream_upstream_get_hash_peer(ngx_peer_connection_t *pc,
    void *data) {
    
    // 计算哈希值（基于变量，如 $remote_addr, $server_name）
    hash = ngx_crc32_short(key.data, key.len);
    
    // 一致性哈希环查找
    for (i = 0; i < 20 * peers->number; i++) {
        h = hash + i * 0x9e3779b9;  // Golden ratio
        
        server = h % peers->number;
        peer = &peers->peer[server];
        
        if (!peer->down && peer->fails < peer->max_fails) {
            pc->sockaddr = peer->sockaddr;
            pc->socklen = peer->socklen;
            pc->name = &peer->name;
            return NGX_OK;
        }
    }
    
    return NGX_BUSY;
}

stream {
    upstream backend {
        server 192.168.1.10:3306;
        server 192.168.1.11:3306;
        server 192.168.1.12:3306;
    }
    
    server {
        listen 3306;
        proxy_pass backend;
        proxy_connect_timeout 1s;
        proxy_timeout 3s;
    }
}

stream {
    upstream dns_servers {
        server 8.8.8.8:53;
        server 8.8.4.4:53;
    }
    
    server {
        listen 53 udp;
        proxy_pass dns_servers;
        proxy_responses 1;  # UDP 响应数量
        proxy_timeout 1s;
    }
}

stream {
    upstream backend {
        # 轮询（默认）
        # Round Robin
        
        # 最少连接
        least_conn;
        
        # 哈希（根据客户端 IP）
        # hash $remote_addr consistent;
        
        # 随机
        # random;
        
        server 192.168.1.10:6379 weight=3 max_fails=2 fail_timeout=10s;
        server 192.168.1.11:6379 weight=2;
        server 192.168.1.12:6379 weight=1;
        server 192.168.1.13:6379 backup;  # 备份服务器
    }
    
    server {
        listen 6379;
        proxy_pass backend;
        proxy_next_upstream on;
        proxy_next_upstream_tries 2;
        proxy_next_upstream_timeout 2s;
    }
}

stream {
    upstream secure_backend {
        server 192.168.1.10:443;
        server 192.168.1.11:443;
    }
    
    server {
        listen 443 ssl;
        
        ssl_certificate      /path/to/cert.pem;
        ssl_certificate_key  /path/to/key.pem;
        ssl_protocols        TLSv1.2 TLSv1.3;
        ssl_ciphers          HIGH:!aNULL:!MD5;
        ssl_session_cache    shared:SSL:10m;
        ssl_session_timeout  10m;
        
        # SSL Preread（根据 SNI 路由）
        ssl_preread on;
        proxy_pass $ssl_preread_server_name;
        
        # 或固定后端
        # proxy_pass secure_backend;
    }
}

stream {
    map $ssl_preread_server_name $backend {
        example.com      backend_example;
        test.com         backend_test;
        default          backend_default;
    }
    
    upstream backend_example {
        server 192.168.1.10:443;
    }
    
    upstream backend_test {
        server 192.168.1.20:443;
    }
    
    upstream backend_default {
        server 192.168.1.30:443;
    }
    
    server {
        listen 443;
        ssl_preread on;
        proxy_pass $backend;
    }
}

stream {
    # 定义地理位置变量
    geo $deny {
        default         0;
        192.168.1.0/24  1;
        10.0.0.0/8      1;
    }
    
    upstream backend {
        server 192.168.1.10:8080;
    }
    
    server {
        listen 8080;
        
        # IP 访问控制
        allow 192.168.1.0/24;
        allow 10.0.0.0/8;
        deny  all;
        
        # 连接数限制
        # limit_conn addr 10;
        
        proxy_pass backend;
    }
}

stream {
    log_format proxy '$remote_addr [$time_local] '
                     '$protocol $status $bytes_sent $bytes_received '
                     '$session_time "$upstream_addr" '
                     '"$upstream_bytes_sent" "$upstream_bytes_received" "$upstream_connect_time"';
    
    access_log /var/log/nginx/stream_access.log proxy;
    error_log /var/log/nginx/stream_error.log info;
    
    upstream redis_cluster {
        hash $remote_addr consistent;
        
        server 192.168.1.10:6379 weight=3 max_fails=2 fail_timeout=30s;
        server 192.168.1.11:6379 weight=2 max_fails=2 fail_timeout=30s;
        server 192.168.1.12:6379 weight=1 max_fails=2 fail_timeout=30s;
        server 192.168.1.13:6379 backup;
    }
    
    server {
        listen 6379;
        proxy_pass redis_cluster;
        
        proxy_connect_timeout 3s;
        proxy_timeout 10s;
        proxy_next_upstream on;
        proxy_next_upstream_tries 3;
        proxy_next_upstream_timeout 10s;
        
        proxy_buffer_size 16k;
        
        # 日志
        access_log /var/log/nginx/redis_access.log proxy;
    }
}

# 在 main 块配置
worker_processes auto;
worker_cpu_affinity auto;
worker_rlimit_nofile 65535;

events {
    use epoll;
    worker_connections 10240;  # Stream 需要两倍连接（客户端+后端）
    multi_accept on;
}

stream {
    # 缓冲区优化
    proxy_buffer_size 16k;
    
    # 超时优化
    proxy_connect_timeout 3s;
    proxy_timeout 1h;  # 长连接场景
    
    # Upstream 连接复用（商业版）
    # proxy_socket_keepalive on;
}

# 启用 stub_status (HTTP 块)
http {
    server {
        listen 8080;
        location /status {
            stub_status;
            allow 127.0.0.1;
            deny all;
        }
    }
}

# 监控指标
# - Active connections: 当前活跃连接数
# - Reading/Writing: 正在读取/写入的连接数
# - Upstream 状态: 后端服务器健康状况
# - 字节数统计: 上下行流量

# 查看 Stream 连接状态
ss -tnp | grep nginx

# 查看 Upstream 连接
netstat -an | grep ESTABLISHED | grep :6379

# 实时日志
tail -f /var/log/nginx/stream_access.log

# 测试配置
nginx -t

# 重载配置
nginx -s reload

stream {
    # 限制连接数
    limit_conn_zone $binary_remote_addr zone=addr:10m;
    
    server {
        listen 3306;
        
        # IP 白名单
        allow 192.168.1.0/24;
        deny  all;
        
        # 连接数限制
        limit_conn addr 10;
        
        # SSL 配置（如需要）
        ssl_protocols TLSv1.2 TLSv1.3;
        ssl_prefer_server_ciphers on;
        
        proxy_pass backend;
    }
}

void ngx_stream_init_connection(ngx_connection_t *c);

void ngx_stream_init_connection(ngx_connection_t *c) {
    // 1) 查找服务器配置（根据监听地址和端口）
    port = c->listening->servers;
    
    // 多地址情况下需要获取本地地址
    if (port->naddrs > 1) {
        ngx_connection_local_sockaddr(c, NULL, 0);
        // 根据本地地址查找对应配置
        sa = c->local_sockaddr;
        // ... 查找 addr_conf
    } else {
        // 单地址直接使用第一个配置
        addr_conf = &addr[0].conf;
    }
    
    // 2) 创建会话对象
    s = ngx_pcalloc(c->pool, sizeof(ngx_stream_session_t));
    s->signature = NGX_STREAM_MODULE;  // "STRM"
    s->connection = c;
    c->data = s;
    
    // 3) 关联配置上下文
    ctx = addr_conf->default_server->ctx;
    s->main_conf = ctx->main_conf;
    s->srv_conf = ctx->srv_conf;
    
    // 4) 创建模块上下文数组
    s->ctx = ngx_pcalloc(c->pool, sizeof(void *) * ngx_stream_max_module);
    
    // 5) 初始化变量数组
    cmcf = ngx_stream_get_module_main_conf(s, ngx_stream_core_module);
    s->variables = ngx_pcalloc(c->pool, cmcf->variables.nelts

                               * sizeof(ngx_stream_variable_value_t));
    
    // 6) 记录会话开始时间
    tp = ngx_timeofday();
    s->start_sec = tp->sec;
    s->start_msec = tp->msec;
    
    // 7) 设置读事件处理器
    rev = c->read;
    rev->handler = ngx_stream_session_handler;
    
    // 8) 处理 PROXY Protocol（如果启用）
    if (addr_conf->proxy_protocol) {
        rev->handler = ngx_stream_proxy_protocol_handler;
        ngx_add_timer(rev, cscf->proxy_protocol_timeout);
        return;
    }
    
    // 9) 启动阶段处理
    ngx_stream_session_handler(rev);

}

ngx_event_accept
  └─> ngx_listening_t->handler  // = ngx_stream_init_connection
        └─> ngx_stream_session_handler
              └─> ngx_stream_core_run_phases

sequenceDiagram
    autonumber
    participant E as Event Module
    participant SI as ngx_stream_init_connection
    participant S as Session Object
    participant PH as Phase Handler
    
    E->>SI: 接受连接
    SI->>SI: 查找服务器配置
    SI->>S: 创建会话对象
    SI->>S: 初始化上下文和变量
    SI->>PH: 设置读事件处理器
    PH->>PH: 启动阶段处理

void ngx_stream_session_handler(ngx_event_t *rev);

void ngx_stream_session_handler(ngx_event_t *rev) {
    c = rev->data;
    s = c->data;
    
    // 处理超时
    if (rev->timedout) {
        ngx_connection_error(c, NGX_ETIMEDOUT, "client timed out");
        ngx_stream_finalize_session(s, NGX_STREAM_OK);
        return;
    }
    
    // 启动阶段处理流程
    ngx_stream_core_run_phases(s);
}

void ngx_stream_finalize_session(ngx_stream_session_t *s, ngx_uint_t rc);

void ngx_stream_finalize_session(ngx_stream_session_t *s, ngx_uint_t rc) {
    c = s->connection;
    
    ngx_log_debug2(NGX_LOG_DEBUG_STREAM, c->log, 0,
                   "finalize stream session: %ui, c:%ui", rc, c->fd);
    
    // 设置状态码
    s->status = rc;
    
    // 已经在 LOG 阶段，直接关闭
    if (s->stat_processing || c->destroyed) {
        ngx_stream_close_connection(c);
        return;
    }
    
    // 执行 LOG 阶段
    s->stat_processing = 1;
    
    cmcf = ngx_stream_get_module_main_conf(s, ngx_stream_core_module);
    s->phase_handler = cmcf->phase_engine.log_phase_handler;
    
    ngx_stream_core_run_phases(s);
    
    // 关闭连接
    ngx_stream_close_connection(c);
}

void ngx_stream_core_run_phases(ngx_stream_session_t *s);

void ngx_stream_core_run_phases(ngx_stream_session_t *s) {
    cmcf = ngx_stream_get_module_main_conf(s, ngx_stream_core_module);
    ph = cmcf->phase_engine.handlers;
    
    // 遍历阶段处理器
    while (ph[s->phase_handler].checker) {
        
        // 调用阶段检查器
        rc = ph[s->phase_handler].checker(s, &ph[s->phase_handler]);
        
        if (rc == NGX_OK) {
            // 阶段处理完成，返回等待事件
            return;
        }
        
        // rc == NGX_DECLINED: 继续下一阶段
        // rc == NGX_AGAIN: 当前阶段需要更多数据
        // rc == NGX_ERROR/其他: 错误，结束会话
    }
}

ngx_int_t ngx_stream_core_generic_phase(ngx_stream_session_t *s,
    ngx_stream_phase_handler_t *ph);

ngx_int_t ngx_stream_core_generic_phase(ngx_stream_session_t *s,
    ngx_stream_phase_handler_t *ph) {
    
    // 调用具体的阶段处理器
    rc = ph->handler(s);
    
    if (rc == NGX_OK) {
        // 处理成功，进入下一阶段
        s->phase_handler = ph->next;
        return NGX_AGAIN;
    }
    
    if (rc == NGX_DECLINED) {
        // 跳过当前处理器，尝试下一个
        s->phase_handler++;
        return NGX_AGAIN;
    }
    
    if (rc == NGX_AGAIN || rc == NGX_DONE) {
        // 需要等待（如异步操作）
        return NGX_OK;
    }
    
    // 错误，结束会话
    ngx_stream_finalize_session(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
    return NGX_OK;
}

ngx_int_t ngx_stream_core_preread_phase(ngx_stream_session_t *s,
    ngx_stream_phase_handler_t *ph);

ngx_int_t ngx_stream_core_preread_phase(ngx_stream_session_t *s,
    ngx_stream_phase_handler_t *ph) {
    
    c = s->connection;
    c->log->action = "prereading client data";
    
    // 调用 PREREAD 处理器
    rc = ph->handler(s);
    
    if (rc == NGX_AGAIN) {
        // 需要等待更多数据
        if (ngx_handle_read_event(c->read, 0) != NGX_OK) {
            ngx_stream_finalize_session(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
            return NGX_OK;
        }
        
        return NGX_OK;  // 暂停阶段处理
    }
    
    if (rc == NGX_OK) {
        // 预读取成功，继续下一阶段
        s->phase_handler = ph->next;
        return NGX_AGAIN;
    }
    
    // 错误处理
    ngx_stream_finalize_session(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
    return NGX_OK;
}

ngx_int_t ngx_stream_core_content_phase(ngx_stream_session_t *s,
    ngx_stream_phase_handler_t *ph);

ngx_int_t ngx_stream_core_content_phase(ngx_stream_session_t *s,
    ngx_stream_phase_handler_t *ph) {
    
    c = s->connection;
    c->log->action = NULL;
    
    cscf = ngx_stream_get_module_srv_conf(s, ngx_stream_core_module);
    
    if (cscf->handler) {
        // 调用内容处理器（如 proxy_pass）
        cscf->handler(s);
        return NGX_OK;
    }
    
    // 没有配置内容处理器
    ngx_log_error(NGX_LOG_ERR, c->log, 0,
                  "no \"proxy_pass\" is defined for the server in %s:%ui",
                  cscf->file_name, cscf->line);
    
    ngx_stream_finalize_session(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
    return NGX_OK;
}

static void ngx_stream_proxy_handler(ngx_stream_session_t *s);

static void ngx_stream_proxy_handler(ngx_stream_session_t *s) {
    // 1) 创建 Upstream 对象
    u = ngx_pcalloc(c->pool, sizeof(ngx_stream_upstream_t));
    s->upstream = u;
    
    u->peer.log = c->log;
    u->peer.log_error = NGX_ERROR_ERR;
    
    // 2) 获取配置
    pscf = ngx_stream_get_module_srv_conf(s, ngx_stream_proxy_module);
    
    // 3) 初始化缓冲区
    u->downstream_buf = ngx_pcalloc(c->pool, sizeof(ngx_buf_t) + pscf->buffer_size);
    u->downstream_buf->start = (u_char *) u->downstream_buf + sizeof(ngx_buf_t);
    u->downstream_buf->end = u->downstream_buf->start + pscf->buffer_size;
    u->downstream_buf->pos = u->downstream_buf->start;
    u->downstream_buf->last = u->downstream_buf->start;
    
    u->upstream_buf = ngx_pcalloc(c->pool, sizeof(ngx_buf_t) + pscf->buffer_size);
    // ... 类似初始化
    
    // 4) 初始化 Upstream
    ngx_stream_proxy_init_upstream(s);
}

static void ngx_stream_proxy_init_upstream(ngx_stream_session_t *s);

static void ngx_stream_proxy_init_upstream(ngx_stream_session_t *s) {
    u = s->upstream;
    pscf = ngx_stream_get_module_srv_conf(s, ngx_stream_proxy_module);
    us = pscf->upstream;
    
    // 初始化 Peer（选择后端服务器）
    if (us) {
        // 使用 Upstream 配置
        rc = ngx_stream_upstream_init_round_robin_peer(s, us);
        
        if (rc != NGX_OK) {
            ngx_stream_finalize_session(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
            return;
        }
    } else {
        // 直接指定后端地址
        // ... 解析地址
    }
    
    u->start_sec = ngx_time();
    
    // 连接后端服务器
    ngx_stream_proxy_connect(s);
}

static void ngx_stream_proxy_connect(ngx_stream_session_t *s);

static void ngx_stream_proxy_connect(ngx_stream_session_t *s) {
    u = s->upstream;
    c = s->connection;
    
    // 发起连接
    rc = ngx_event_connect_peer(&u->peer);
    
    if (rc == NGX_ERROR) {
        ngx_stream_proxy_finalize(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
        return;
    }
    
    if (rc == NGX_BUSY || rc == NGX_DECLINED) {
        // 后端不可用，尝试下一个
        ngx_stream_proxy_next_upstream(s);
        return;
    }
    
    // 连接成功或进行中
    pc = u->peer.connection;
    pc->data = s;
    pc->log = c->log;
    
    // 设置事件处理器
    pc->read->handler = ngx_stream_proxy_upstream_handler;
    pc->write->handler = ngx_stream_proxy_upstream_handler;
    
    c->read->handler = ngx_stream_proxy_downstream_handler;
    c->write->handler = ngx_stream_proxy_downstream_handler;
    
    if (rc != NGX_AGAIN) {
        // 连接已完成（Unix域套接字）
        ngx_stream_proxy_init(s);
        return;
    }
    
    // 连接进行中，设置连接处理器
    pc->write->handler = ngx_stream_proxy_connect_handler;
    
    // 添加连接超时定时器
    ngx_add_timer(pc->write, pscf->connect_timeout);
}

static void ngx_stream_proxy_process(ngx_stream_session_t *s,
    ngx_uint_t from_upstream, ngx_uint_t do_write);

static void ngx_stream_proxy_process(ngx_stream_session_t *s,
    ngx_uint_t from_upstream, ngx_uint_t do_write) {
    
    c = s->connection;
    u = s->upstream;
    
    // 确定源和目标连接
    if (from_upstream) {
        src = u->peer.connection;
        dst = c;
        b = &u->upstream_buf;
    } else {
        src = c;
        dst = u->peer.connection;
        b = &u->downstream_buf;
    }
    
    // 读取数据
    while (1) {
        // 缓冲区已满，需要先写出
        if (b->last == b->end && do_write) {
            n = dst->send(dst, b->pos, b->last - b->pos);
            
            if (n == NGX_ERROR) {
                ngx_stream_proxy_finalize(s, NGX_STREAM_OK);
                return;
            }
            
            if (n == NGX_AGAIN) {
                // 目标连接写阻塞，等待可写
                return;
            }
            
            b->pos += n;
            
            if (b->pos == b->last) {
                // 缓冲区已清空
                b->pos = b->start;
                b->last = b->start;
            }
        }
        
        // 读取数据
        size = b->end - b->last;
        
        if (size == 0) {
            // 缓冲区已满，等待写出
            return;
        }
        
        n = src->recv(src, b->last, size);
        
        if (n == NGX_AGAIN) {
            // 源连接读阻塞，等待数据
            return;
        }
        
        if (n == NGX_ERROR || n == 0) {
            // 连接关闭或错误
            ngx_stream_proxy_finalize(s, NGX_STREAM_OK);
            return;
        }
        
        // 累加接收字节数
        if (from_upstream) {
            u->received += n;
        } else {
            s->received += n;
        }
        
        b->last += n;
        
        // 立即写出
        if (do_write && b->pos < b->last) {
            // ... 写入数据
        }
    }
}

static void ngx_stream_proxy_next_upstream(ngx_stream_session_t *s);

static void ngx_stream_proxy_next_upstream(ngx_stream_session_t *s) {
    u = s->upstream;
    pc = u->peer.connection;
    pscf = ngx_stream_get_module_srv_conf(s, ngx_stream_proxy_module);
    
    // 检查重试次数
    if (u->tries == 0 || !pscf->next_upstream) {
        // 达到重试上限或禁用重试
        ngx_stream_proxy_finalize(s, NGX_STREAM_BAD_GATEWAY);
        return;
    }
    
    // 关闭当前后端连接
    if (pc) {
        ngx_close_connection(pc);
        u->peer.connection = NULL;
    }
    
    // 重置状态
    u->tries--;
    
    // 重新连接
    ngx_stream_proxy_connect(s);
}

ngx_int_t ngx_stream_upstream_init_round_robin_peer(ngx_stream_session_t *s,
    ngx_stream_upstream_srv_conf_t *us);

ngx_int_t ngx_stream_upstream_init_round_robin_peer(ngx_stream_session_t *s,
    ngx_stream_upstream_srv_conf_t *us) {
    
    u = s->upstream;
    
    // 创建 Round Robin Peer 数据
    rrp = ngx_pcalloc(s->connection->pool,
                      sizeof(ngx_stream_upstream_rr_peer_data_t));
    
    rrp->peers = us->peer.data;
    rrp->current = ngx_random() % rrp->peers->number;  // 随机起始点
    rrp->tries = rrp->peers->number;
    
    // 设置 Peer 操作函数
    u->peer.get = ngx_stream_upstream_get_round_robin_peer;
    u->peer.free = ngx_stream_upstream_free_round_robin_peer;
    u->peer.data = rrp;
    
    return NGX_OK;
}

ngx_int_t ngx_stream_upstream_get_round_robin_peer(ngx_peer_connection_t *pc,
    void *data);

ngx_int_t ngx_stream_upstream_get_round_robin_peer(ngx_peer_connection_t *pc,
    void *data) {
    
    rrp = data;
    peers = rrp->peers;
    
    // 遍历所有后端服务器
    for (i = 0; i < peers->number; i++) {
        peer = &peers->peer[rrp->current];
        
        // 检查服务器是否可用
        if (!peer->down) {
            // 检查失败次数
            if (peer->max_fails == 0 || peer->fails < peer->max_fails) {
                // 选择此服务器
                pc->sockaddr = peer->sockaddr;
                pc->socklen = peer->socklen;
                pc->name = &peer->name;
                pc->cached = 0;
                
                rrp->current = (rrp->current + 1) % peers->number;
                peer->conns++;  // 增加连接计数
                
                return NGX_OK;
            }
            
            // 检查是否过了失败超时时间
            if (ngx_time() - peer->accessed > peer->fail_timeout) {
                // 重置失败计数
                peer->fails = 0;
            }
        }
        
        rrp->current = (rrp->current + 1) % peers->number;
    }
    
    // 所有服务器不可用
    return NGX_BUSY;
}

ngx_stream_variable_value_t *ngx_stream_get_indexed_variable(
    ngx_stream_session_t *s, ngx_uint_t index);

ngx_stream_variable_value_t *ngx_stream_get_indexed_variable(
    ngx_stream_session_t *s, ngx_uint_t index) {
    
    v = &s->variables[index];
    
    // 变量已缓存
    if (v->not_found || v->valid) {
        return v;
    }
    
    // 调用 get_handler 获取变量值
    cmcf = ngx_stream_get_module_main_conf(s, ngx_stream_core_module);
    var = cmcf->variables.elts;
    
    if (var[index].get_handler(s, v, var[index].data) == NGX_OK) {
        v->valid = 1;
        return v;
    }
    
    v->not_found = 1;
    return v;
}

sequenceDiagram
    autonumber
    participant C as Client
    participant IC as Init Connection
    participant PH as Phase Handler
    participant PROXY as Proxy Handler
    participant UP as Upstream
    participant B as Backend
    
    C->>IC: TCP 连接
    IC->>IC: 创建会话对象
    IC->>PH: 启动阶段处理
    PH->>PH: POST_ACCEPT Phase
    PH->>PH: PREACCESS Phase
    PH->>PH: ACCESS Phase
    PH->>PH: SSL Phase (if needed)
    PH->>PH: PREREAD Phase
    PH->>PROXY: CONTENT Phase
    PROXY->>UP: 初始化 Upstream
    UP->>UP: 选择后端服务器
    UP->>B: 连接后端
    B-->>UP: 连接成功
    UP-->>PROXY: 连接建立
    
    loop 数据转发
        C->>PROXY: 发送数据
        PROXY->>B: 转发到后端
        B->>PROXY: 响应数据
        PROXY->>C: 转发到客户端
    end
    
    C->>PROXY: 关闭连接
    PROXY->>B: 关闭后端连接
    PROXY->>PH: LOG Phase
    PH->>IC: 结束会话

sequenceDiagram
    autonumber
    participant S as Session
    participant P as Proxy
    participant U as Upstream
    participant B1 as Backend 1
    participant B2 as Backend 2
    
    S->>U: 选择后端服务器
    U->>B1: 连接 Backend 1
    B1--xU: 连接失败
    U->>P: NGX_ERROR
    P->>P: 检查重试次数
    P->>U: 请求下一个后端
    U->>U: 标记 Backend 1 为 down
    U->>B2: 连接 Backend 2
    B2-->>U: 连接成功
    U-->>P: 连接建立
    P->>P: 开始数据转发

classDiagram
    class ngx_stream_session_t {
        +uint32_t signature
        +ngx_connection_t* connection
        +off_t received
        +time_t start_sec
        +ngx_msec_t start_msec
        +void** ctx
        +void** main_conf
        +void** srv_conf
        +ngx_stream_upstream_t* upstream
        +ngx_array_t* upstream_states
        +ngx_stream_variable_value_t* variables
        +ngx_int_t phase_handler
        +ngx_uint_t status
        +unsigned ssl:1
        +unsigned stat_processing:1
    }
    
    class ngx_stream_upstream_t {
        +ngx_peer_connection_t peer
        +ngx_buf_t* downstream_buf
        +ngx_buf_t* upstream_buf
        +off_t received
        +time_t start_sec
        +ngx_msec_t start_msec
        +ngx_uint_t responses
        +ngx_uint_t connect_timeout
        +ngx_uint_t timeout
        +unsigned connected:1
        +unsigned proxy_protocol:1
    }
    
    class ngx_stream_phase_handler_t {
        +ngx_stream_phase_handler_pt checker
        +ngx_stream_handler_pt handler
        +ngx_uint_t next
    }
    
    class ngx_stream_phase_engine_t {
        +ngx_stream_phase_handler_t* handlers
        +ngx_uint_t log_phase_handler
    }
    
    class ngx_stream_core_main_conf_t {
        +ngx_array_t servers
        +ngx_stream_phase_engine_t phase_engine
        +ngx_hash_t variables_hash
        +ngx_array_t variables
        +ngx_stream_phase_t phases[7]
    }
    
    class ngx_stream_core_srv_conf_t {
        +ngx_array_t server_names
        +ngx_stream_content_handler_pt handler
        +ngx_stream_conf_ctx_t* ctx
        +ngx_str_t server_name
        +ngx_flag_t tcp_nodelay
        +size_t preread_buffer_size
        +ngx_msec_t preread_timeout
        +ngx_log_t* error_log
        +ngx_resolver_t* resolver
    }
    
    class ngx_peer_connection_t {
        +ngx_connection_t* connection
        +struct sockaddr* sockaddr
        +socklen_t socklen
        +ngx_str_t* name
        +ngx_uint_t tries
        +ngx_event_get_peer_pt get
        +ngx_event_free_peer_pt free
        +void* data
        +unsigned cached:1
    }
    
    ngx_stream_session_t --> ngx_stream_upstream_t
    ngx_stream_session_t --> ngx_stream_core_srv_conf_t
    ngx_stream_upstream_t --> ngx_peer_connection_t
    ngx_stream_core_main_conf_t --> ngx_stream_phase_engine_t
    ngx_stream_phase_engine_t --> ngx_stream_phase_handler_t