package proxy import ( "bufio" "bytes" "encoding/base64" "encoding/json" "io" "net/http" "time" "github.com/gin-gonic/gin" ) // ProxyHTTPStream 以 NDJSON 流式转发响应(SSE / chunked / 大响应体)。 // 事件序列:meta → chunk* → end | error type streamEvent struct { Type string `json:"type"` Status int `json:"status,omitempty"` Headers map[string]string `json:"headers,omitempty"` Data string `json:"data,omitempty"` Message string `json:"message,omitempty"` TimeMS int64 `json:"timeMs,omitempty"` Size int `json:"size,omitempty"` } func (h *Handler) ProxyHTTPStream(c *gin.Context) { var spec HttpSpec if err := c.ShouldBindJSON(&spec); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid HttpSpec", "detail": err.Error()}) return } method := normalizeMethod(spec.Method) u, err := ValidateURL(spec.URL) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } query := u.Query() for key, value := range spec.Query { query.Set(key, value) } u.RawQuery = query.Encode() var body io.Reader if spec.Body != nil && spec.Body.Mode != "none" && method != http.MethodGet && method != http.MethodHead { body = bytes.NewBufferString(spec.Body.Raw) } req, err := http.NewRequestWithContext(c.Request.Context(), method, u.String(), body) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "cannot create request", "detail": err.Error()}) return } for key, value := range spec.Headers { if !isHopByHop(key) { req.Header.Set(key, value) } } if spec.Body != nil && spec.Body.Mode == "json" && req.Header.Get("Content-Type") == "" { req.Header.Set("Content-Type", "application/json") } client := h.BaseClient if spec.TimeoutMS > 0 { client = &http.Client{Timeout: time.Duration(spec.TimeoutMS) * time.Millisecond} } else { // 流式场景默认不设整体超时(SSE 长连接),交给客户端断开 client = &http.Client{} } started := time.Now() res, err := client.Do(req) if err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": "upstream request failed", "detail": err.Error()}) return } defer res.Body.Close() headers := make(map[string]string, len(res.Header)) for key, values := range res.Header { headers[key] = joinNonEmpty(values) } flusher, ok := c.Writer.(http.Flusher) if !ok { c.JSON(http.StatusInternalServerError, gin.H{"error": "streaming unsupported"}) return } c.Header("Content-Type", "application/x-ndjson") c.Header("Cache-Control", "no-cache") c.Header("X-Accel-Buffering", "no") c.Status(http.StatusOK) enc := json.NewEncoder(c.Writer) writeEvent := func(ev streamEvent) bool { if err := enc.Encode(ev); err != nil { return false } flusher.Flush() return true } headers["content-type"] = res.Header.Get("Content-Type") if !writeEvent(streamEvent{Type: "meta", Status: res.StatusCode, Headers: headers}) { return } reader := bufio.NewReaderSize(res.Body, 32*1024) buf := make([]byte, 16*1024) total := 0 for { n, readErr := reader.Read(buf) if n > 0 { total += n ok := writeEvent(streamEvent{ Type: "chunk", Data: base64.StdEncoding.EncodeToString(buf[:n]), Size: n, }) if !ok { return } } if readErr != nil { break } if c.Request.Context().Err() != nil { return } } writeEvent(streamEvent{Type: "end", TimeMS: time.Since(started).Milliseconds(), Size: total}) } func joinNonEmpty(values []string) string { out := "" for i, v := range values { if i > 0 { out += ", " } out += v } return out }