use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use helix_core::effect::{HttpRequest, HttpResponse};
use helix_core::ports::HttpRequester;
use helix_core::PortError;
use helix_driver_host::{
AsyncMetricSink, AuthTokenRegistry, CrossCuttingHttp, HostHeaderRegistry, HostNetworkConfig,
HostOtelRuntime, HttpOutcomeObserver, SharedHttpClient,
};
use crate::event_sink::NativeEventSink;
type NativeCrossHttp = CrossCuttingHttp<SharedHttpClient, NativeEventSink>;
static WDIO_FAIL_NEXT_POST_CREATE: AtomicBool = AtomicBool::new(false);
pub fn configure_wdio_fail_next_post_create(armed: bool) {
WDIO_FAIL_NEXT_POST_CREATE.store(armed, Ordering::SeqCst);
}
fn consume_wdio_post_create_fault(req: &HttpRequest) -> bool {
if !req.method.eq_ignore_ascii_case("POST") || !req.url.contains("/posts/create") {
return false;
}
WDIO_FAIL_NEXT_POST_CREATE
.compare_exchange(true, false, Ordering::SeqCst, Ordering::SeqCst)
.is_ok()
}
#[derive(Clone)]
pub struct NativeHttp {
inner: HttpInner,
headers: HostHeaderRegistry,
}
#[derive(Clone)]
enum HttpInner {
Plain(SharedHttpClient),
Cross(NativeCrossHttp),
}
impl NativeHttp {
pub fn with_config(config: HostNetworkConfig) -> Result<Self, PortError> {
SharedHttpClient::new(config).map(Self::from_shared)
}
pub fn with_config_and_metric_sink(
config: HostNetworkConfig,
metrics: Arc<dyn AsyncMetricSink>,
) -> Result<Self, PortError> {
SharedHttpClient::new(config)
.map(|client| client.with_metric_sink(metrics))
.map(Self::from_shared)
}
pub fn from_shared(inner: SharedHttpClient) -> Self {
let headers = inner.headers();
Self {
inner: HttpInner::Plain(inner),
headers,
}
}
pub fn with_crosscut(
client: SharedHttpClient,
event_sink: Arc<NativeEventSink>,
auth: AuthTokenRegistry,
outcome_observer: HttpOutcomeObserver,
) -> Self {
let headers = client.headers();
Self {
inner: HttpInner::Cross(CrossCuttingHttp::new(
client,
event_sink,
auth,
outcome_observer,
)),
headers,
}
}
pub fn with_crosscut_and_otel(
client: SharedHttpClient,
event_sink: Arc<NativeEventSink>,
auth: AuthTokenRegistry,
outcome_observer: HttpOutcomeObserver,
otel: HostOtelRuntime,
) -> Self {
let headers = client.headers();
Self {
inner: HttpInner::Cross(
CrossCuttingHttp::new(client, event_sink, auth, outcome_observer).with_otel(otel),
),
headers,
}
}
pub async fn set_global_header(&self, name: &str, value: &str) -> Result<(), PortError> {
self.headers.set_header(name, value).await
}
pub async fn remove_global_header(&self, name: &str) -> Result<(), PortError> {
self.headers.remove_header(name).await
}
pub fn is_otel_enabled(&self) -> bool {
match &self.inner {
HttpInner::Plain(_) => false,
HttpInner::Cross(inner) => inner.is_otel_enabled(),
}
}
}
#[async_trait::async_trait]
impl HttpRequester for NativeHttp {
async fn request(&self, req: HttpRequest) -> Result<HttpResponse, PortError> {
if consume_wdio_post_create_fault(&req) {
return Err(PortError::Transport(
"wdio coarse-real injected posts/create failure".to_string(),
));
}
match &self.inner {
HttpInner::Plain(c) => c.request(req).await,
HttpInner::Cross(c) => c.request(req).await,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use helix_core::effect::HttpRequest;
use std::time::Duration;
#[tokio::test]
async fn test_http_client_builds_with_external_config() {
let http = NativeHttp::with_config(test_config()).expect("should build client");
assert_eq!(http.receiver_count_for_test(), 0);
}
#[tokio::test]
async fn test_set_and_remove_global_header() {
let http = NativeHttp::with_config(test_config()).expect("build");
http.set_global_header("x-custom", "test-value")
.await
.expect("set header");
http.remove_global_header("x-custom")
.await
.expect("remove header");
}
#[tokio::test]
async fn test_invalid_method_returns_err() {
let http = NativeHttp::with_config(test_config()).expect("build");
let req = HttpRequest {
method: "INVALID METHOD!".to_string(),
url: "/".to_string(),
headers: vec![],
body: None,
};
let result = http.request(req).await;
assert!(result.is_err(), "invalid method should return Err");
}
#[tokio::test]
async fn test_timeout_produces_transport_error() {
let http = NativeHttp::with_config(test_config().with_timeout(Duration::from_millis(10)))
.expect("build");
let req = HttpRequest {
method: "GET".to_string(),
url: "http://10.255.255.1:1/never".to_string(),
headers: vec![],
body: None,
};
let result = http.request(req).await;
assert!(
matches!(result, Err(PortError::Transport(_))),
"timeout/network should produce PortError::Transport, got {:?}",
result
);
}
#[tokio::test]
async fn test_with_crosscut_keeps_header_path_and_transport_error() {
use crate::event_sink::NativeEventSink;
use helix_driver_host::AuthTokenRegistry;
use std::sync::Arc;
let client = SharedHttpClient::new(test_config().with_timeout(Duration::from_millis(10)))
.expect("build client");
let (sink, _rx) = NativeEventSink::new();
let auth = AuthTokenRegistry::new();
auth.set_session("cookieId", "u1");
let http = NativeHttp::with_crosscut(client, Arc::new(sink), auth, |_, _| None);
assert!(!http.is_otel_enabled(), "兼容入口必须保持真 no-op");
http.set_global_header("connectionId", "conn-1")
.await
.expect("set header on crosscut http");
http.remove_global_header("connectionId")
.await
.expect("remove header on crosscut http");
let req = HttpRequest {
method: "GET".to_string(),
url: "http://10.255.255.1:1/never".to_string(),
headers: vec![],
body: None,
};
assert!(matches!(
http.request(req).await,
Err(PortError::Transport(_))
));
}
#[test]
fn explicit_crosscut_constructor_installs_enabled_otel() {
use helix_driver_host::{HostOtelConfig, HostOtelRuntime};
let client = SharedHttpClient::new(test_config()).expect("build client");
let (sink, _rx) = NativeEventSink::new();
let runtime = HostOtelRuntime::new(HostOtelConfig {
enabled: true,
service_name: "native-http-test".to_string(),
endpoint: "noop".to_string(),
protocol: "noop".to_string(),
});
let http = NativeHttp::with_crosscut_and_otel(
client,
Arc::new(sink),
AuthTokenRegistry::new(),
|_, _| None,
runtime,
);
assert!(http.is_otel_enabled());
}
fn test_config() -> HostNetworkConfig {
HostNetworkConfig::new("http://127.0.0.1", "ws://127.0.0.1")
}
}
impl NativeHttp {
#[cfg(test)]
fn receiver_count_for_test(&self) -> usize {
0
}
}