001/*
002 *  Copyright (c) 2023-2026, Agents-Flex (fuhai999@gmail.com).
003 *  <p>
004 *  Licensed under the Apache License, Version 2.0 (the "License");
005 *  you may not use this file except in compliance with the License.
006 *  You may obtain a copy of the License at
007 *  <p>
008 *  http://www.apache.org/licenses/LICENSE-2.0
009 *  <p>
010 *  Unless required by applicable law or agreed to in writing, software
011 *  distributed under the License is distributed on an "AS IS" BASIS,
012 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 *  See the License for the specific language governing permissions and
014 *  limitations under the License.
015 */
016package com.agentsflex.core.model.client;
017
018import com.agentsflex.core.observability.Observability;
019import com.agentsflex.core.util.IOUtil;
020import io.opentelemetry.api.common.AttributeKey;
021import io.opentelemetry.api.common.Attributes;
022import io.opentelemetry.api.metrics.DoubleHistogram;
023import io.opentelemetry.api.metrics.LongCounter;
024import io.opentelemetry.api.metrics.Meter;
025import io.opentelemetry.api.trace.Span;
026import io.opentelemetry.api.trace.StatusCode;
027import io.opentelemetry.api.trace.Tracer;
028import io.opentelemetry.context.Scope;
029import okhttp3.*;
030import okio.BufferedSink;
031import org.slf4j.Logger;
032import org.slf4j.LoggerFactory;
033
034import java.io.File;
035import java.io.IOException;
036import java.io.InputStream;
037import java.net.URI;
038import java.net.URISyntaxException;
039import java.util.Map;
040
041public class HttpClient {
042    private static final Logger LOG = LoggerFactory.getLogger(HttpClient.class);
043    private static final MediaType JSON_TYPE = MediaType.parse("application/json; charset=utf-8");
044
045    // ===== Observability Components =====
046    private static final Tracer TRACER = Observability.getTracer();
047    private static final Meter METER = Observability.getMeter();
048
049    private static final LongCounter HTTP_REQUEST_COUNT = METER.counterBuilder("http.client.request.count")
050        .setDescription("Total number of HTTP client requests")
051        .build();
052
053    private static final DoubleHistogram HTTP_LATENCY_HISTOGRAM = METER.histogramBuilder("http.client.request.duration")
054        .setDescription("HTTP client request duration in seconds")
055        .setUnit("s")
056        .build();
057
058    private static final LongCounter HTTP_ERROR_COUNT = METER.counterBuilder("http.client.request.error.count")
059        .setDescription("Total number of HTTP client request errors")
060        .build();
061
062    private final OkHttpClient okHttpClient;
063
064    public HttpClient() {
065        this(OkHttpClientUtil.buildDefaultClient());
066    }
067
068    public HttpClient(OkHttpClient okHttpClient) {
069        this.okHttpClient = okHttpClient;
070    }
071
072
073    public String get(String url) {
074        return tracedCall(url, "GET", null, null, this::executeString);
075    }
076
077    public byte[] getBytes(String url) {
078        return tracedCall(url, "GET", null, null, this::executeBytes);
079    }
080
081    public String get(String url, Map<String, String> headers) {
082        return tracedCall(url, "GET", headers, null, this::executeString);
083    }
084
085    public Response getResponse(String url, Map<String, String> headers) {
086        return tracedCall(url, "GET", headers, null, this::executeResponse);
087    }
088
089    public String post(String url, Map<String, String> headers, String payload) {
090        return tracedCall(url, "POST", headers, payload, this::executeString);
091    }
092
093    public byte[] postBytes(String url, Map<String, String> headers, String payload) {
094        return tracedCall(url, "POST", headers, payload, this::executeBytes);
095    }
096
097    public String put(String url, Map<String, String> headers, String payload) {
098        return tracedCall(url, "PUT", headers, payload, this::executeString);
099    }
100
101    public String delete(String url, Map<String, String> headers, String payload) {
102        return tracedCall(url, "DELETE", headers, payload, this::executeString);
103    }
104
105    public String multipartString(String url, Map<String, String> headers, Map<String, Object> payload) {
106        return tracedCall(url, "POST", headers, payload, (u, m, h, p, s) -> {
107            //noinspection unchecked
108            try (Response response = multipart(u, h, (Map<String, Object>) p);
109                 ResponseBody body = response.body()) {
110                if (body != null) {
111                    return body.string();
112                }
113            } catch (IOException ioe) {
114                LOG.error("HTTP multipartString failed: " + u, ioe);
115                s.setStatus(StatusCode.ERROR, ioe.getMessage());
116                s.recordException(ioe);
117            } catch (Exception e) {
118                LOG.error(e.toString(), e);
119                throw e;
120            }
121            return null;
122        });
123    }
124
125    public byte[] multipartBytes(String url, Map<String, String> headers, Map<String, Object> payload) {
126        return tracedCall(url, "POST", headers, payload, (u, m, h, p, s) -> {
127            //noinspection unchecked
128            try (Response response = multipart(u, h, (Map<String, Object>) p);
129                 ResponseBody body = response.body()) {
130                if (body != null) {
131                    return body.bytes();
132                }
133            } catch (IOException ioe) {
134                LOG.error("HTTP multipartBytes failed: " + u, ioe);
135                s.setStatus(StatusCode.ERROR, ioe.getMessage());
136                s.recordException(ioe);
137            } catch (Exception e) {
138                LOG.error(e.toString(), e);
139                throw e;
140            }
141            return null;
142        });
143    }
144
145    // ===== Internal execution methods =====
146
147    public String executeString(String url, String method, Map<String, String> headers, Object payload, Span span) {
148        try (Response response = execute0(url, method, headers, payload);
149             ResponseBody body = response.body()) {
150            if (body != null) {
151                return body.string();
152            }
153        } catch (IOException ioe) {
154            LOG.error("HTTP executeString failed: " + url, ioe);
155            span.setStatus(StatusCode.ERROR, ioe.getMessage());
156            span.recordException(ioe);
157        } catch (Exception e) {
158            LOG.error(e.toString(), e);
159            throw e;
160        }
161        return null;
162    }
163
164    public byte[] executeBytes(String url, String method, Map<String, String> headers, Object payload, Span span) {
165        try (Response response = execute0(url, method, headers, payload);
166             ResponseBody body = response.body()) {
167            if (body != null) {
168                return body.bytes();
169            }
170        } catch (IOException ioe) {
171            LOG.error("HTTP executeBytes failed: " + url, ioe);
172            span.setStatus(StatusCode.ERROR, ioe.getMessage());
173            span.recordException(ioe);
174        } catch (Exception e) {
175            LOG.error(e.toString(), e);
176            throw e;
177        }
178        return null;
179    }
180
181    public Response executeResponse(
182        String url, String method, Map<String, String> headers, Object payload, Span span) {
183        try (Response response = execute0(url, method, headers, payload)) {
184            if (response != null) {
185                return response;
186            }
187        } catch (IOException ioe) {
188            LOG.error("HTTP executeString failed: " + url, ioe);
189            span.setStatus(StatusCode.ERROR, ioe.getMessage());
190            span.recordException(ioe);
191        } catch (Exception e) {
192            LOG.error(e.toString(), e);
193            throw e;
194        }
195        return null;
196    }
197
198    private Response execute0(String url, String method, Map<String, String> headers, Object payload) throws IOException {
199        Request.Builder builder = new Request.Builder().url(url);
200        if (headers != null && !headers.isEmpty()) {
201            headers.forEach((key, value) -> {
202                if (key != null && value != null) {
203                    builder.addHeader(key, value);
204                }
205            });
206        }
207
208        Request request;
209        if ("GET".equalsIgnoreCase(method)) {
210            request = builder.build();
211        } else {
212            RequestBody body = RequestBody.create(payload == null ? "" : payload.toString(), JSON_TYPE);
213            request = builder.method(method, body).build();
214        }
215
216        Response response = okHttpClient.newCall(request).execute();
217
218        // Inject status code into current span
219        injectStatusCodeToCurrentSpan(response);
220        return response;
221    }
222
223    public Response multipart(String url, Map<String, String> headers, Map<String, Object> payload) throws IOException {
224        Request.Builder builder = new Request.Builder().url(url);
225        if (headers != null && !headers.isEmpty()) {
226            headers.forEach(builder::addHeader);
227        }
228
229        MultipartBody.Builder mbBuilder = new MultipartBody.Builder().setType(MultipartBody.FORM);
230        payload.forEach((key, value) -> {
231            if (value instanceof File) {
232                File file = (File) value;
233                RequestBody body = RequestBody.create(file, MediaType.parse("application/octet-stream"));
234                mbBuilder.addFormDataPart(key, file.getName(), body);
235            } else if (value instanceof InputStream) {
236                RequestBody body = new InputStreamRequestBody(MediaType.parse("application/octet-stream"), (InputStream) value);
237                mbBuilder.addFormDataPart(key, key, body);
238            } else if (value instanceof byte[]) {
239                mbBuilder.addFormDataPart(key, key, RequestBody.create((byte[]) value));
240            } else {
241                mbBuilder.addFormDataPart(key, String.valueOf(value));
242            }
243        });
244
245        MultipartBody multipartBody = mbBuilder.build();
246        Request request = builder.post(multipartBody).build();
247        Response response = okHttpClient.newCall(request).execute();
248
249        // Inject status code into current span (same as execute0)
250        injectStatusCodeToCurrentSpan(response);
251        return response;
252    }
253
254    // ===== Shared helper for span status code injection =====
255
256    private void injectStatusCodeToCurrentSpan(Response response) {
257        Span currentSpan = Span.current();
258        if (currentSpan != null && currentSpan != Span.getInvalid()) {
259            int statusCode = response.code();
260            currentSpan.setAttribute("http.status_code", statusCode);
261            if (statusCode >= 400) {
262                currentSpan.setStatus(StatusCode.ERROR, "HTTP " + statusCode);
263            }
264        }
265    }
266
267    // ===== Observability wrapper =====
268    @FunctionalInterface
269    private interface HttpClientCall<T> {
270        T call(String url, String method, Map<String, String> headers, Object payload, Span span) throws Exception;
271    }
272
273    private <T> T tracedCall(String url, String method, Map<String, String> headers, Object payload, HttpClientCall<T> call) {
274        String host = extractHost(url);
275        Span span = TRACER.spanBuilder("http.client.request")
276            .setAttribute("http.method", method)
277            .setAttribute("http.url", url)
278            .setAttribute("server.address", host)
279            .startSpan();
280
281        long startTime = System.nanoTime();
282        boolean success = true;
283
284        try (Scope scope = span.makeCurrent()) {
285            return call.call(url, method, headers, payload, span);
286        } catch (Exception e) {
287            success = false;
288            span.setStatus(StatusCode.ERROR, e.getMessage());
289            span.recordException(e);
290            throw new RuntimeException("HTTP request failed", e);
291        } finally {
292            span.end();
293            double latency = (System.nanoTime() - startTime) / 1_000_000_000.0;
294            Attributes attrs = Attributes.of(
295                AttributeKey.stringKey("http.method"), method,
296                AttributeKey.stringKey("server.address"), host,
297                AttributeKey.stringKey("http.success"), String.valueOf(success)
298            );
299            HTTP_REQUEST_COUNT.add(1, attrs);
300            HTTP_LATENCY_HISTOGRAM.record(latency, attrs);
301            if (!success) {
302                HTTP_ERROR_COUNT.add(1, attrs);
303            }
304        }
305    }
306
307    // ===== Utility =====
308    private static String extractHost(String url) {
309        try {
310            URI uri = new URI(url);
311            String host = uri.getHost();
312            int port = uri.getPort();
313            if (port != -1) {
314                return host + ":" + port;
315            }
316            return host;
317        } catch (URISyntaxException e) {
318            return "unknown";
319        }
320    }
321
322    // ===== Inner class =====
323
324    public static class InputStreamRequestBody extends RequestBody {
325        private final InputStream inputStream;
326        private final MediaType contentType;
327
328        public InputStreamRequestBody(MediaType contentType, InputStream inputStream) {
329            if (inputStream == null) throw new NullPointerException("inputStream == null");
330            this.contentType = contentType;
331            this.inputStream = inputStream;
332        }
333
334        @Override
335        public MediaType contentType() {
336            return contentType;
337        }
338
339        @Override
340        public long contentLength() throws IOException {
341            return inputStream.available() == 0 ? -1 : inputStream.available();
342        }
343
344        @Override
345        public void writeTo(BufferedSink sink) throws IOException {
346            IOUtil.copy(inputStream, sink);
347        }
348    }
349}