修复close阻塞

Author: zhangxin

java/src/main/java/com/baidubce/appbuilder/base/utils/http/HttpClient.java

@@ -5,22 +5,10 @@
 import com.baidubce.appbuilder.base.exception.AppBuilderServerException;
 import com.baidubce.appbuilder.base.utils.iterator.StreamIterator;
 import com.baidubce.appbuilder.base.utils.json.JsonUtils;
-
-import java.io.IOException;
-import java.lang.reflect.Type;
-import java.net.URLEncoder;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import java.util.logging.ConsoleHandler;
-import java.util.logging.Level;
-import java.util.logging.FileHandler;
-import java.util.logging.Logger;
-
+import org.apache.hc.client5.http.classic.methods.HttpDelete;
 import org.apache.hc.client5.http.classic.methods.HttpGet;
 import org.apache.hc.client5.http.classic.methods.HttpPost;
-import org.apache.hc.client5.http.classic.methods.HttpDelete;
+import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
 import org.apache.hc.client5.http.config.RequestConfig;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
@@ -31,6 +19,17 @@
 import org.apache.hc.core5.http.ParseException;
 import org.apache.hc.core5.http.io.entity.EntityUtils;
 
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URLEncoder;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.ConsoleHandler;
+import java.util.logging.FileHandler;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
 public class HttpClient {
     public String SecretKey;
     public String Gateway;
@@ -169,11 +168,11 @@ public ClassicHttpRequest createGetRequestV2(String url, Map<String, Object> map
         return httpGet;
     }
 
-    public ClassicHttpRequest createDeleteRequestV2(String url, Map<String, Object> map) {       
+    public ClassicHttpRequest createDeleteRequestV2(String url, Map<String, Object> map) {
         // 处理查询参数
         String urlParams = toQueryString(map);
         // 如果 URL 已经包含 '?', 不需要再添加
-        String requestURL = GatewayV2 + ConsoleOpenAPIPrefix + ConsoleOpenAPIVersion + url 
+        String requestURL = GatewayV2 + ConsoleOpenAPIPrefix + ConsoleOpenAPIVersion + url
                     + (url.contains("?") ? "&" : "?") + urlParams;
 
         LOGGER.log(Level.FINE, "requestURL: " + requestURL);
@@ -222,10 +221,10 @@ public <T> HttpResponse<T> execute(ClassicHttpRequest request, Type bodyType)
         });
         if (httpResponse.getCode() != 200) {
             String errorMessage = String.format(
-            "Error after processing response with code %d for request ID: %s, message: %s",
-            httpResponse.getCode(),
-            httpResponse.getRequestId(),
-            httpResponse.getMessage()
+                    "Error after processing response with code %d for request ID: %s, message: %s",
+                    httpResponse.getCode(),
+                    httpResponse.getRequestId(),
+                    httpResponse.getMessage()
             );
             LOGGER.log(Level.SEVERE, errorMessage);
             throw new AppBuilderServerException(httpResponse.getRequestId(), httpResponse.getCode(),
@@ -256,7 +255,7 @@ public <T> HttpResponse<StreamIterator<T>> executeSSE(ClassicHttpRequest request
         }
         return new HttpResponse<StreamIterator<T>>().setCode(resp.getCode())
                 .setMessage(resp.getReasonPhrase()).setRequestId(requestId).setHeaders(headers)
-                .setBody(new StreamIterator<>(resp, bodyType));
+                .setBody(new StreamIterator<>(resp, bodyType, (HttpUriRequestBase) request));
     }
 
     private String toQueryString(Map<String, Object> map) {

java/src/main/java/com/baidubce/appbuilder/base/utils/iterator/StreamIterator.java

@@ -1,6 +1,7 @@
 package com.baidubce.appbuilder.base.utils.iterator;
 
 import com.baidubce.appbuilder.base.utils.json.JsonUtils;
+import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
 
 import java.io.BufferedReader;
@@ -14,12 +15,14 @@ public class StreamIterator<T> implements Iterator<T>, AutoCloseable {
     private final CloseableHttpResponse resp;
     private final BufferedReader reader;
     private final Type bodyType;
+    private final HttpUriRequestBase cancellable;
     private String nextLine;
 
-    public StreamIterator(CloseableHttpResponse resp, Type type) throws IOException {
+    public StreamIterator(CloseableHttpResponse resp, Type type, HttpUriRequestBase cancellable) throws IOException {
         this.resp = resp;
         this.reader = new BufferedReader(new InputStreamReader(resp.getEntity().getContent()));
         this.bodyType = type;
+        this.cancellable = cancellable;
     }
 
     @Override
@@ -57,6 +60,9 @@ public T next() {
 
     @Override
     public void close() {
+        if (cancellable != null) {
+            cancellable.abort();
+        }
         try {
             if (reader != null) {
                 reader.close();