diff --git a/src/main/java/com/yunhai/platform/gateway/forward/StreamingForwarder.java b/src/main/java/com/yunhai/platform/gateway/forward/StreamingForwarder.java new file mode 100644 index 0000000..ade810c --- /dev/null +++ b/src/main/java/com/yunhai/platform/gateway/forward/StreamingForwarder.java @@ -0,0 +1,257 @@ +package com.yunhai.platform.gateway.forward; + +import com.yunhai.platform.gateway.envelope.EnvelopeWriter; +import org.apache.http.HttpEntity; +import org.apache.http.Header; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpUriRequest; +import org.apache.http.client.methods.RequestBuilder; +import org.apache.http.entity.InputStreamEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; + +import javax.annotation.PreDestroy; +import javax.servlet.AsyncContext; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.SocketTimeoutException; +import java.net.URI; +import java.util.Collections; +import java.util.Enumeration; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +/** + * 流式转发器（T1.3）：网关 → 提供方的 HTTP 转发，请求/响应 body 均流式透传 + * （不聚合、不落盘、不改写响应体），经异步 Servlet（{@link AsyncContext}）+ Apache HttpClient 执行， + * 请求线程不阻塞等待后端。 + * + */ +@Component +public class StreamingForwarder { + + /** 读超时默认 30s（route.timeout_ms 未配时的统一默认） */ + public static final int DEFAULT_READ_TIMEOUT_MS = 30_000; + /** body 上限默认 10MB（route.request_body_limit_mb 未配时的统一默认） */ + public static final long DEFAULT_BODY_LIMIT_BYTES = 10L * 1024 * 1024; + /** 连接建立超时（容器网内建连应极快；非卡参数，防御挂死的 SYN） */ + private static final int CONNECT_TIMEOUT_MS = 5_000; + /** 身份头：转发时剥除客户端伪造值，注入权属网关（T1.4） */ + public static final String HEADER_CALLER_APP_KEY = "X-Caller-App-Key"; + + private static final Logger log = LoggerFactory.getLogger(StreamingForwarder.class); + + /** RFC 2616 hop-by-hop + 由客户端库重算的头 */ + private static final Set REQ_SKIP; + private static final Set RES_SKIP; + + static { + Set s = new HashSet<>(); + Collections.addAll(s, "connection", "keep-alive", "proxy-connection", "proxy-authenticate", + "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade", + "content-length", "host", HEADER_CALLER_APP_KEY.toLowerCase()); + REQ_SKIP = s; + Set r = new HashSet<>(); + Collections.addAll(r, "connection", "keep-alive", "proxy-authenticate", "proxy-authorization", + "te", "trailer", "transfer-encoding", "upgrade"); + RES_SKIP = r; + } + + private final CloseableHttpClient client; + private final Executor executor; + private final ExecutorService ownedPool; + + /** Spring 装配：默认连接池 client + 异步执行线程池（IO 密集，缓存池） */ + public StreamingForwarder() { + this.client = HttpClients.custom() + .disableRedirectHandling() + .setMaxConnTotal(400) + .setMaxConnPerRoute(200) + .build(); + this.ownedPool = Executors.newCachedThreadPool(r -> { + Thread t = new Thread(r, "gw-forward"); + t.setDaemon(true); + return t; + }); + this.executor = ownedPool; + } + + /** 测试/显式装配入口（同 TargetRouter.setProperties 惯例）：注入 client 与执行器 */ + StreamingForwarder(CloseableHttpClient client, Executor executor) { + this.client = client; + this.executor = executor; + this.ownedPool = null; + } + + @PreDestroy + public void shutdown() throws IOException { + client.close(); + if (ownedPool != null) { + ownedPool.shutdown(); + try { + ownedPool.awaitTermination(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + + /** + * 转发并原样回写。声明长度超限在建连之前拦（裸 413）；否则转异步： + * 工作线程流式收发，完成后 {@link AsyncContext#complete()}。 + */ + public void forward(HttpServletRequest req, HttpServletResponse res, String forwardUrl, + int readTimeoutMs, long bodyLimitBytes) throws IOException { + if (req.getContentLengthLong() > bodyLimitBytes) { + EnvelopeWriter.writeBare(res, 413); + return; + } + final AsyncContext ctx = req.startAsync(req, res); + executor.execute(() -> { + try { + doForward(req, res, forwardUrl, readTimeoutMs, bodyLimitBytes); + } catch (BodyTooLargeException e) { + writeBareGuard(res, 413); // 未声明长度的 chunked 请求在流式读取中超限 + } catch (SocketTimeoutException e) { + writeBareGuard(res, 504); + } catch (IOException e) { + if (hasBodyTooLargeCause(e)) { + writeBareGuard(res, 413); + } else { + log.warn("[GatewayForward] backend io failure url={}", forwardUrl, e); + writeBareGuard(res, 502); // 连接失败/协议断流 + } + } catch (RuntimeException e) { + log.error("[GatewayForward] unexpected forward error url={}", forwardUrl, e); + writeBareGuard(res, 502); + } finally { + ctx.complete(); + } + }); + } + + private void doForward(HttpServletRequest req, HttpServletResponse res, String forwardUrl, + int readTimeoutMs, long bodyLimitBytes) throws IOException { + RequestBuilder rb = RequestBuilder.create(req.getMethod()) + .setUri(URI.create(forwardUrl)) + .setConfig(RequestConfig.custom() + .setConnectTimeout(CONNECT_TIMEOUT_MS) + .setSocketTimeout(readTimeoutMs) + .setConnectionRequestTimeout(CONNECT_TIMEOUT_MS) + .setRedirectsEnabled(false) + .build()); + Enumeration names = req.getHeaderNames(); + while (names != null && names.hasMoreElements()) { + String name = names.nextElement(); + if (REQ_SKIP.contains(name.toLowerCase())) { + continue; + } + Enumeration values = req.getHeaders(name); + while (values.hasMoreElements()) { + rb.addHeader(name, values.nextElement()); + } + } + String method = req.getMethod().toUpperCase(); + if (!"GET".equals(method) && !"HEAD".equals(method) + && !"OPTIONS".equals(method) && !"TRACE".equals(method)) { + // 长度未知 → chunked 交由 HttpClient 重算；流式读取、超上限即断 + rb.setEntity(new InputStreamEntity(new LimitedInputStream(req.getInputStream(), bodyLimitBytes), -1)); + } + HttpUriRequest request = rb.build(); + try (CloseableHttpResponse up = client.execute(request)) { + res.setStatus(up.getStatusLine().getStatusCode()); + for (Header h : up.getAllHeaders()) { + if (!RES_SKIP.contains(h.getName().toLowerCase())) { + res.addHeader(h.getName(), h.getValue()); + } + } + HttpEntity entity = up.getEntity(); + if (entity != null) { + entity.writeTo(res.getOutputStream()); + } + } + } + + private static void writeBareGuard(HttpServletResponse res, int status) { + try { + if (!res.isCommitted()) { + EnvelopeWriter.writeBare(res, status); + } + } catch (IOException ignored) { + // 裸码写出失败无从补救（响应面已坏） + } + } + + private static boolean hasBodyTooLargeCause(Throwable e) { + for (Throwable t = e; t != null; t = t.getCause()) { + if (t instanceof BodyTooLargeException) { + return true; + } + } + return false; + } + + /** 流式 body 上限哨兵异常 */ + static class BodyTooLargeException extends IOException { + BodyTooLargeException() { + super("request body exceeds limit"); + } + } + + /** 计数透传流：超过 limit 立即抛哨兵异常（绝不缓冲整体） */ + private static final class LimitedInputStream extends FilterInputStream { + private final long limit; + private long count; + + LimitedInputStream(InputStream in, long limit) { + super(in); + this.limit = limit; + } + + private void counted(long n) throws BodyTooLargeException { + if (n > 0) { + count += n; + if (count > limit) { + throw new BodyTooLargeException(); + } + } + } + + @Override + public int read() throws IOException { + int b = super.read(); + if (b >= 0) { + counted(1); + } + return b; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + int n = super.read(b, off, len); + counted(n); + return n; + } + } +} diff --git a/src/test/java/com/yunhai/platform/gateway/forward/StreamingForwarderTest.java b/src/test/java/com/yunhai/platform/gateway/forward/StreamingForwarderTest.java new file mode 100644 index 0000000..9e26e5c --- /dev/null +++ b/src/test/java/com/yunhai/platform/gateway/forward/StreamingForwarderTest.java @@ -0,0 +1,228 @@ +package com.yunhai.platform.gateway.forward; + +import com.sun.net.httpserver.HttpServer; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.mock.web.MockHttpServletRequest; +import org.springframework.mock.web.MockHttpServletResponse; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.ServerSocket; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * StreamingForwarder 五用例（T1.3 TDD，假后端 = JDK 自带 com.sun.net.httpserver.HttpServer， + * 零新增测试依赖）：GET 转发 forwardUrl 逐字符 + 响应透传；POST body 流式透传； + * 客户端伪造 X-Caller-App-Key 不出现在转发请求；签名头四件原样到达； + * 超时裸 504 / 连接拒绝裸 502（另附 body 超限裸 413——实现要点契约）。 + */ +class StreamingForwarderTest { + + private static HttpServer server; + private static int port; + + /** 假后端最近一次收到的请求快照 */ + private static volatile Map lastHeaders; + private static volatile String lastMethod; + private static volatile String lastUri; + private static volatile byte[] lastBody; + /** 假后端行为控制 */ + private static volatile int respStatus = 201; + private static volatile String respHeader = "X-Upstream:v1"; // name:value + private static volatile byte[] respBody = "hello-gateway".getBytes(StandardCharsets.UTF_8); + private static volatile long respSleepMs = 0; + private static volatile int hitCount = 0; + private static volatile boolean echoBody = false; + + @BeforeAll + static void startFakeBackend() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.setExecutor(Executors.newFixedThreadPool(4)); + server.createContext("/", exchange -> { + hitCount++; + lastMethod = exchange.getRequestMethod(); + lastUri = exchange.getRequestURI().toString(); + lastHeaders = new ConcurrentHashMap<>(); + for (Map.Entry> e : exchange.getRequestHeaders().entrySet()) { + lastHeaders.put(e.getKey().toLowerCase(), String.join(",", e.getValue())); + } + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + byte[] buf = new byte[8192]; + int n; + while ((n = exchange.getRequestBody().read(buf)) > 0) { + bos.write(buf, 0, n); + } + lastBody = bos.toByteArray(); + if (respSleepMs > 0) { + try { + Thread.sleep(respSleepMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + } + byte[] body = echoBody ? lastBody : respBody; + if (respHeader != null) { + String[] kv = respHeader.split(":", 2); + exchange.getResponseHeaders().add(kv[0], kv[1]); + } + exchange.sendResponseHeaders(respStatus, body.length == 0 ? -1 : body.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(body); + } + exchange.close(); + }); + server.start(); + port = server.getAddress().getPort(); + } + + @AfterAll + static void stopFakeBackend() { + server.stop(0); + } + + @BeforeEach + @AfterEach + void resetFakeBackend() { + respStatus = 201; + respHeader = "X-Upstream:v1"; + respBody = "hello-gateway".getBytes(StandardCharsets.UTF_8); + respSleepMs = 0; + hitCount = 0; + echoBody = false; + lastHeaders = null; + lastMethod = null; + lastUri = null; + lastBody = null; + } + + /** 同线程执行器：转发逻辑同步完成，测试无需起线程等待 */ + private StreamingForwarder forwarder() { + CloseableHttpClient client = HttpClients.custom().disableRedirectHandling().build(); + return new StreamingForwarder(client, Runnable::run); + } + + private MockHttpServletRequest request(String method, String uri) { + MockHttpServletRequest req = new MockHttpServletRequest(method, uri); + req.setAsyncSupported(true); // 异步 Servlet 契约（T1.6 链上 filter 需 async-supported） + return req; + } + + @Test + void GET转发_forwardUrl逐字符且响应原样透传() throws Exception { + MockHttpServletRequest req = request("GET", "/userbase/oapi/x"); + req.setQueryString("a=1&b=2"); + MockHttpServletResponse res = new MockHttpServletResponse(); + String forwardUrl = "http://127.0.0.1:" + port + "/userbase/oapi/x?a=1&b=2"; + + forwarder().forward(req, res, forwardUrl, 5000, 1024 * 1024); + + assertThat(lastUri).as("换主机不换路径：URI+query 逐字符送达").isEqualTo("/userbase/oapi/x?a=1&b=2"); + assertThat(lastMethod).isEqualTo("GET"); + assertThat(res.getStatus()).as("响应 status 透传").isEqualTo(201); + assertThat(res.getHeader("X-Upstream")).as("响应 header 透传").isEqualTo("v1"); + assertThat(res.getContentAsString()).as("响应 body 原样透传").isEqualTo("hello-gateway"); + } + + @Test + void POST_body流式透传() throws Exception { + echoBody = true; + respStatus = 200; + byte[] payload = new byte[100 * 1024]; + for (int i = 0; i < payload.length; i++) { + payload[i] = (byte) (i % 251); + } + MockHttpServletRequest req = request("POST", "/usermsg/oapi/msg/sendUserMsg"); + req.setContent(payload); + MockHttpServletResponse res = new MockHttpServletResponse(); + + forwarder().forward(req, res, "http://127.0.0.1:" + port + "/usermsg/oapi/msg/sendUserMsg", + 5000, 4 * 1024 * 1024); + + assertThat(lastBody).as("请求 body 完整到达假后端").isEqualTo(payload); + assertThat(res.getContentAsByteArray()).as("响应 body 回显一致（不改写）").isEqualTo(payload); + } + + @Test + void 客户端伪造XCallerAppKey不出现在转发请求() throws Exception { + MockHttpServletRequest req = request("GET", "/userbase/oapi/x"); + req.addHeader("X-Caller-App-Key", "hack"); + req.addHeader("Connection", "close"); // hop-by-hop 也不得透传 + MockHttpServletResponse res = new MockHttpServletResponse(); + + forwarder().forward(req, res, "http://127.0.0.1:" + port + "/userbase/oapi/x", 5000, 1024 * 1024); + + assertThat(res.getStatus()).isEqualTo(201); + assertThat(lastHeaders).as("身份头唯一合法来源是网关（T1.4）").doesNotContainKey("x-caller-app-key"); + } + + @Test + void 签名头四件原样到达假后端() throws Exception { + MockHttpServletRequest req = request("GET", "/userbase/oapi/x"); + req.addHeader("X-App-Key", "100617"); + req.addHeader("X-Timestamp", "1758441600000"); + req.addHeader("X-Signature", "Regw08O602Tp61nhH6oG/5VIvyfmU9ZbWs42P7OHzj4="); + req.addHeader("X-Nonce", "c7ce1f3e-9a1d-4b3e-8f2a-0b1c2d3e4f5a"); + MockHttpServletResponse res = new MockHttpServletResponse(); + + forwarder().forward(req, res, "http://127.0.0.1:" + port + "/userbase/oapi/x", 5000, 1024 * 1024); + + assertThat(lastHeaders.get("x-app-key")).isEqualTo("100617"); + assertThat(lastHeaders.get("x-timestamp")).isEqualTo("1758441600000"); + assertThat(lastHeaders.get("x-signature")).isEqualTo("Regw08O602Tp61nhH6oG/5VIvyfmU9ZbWs42P7OHzj4="); + assertThat(lastHeaders.get("x-nonce")).as("R3：UserBase 自验签依赖").isEqualTo("c7ce1f3e-9a1d-4b3e-8f2a-0b1c2d3e4f5a"); + } + + @Test + void 假后端超时_裸504() throws Exception { + respSleepMs = 600; + MockHttpServletRequest req = request("GET", "/userbase/oapi/slow"); + MockHttpServletResponse res = new MockHttpServletResponse(); + + forwarder().forward(req, res, "http://127.0.0.1:" + port + "/userbase/oapi/slow", 100, 1024 * 1024); + + assertThat(res.getStatus()).isEqualTo(504); + assertThat(res.getContentAsByteArray()).as("裸码无信封").isEmpty(); + } + + @Test + void 连接拒绝_裸502() throws Exception { + int deadPort; + try (ServerSocket ss = new ServerSocket(0)) { + deadPort = ss.getLocalPort(); + } + MockHttpServletRequest req = request("GET", "/userbase/oapi/x"); + MockHttpServletResponse res = new MockHttpServletResponse(); + + forwarder().forward(req, res, "http://127.0.0.1:" + deadPort + "/userbase/oapi/x", 5000, 1024 * 1024); + + assertThat(res.getStatus()).isEqualTo(502); + assertThat(res.getContentAsByteArray()).isEmpty(); + } + + @Test + void body超限_裸413且不连后端() throws Exception { + MockHttpServletRequest req = request("POST", "/userbase/oapi/x"); + req.setContent("0123456789".getBytes(StandardCharsets.UTF_8)); // 10B > 4B + MockHttpServletResponse res = new MockHttpServletResponse(); + + forwarder().forward(req, res, "http://127.0.0.1:" + port + "/userbase/oapi/x", 5000, 4); + + assertThat(res.getStatus()).isEqualTo(413); + assertThat(res.getContentAsByteArray()).as("裸码").isEmpty(); + assertThat(hitCount).as("上限拦截在连接之前").isZero(); + } +}