feat(adapters): from_jsonschema single-endpoint adapter + shared forwarding core
- src/adapters/forward.rs: shared HTTP forwarding core (extracted from old from_openapi) — HttpServiceConfig/HttpAuthScheme, build_request with capabilities-based credential injection (no env vars), forward + forward_stream (SSE projection), parse_sse_frames - src/adapters/from_jsonschema.rs: FromJsonSchema OperationAdapter (ADR-066) — one registration per call, FromJsonSchema provenance (leaf, Internal default), Sub -> HandlerKind::Stream (text/event-stream) - adapted to alkcall 0.1.1: OperationType::Sub, alkcall::client adapter traits Verified: cargo test (106 lib tests), clippy -D warnings, fmt.
This commit is contained in:
@@ -0,0 +1,442 @@
|
||||
//! Shared HTTP forwarding core for the HTTP-backed adapters
|
||||
//! (`from_openapi`, `from_jsonschema`): request construction
|
||||
//! (path templates, query split, body, credential injection), the
|
||||
//! reqwest forwarding handlers (`forward` / `forward_stream`), and the
|
||||
//! SSE frame parser for streaming (SSE → `ResponseEnvelope`) handlers
|
||||
//! (ADR-049).
|
||||
//!
|
||||
//! The forwarding handler is the no-env-vars credential injection point
|
||||
//! (ADR-014): it reads `OperationContext.capabilities`, never
|
||||
//! `std::env::var`. Imported error codes are `HTTP_<status>` to avoid
|
||||
//! collision with the protocol-level codes (ADR-023).
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
|
||||
use alkcall::registry::context::OperationContext;
|
||||
use alkcall::registry::registration::ResponseStream;
|
||||
use alkcall::registry::spec::OperationType;
|
||||
use futures::stream;
|
||||
use futures::StreamExt;
|
||||
use reqwest::header::{HeaderMap, HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, CONTENT_TYPE};
|
||||
use reqwest::Method;
|
||||
use serde_json::Value;
|
||||
use url::Url;
|
||||
|
||||
use crate::client::SharedHttpClient;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub enum HttpAuthScheme {
|
||||
Bearer,
|
||||
ApiKey { header_name: String },
|
||||
Basic,
|
||||
}
|
||||
|
||||
pub struct HttpServiceConfig {
|
||||
pub namespace: String,
|
||||
pub base_url: String,
|
||||
pub auth: Option<HttpAuthScheme>,
|
||||
pub default_headers: HashMap<String, String>,
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(crate) fn build_request(
|
||||
base_url: &str,
|
||||
path_template: &str,
|
||||
method: &str,
|
||||
auth_scheme: &Option<HttpAuthScheme>,
|
||||
default_headers: &HashMap<String, String>,
|
||||
namespace: &str,
|
||||
input: &Value,
|
||||
context: &OperationContext,
|
||||
) -> Result<(Method, Url, Option<Value>, HeaderMap), CallError> {
|
||||
let input_obj = input.as_object();
|
||||
|
||||
let mut url_path = path_template.to_string();
|
||||
let mut query_params: Vec<(String, String)> = Vec::new();
|
||||
let mut body: Option<Value> = None;
|
||||
|
||||
if let Some(obj) = input_obj {
|
||||
for (key, value) in obj {
|
||||
let placeholder = format!("{{{key}}}");
|
||||
if url_path.contains(&placeholder) {
|
||||
let rendered = value_to_path_segment(value);
|
||||
url_path = url_path.replace(&placeholder, &rendered);
|
||||
} else if key == "body" {
|
||||
body = Some(value.clone());
|
||||
} else {
|
||||
query_params.push((key.clone(), value_to_query(value)));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let base = Url::parse(base_url)
|
||||
.map_err(|e| CallError::internal(format!("invalid base_url `{base_url}`: {e}")))?;
|
||||
let mut url = base
|
||||
.join(url_path.trim_start_matches('/'))
|
||||
.map_err(|e| CallError::internal(format!("invalid path `{url_path}`: {e}")))?;
|
||||
if !query_params.is_empty() {
|
||||
let mut pairs = url.query_pairs_mut();
|
||||
for (k, v) in &query_params {
|
||||
pairs.append_pair(k, v);
|
||||
}
|
||||
}
|
||||
|
||||
let mut headers = HeaderMap::new();
|
||||
for (k, v) in default_headers {
|
||||
if let (Ok(name), Ok(value)) = (
|
||||
HeaderName::try_from(k.as_str()),
|
||||
HeaderValue::try_from(v.as_str()),
|
||||
) {
|
||||
headers.insert(name, value);
|
||||
}
|
||||
}
|
||||
|
||||
if body.is_some() {
|
||||
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
|
||||
}
|
||||
|
||||
if let Some(scheme) = auth_scheme {
|
||||
if let Some(secret) = context.capabilities.get(namespace) {
|
||||
let credential = secret.expose_secret().clone();
|
||||
match scheme {
|
||||
HttpAuthScheme::Bearer => {
|
||||
let header_value = format!("Bearer {credential}");
|
||||
if let Ok(value) = HeaderValue::try_from(header_value) {
|
||||
headers.insert(AUTHORIZATION, value);
|
||||
}
|
||||
}
|
||||
HttpAuthScheme::ApiKey { header_name } => {
|
||||
if let (Ok(name), Ok(value)) = (
|
||||
HeaderName::try_from(header_name.as_str()),
|
||||
HeaderValue::try_from(credential.as_str()),
|
||||
) {
|
||||
headers.insert(name, value);
|
||||
}
|
||||
}
|
||||
HttpAuthScheme::Basic => {
|
||||
let header_value = format!("Basic {credential}");
|
||||
if let Ok(value) = HeaderValue::try_from(header_value) {
|
||||
headers.insert(AUTHORIZATION, value);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let http_method = Method::from_bytes(method.as_bytes())
|
||||
.map_err(|_| CallError::internal(format!("invalid HTTP method `{method}`")))?;
|
||||
Ok((http_method, url, body, headers))
|
||||
}
|
||||
|
||||
pub(crate) fn value_to_path_segment(value: &Value) -> String {
|
||||
match value {
|
||||
Value::String(s) => s.clone(),
|
||||
Value::Number(n) => n.to_string(),
|
||||
Value::Bool(b) => b.to_string(),
|
||||
Value::Null => String::new(),
|
||||
other => other.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn value_to_query(value: &Value) -> String {
|
||||
match value {
|
||||
Value::String(s) => s.clone(),
|
||||
Value::Number(n) => n.to_string(),
|
||||
Value::Bool(b) => b.to_string(),
|
||||
Value::Null => String::new(),
|
||||
other => other.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(crate) async fn forward(
|
||||
http_client: &Arc<SharedHttpClient>,
|
||||
base_url: &str,
|
||||
path_template: &str,
|
||||
method: &str,
|
||||
auth_scheme: &Option<HttpAuthScheme>,
|
||||
default_headers: &HashMap<String, String>,
|
||||
namespace: &str,
|
||||
error_status_codes: &[(u16, String)],
|
||||
op_type: OperationType,
|
||||
input: Value,
|
||||
context: OperationContext,
|
||||
) -> ResponseEnvelope {
|
||||
let request_id = context.request_id.clone();
|
||||
|
||||
let (http_method, url, body, headers) = match build_request(
|
||||
base_url,
|
||||
path_template,
|
||||
method,
|
||||
auth_scheme,
|
||||
default_headers,
|
||||
namespace,
|
||||
&input,
|
||||
&context,
|
||||
) {
|
||||
Ok(parts) => parts,
|
||||
Err(err) => return ResponseEnvelope::error(request_id, err),
|
||||
};
|
||||
|
||||
let http_client = http_client.client();
|
||||
|
||||
let request_builder = http_client
|
||||
.request(http_method, url.as_str())
|
||||
.headers(headers);
|
||||
|
||||
let request_builder = if op_type == OperationType::Sub {
|
||||
request_builder.header(ACCEPT, "text/event-stream")
|
||||
} else {
|
||||
request_builder.header(ACCEPT, "*/*")
|
||||
};
|
||||
|
||||
let request_builder = match body.as_ref() {
|
||||
Some(b) => {
|
||||
let serialized = serde_json::to_string(b).unwrap_or_else(|_| String::from("null"));
|
||||
request_builder.body(serialized)
|
||||
}
|
||||
None => request_builder,
|
||||
};
|
||||
|
||||
let response: reqwest::Response = match request_builder.send().await {
|
||||
Ok(r) => r,
|
||||
Err(err) => {
|
||||
return ResponseEnvelope::error(
|
||||
request_id,
|
||||
CallError::internal(format!("HTTP request failed: {err}")),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
let status = response.status();
|
||||
|
||||
if !status.is_success() {
|
||||
let code = error_status_codes
|
||||
.iter()
|
||||
.find(|(s, _)| *s == status.as_u16())
|
||||
.map(|(_, code)| code.clone())
|
||||
.unwrap_or_else(|| format!("HTTP_{}", status.as_u16()));
|
||||
let message = format!(
|
||||
"HTTP {}: {}",
|
||||
status.as_u16(),
|
||||
status.canonical_reason().unwrap_or("")
|
||||
);
|
||||
return ResponseEnvelope::error(request_id, CallError::new(code, message, false));
|
||||
}
|
||||
|
||||
let content_type = response
|
||||
.headers()
|
||||
.get(reqwest::header::CONTENT_TYPE)
|
||||
.and_then(|v: &reqwest::header::HeaderValue| v.to_str().ok())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
|
||||
if content_type.contains("application/json") {
|
||||
match response.json::<Value>().await {
|
||||
Ok(v) => ResponseEnvelope::ok(request_id, v),
|
||||
Err(err) => ResponseEnvelope::error(
|
||||
request_id,
|
||||
CallError::internal(format!("failed to decode JSON body: {err}")),
|
||||
),
|
||||
}
|
||||
} else if content_type.starts_with("text/") {
|
||||
match response.text().await {
|
||||
Ok(t) => ResponseEnvelope::ok(request_id, Value::String(t)),
|
||||
Err(err) => ResponseEnvelope::error(
|
||||
request_id,
|
||||
CallError::internal(format!("failed to decode text body: {err}")),
|
||||
),
|
||||
}
|
||||
} else {
|
||||
match response.bytes().await {
|
||||
Ok(b) => {
|
||||
let arr: Vec<Value> = b.iter().map(|byte| Value::Number((*byte).into())).collect();
|
||||
ResponseEnvelope::ok(request_id, Value::Array(arr))
|
||||
}
|
||||
Err(err) => ResponseEnvelope::error(
|
||||
request_id,
|
||||
CallError::internal(format!("failed to read body: {err}")),
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(crate) fn forward_stream(
|
||||
http_client: &Arc<SharedHttpClient>,
|
||||
base_url: &str,
|
||||
path_template: &str,
|
||||
method: &str,
|
||||
auth_scheme: &Option<HttpAuthScheme>,
|
||||
default_headers: &HashMap<String, String>,
|
||||
namespace: &str,
|
||||
error_status_codes: &[(u16, String)],
|
||||
input: Value,
|
||||
context: OperationContext,
|
||||
) -> ResponseStream {
|
||||
let request_id = context.request_id.clone();
|
||||
|
||||
let (http_method, url, body, headers) = match build_request(
|
||||
base_url,
|
||||
path_template,
|
||||
method,
|
||||
auth_scheme,
|
||||
default_headers,
|
||||
namespace,
|
||||
&input,
|
||||
&context,
|
||||
) {
|
||||
Ok(parts) => parts,
|
||||
Err(err) => {
|
||||
return Box::pin(stream::once(async move {
|
||||
ResponseEnvelope::error(request_id, err)
|
||||
}));
|
||||
}
|
||||
};
|
||||
|
||||
let http_client = Arc::clone(http_client);
|
||||
let error_status_codes = error_status_codes.to_vec();
|
||||
|
||||
let request_id_stream = request_id.clone();
|
||||
let error_status_codes_stream = error_status_codes.clone();
|
||||
|
||||
let init = async move {
|
||||
let request_builder = http_client
|
||||
.client()
|
||||
.request(http_method, url.as_str())
|
||||
.headers(headers)
|
||||
.header(ACCEPT, "text/event-stream");
|
||||
let request_builder = match body.as_ref() {
|
||||
Some(b) => {
|
||||
let serialized = serde_json::to_string(b).unwrap_or_else(|_| String::from("null"));
|
||||
request_builder.body(serialized)
|
||||
}
|
||||
None => request_builder,
|
||||
};
|
||||
request_builder.send().await
|
||||
};
|
||||
|
||||
let sse = stream::once(init).flat_map(move |result| {
|
||||
let request_id = request_id_stream.clone();
|
||||
let error_status_codes = error_status_codes_stream.clone();
|
||||
match result {
|
||||
Err(err) => Box::pin(stream::once(async move {
|
||||
ResponseEnvelope::error(
|
||||
request_id,
|
||||
CallError::internal(format!("HTTP request failed: {err}")),
|
||||
)
|
||||
})) as ResponseStream,
|
||||
Ok(response) => {
|
||||
let status = response.status();
|
||||
if !status.is_success() {
|
||||
let code = error_status_codes
|
||||
.iter()
|
||||
.find(|(s, _)| *s == status.as_u16())
|
||||
.map(|(_, c)| c.clone())
|
||||
.unwrap_or_else(|| format!("HTTP_{}", status.as_u16()));
|
||||
let message = format!(
|
||||
"HTTP {}: {}",
|
||||
status.as_u16(),
|
||||
status.canonical_reason().unwrap_or("")
|
||||
);
|
||||
Box::pin(stream::once(async move {
|
||||
ResponseEnvelope::error(request_id, CallError::new(code, message, false))
|
||||
})) as ResponseStream
|
||||
} else {
|
||||
let request_id_inner = request_id.clone();
|
||||
Box::pin(
|
||||
stream::unfold(
|
||||
(response.bytes_stream(), String::new()),
|
||||
move |(mut bytes, mut buffer)| {
|
||||
let request_id = request_id_inner.clone();
|
||||
async move {
|
||||
match bytes.next().await {
|
||||
Some(Ok(chunk)) => {
|
||||
buffer.push_str(&String::from_utf8_lossy(&chunk));
|
||||
let (events, remaining) = parse_sse_frames(&buffer);
|
||||
let envelopes: Vec<ResponseEnvelope> = events
|
||||
.into_iter()
|
||||
.map(|e| {
|
||||
let parsed = if e.data.trim().is_empty() {
|
||||
Value::Null
|
||||
} else {
|
||||
serde_json::from_str(&e.data).unwrap_or(
|
||||
Value::String(e.data.clone()),
|
||||
)
|
||||
};
|
||||
ResponseEnvelope::ok(&request_id, parsed)
|
||||
})
|
||||
.collect();
|
||||
Some((envelopes, (bytes, remaining)))
|
||||
}
|
||||
Some(Err(err)) => {
|
||||
let error = CallError::internal(format!(
|
||||
"SSE stream error: {err}"
|
||||
));
|
||||
Some((
|
||||
vec![ResponseEnvelope::error(request_id, error)],
|
||||
(bytes, buffer),
|
||||
))
|
||||
}
|
||||
None => None,
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
.flat_map(stream::iter),
|
||||
) as ResponseStream
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
Box::pin(sse)
|
||||
}
|
||||
|
||||
pub(crate) struct SseEvent {
|
||||
pub(crate) data: String,
|
||||
}
|
||||
|
||||
pub(crate) fn parse_sse_frames(buffer: &str) -> (Vec<SseEvent>, String) {
|
||||
let mut events = Vec::new();
|
||||
let text = if let Some(stripped) = buffer.strip_prefix('\u{feff}') {
|
||||
stripped
|
||||
} else {
|
||||
buffer
|
||||
};
|
||||
let lines: Vec<&str> = text.split('\n').collect();
|
||||
let mut data_buffer: Vec<String> = Vec::new();
|
||||
let mut remaining = String::new();
|
||||
|
||||
for (i, line) in lines.iter().enumerate() {
|
||||
if i == lines.len() - 1 {
|
||||
remaining = line.to_string();
|
||||
break;
|
||||
}
|
||||
let line = line.strip_suffix('\r').unwrap_or(line);
|
||||
if line.is_empty() {
|
||||
if !data_buffer.is_empty() {
|
||||
events.push(SseEvent {
|
||||
data: data_buffer.join("\n"),
|
||||
});
|
||||
}
|
||||
data_buffer.clear();
|
||||
continue;
|
||||
}
|
||||
if line.starts_with(':') {
|
||||
continue;
|
||||
}
|
||||
if let Some((field, value)) = line.split_once(':') {
|
||||
let value = value.strip_prefix(' ').unwrap_or(value);
|
||||
if field == "data" {
|
||||
data_buffer.push(value.to_string());
|
||||
}
|
||||
} else if line == "data" {
|
||||
data_buffer.push(String::new());
|
||||
}
|
||||
}
|
||||
|
||||
(events, remaining)
|
||||
}
|
||||
@@ -0,0 +1,493 @@
|
||||
//! `from_jsonschema` adapter: register a single HTTP-backed operation from a
|
||||
//! caller-supplied [`OperationSpec`], path template, and HTTP method
|
||||
//! ([ADR-066]).
|
||||
//!
|
||||
//! One `HandlerRegistration` per call with a reqwest forwarding handler
|
||||
//! (the shared [`super::forward`] code path with `from_openapi`).
|
||||
//! Provenance is `FromJsonSchema` (leaf, `composition_authority: None`,
|
||||
//! `scoped_env: None`, `Internal` by default — ADR-015/022). `Sub` op
|
||||
//! type → `HandlerKind::Stream` expecting `text/event-stream` (ADR-049).
|
||||
//!
|
||||
//! [ADR-066]: crate::docs
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use alkcall::client::{AdapterError, OperationAdapter};
|
||||
use alkcall::core::types::Capabilities;
|
||||
use alkcall::registry::context::OperationContext;
|
||||
use alkcall::registry::registration::{
|
||||
make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
|
||||
};
|
||||
use alkcall::registry::spec::{OperationSpec, OperationType};
|
||||
use async_trait::async_trait;
|
||||
use serde_json::Value;
|
||||
|
||||
use super::forward::{forward, forward_stream, HttpServiceConfig};
|
||||
use crate::client::SharedHttpClient;
|
||||
|
||||
pub struct FromJsonSchema {
|
||||
spec: OperationSpec,
|
||||
config: HttpServiceConfig,
|
||||
path_template: String,
|
||||
method: String,
|
||||
http_client: Arc<SharedHttpClient>,
|
||||
}
|
||||
|
||||
impl FromJsonSchema {
|
||||
pub fn new(
|
||||
spec: OperationSpec,
|
||||
config: HttpServiceConfig,
|
||||
path_template: String,
|
||||
method: String,
|
||||
http_client: Arc<SharedHttpClient>,
|
||||
) -> Self {
|
||||
Self {
|
||||
spec,
|
||||
config,
|
||||
path_template,
|
||||
method,
|
||||
http_client,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl OperationAdapter for FromJsonSchema {
|
||||
async fn import(&self) -> Result<Vec<HandlerRegistration>, AdapterError> {
|
||||
let path_template = self.path_template.clone();
|
||||
let method_upper = self.method.to_ascii_uppercase();
|
||||
let auth_scheme = self.config.auth.clone();
|
||||
let default_headers = self.config.default_headers.clone();
|
||||
let base_url = self.config.base_url.clone();
|
||||
let namespace = self.config.namespace.clone();
|
||||
let http_client = Arc::clone(&self.http_client);
|
||||
let op_type = self.spec.op_type;
|
||||
|
||||
let error_status_codes: Vec<(u16, String)> = self
|
||||
.spec
|
||||
.error_schemas
|
||||
.iter()
|
||||
.map(|e| (e.http_status.unwrap_or(0), e.code.clone()))
|
||||
.collect();
|
||||
|
||||
let handler = if op_type == OperationType::Sub {
|
||||
let stream_handler =
|
||||
make_streaming_handler(move |input: Value, context: OperationContext| {
|
||||
let path_template = path_template.clone();
|
||||
let method_upper = method_upper.clone();
|
||||
let auth_scheme = auth_scheme.clone();
|
||||
let default_headers = default_headers.clone();
|
||||
let base_url = base_url.clone();
|
||||
let namespace = namespace.clone();
|
||||
let http_client = Arc::clone(&http_client);
|
||||
let error_status_codes = error_status_codes.clone();
|
||||
forward_stream(
|
||||
&http_client,
|
||||
&base_url,
|
||||
&path_template,
|
||||
&method_upper,
|
||||
&auth_scheme,
|
||||
&default_headers,
|
||||
&namespace,
|
||||
&error_status_codes,
|
||||
input,
|
||||
context,
|
||||
)
|
||||
});
|
||||
HandlerKind::Stream(stream_handler)
|
||||
} else {
|
||||
let once_handler = make_handler(move |input: Value, context: OperationContext| {
|
||||
let path_template = path_template.clone();
|
||||
let method_upper = method_upper.clone();
|
||||
let auth_scheme = auth_scheme.clone();
|
||||
let default_headers = default_headers.clone();
|
||||
let base_url = base_url.clone();
|
||||
let namespace = namespace.clone();
|
||||
let http_client = Arc::clone(&http_client);
|
||||
let error_status_codes = error_status_codes.clone();
|
||||
let op_type = op_type;
|
||||
async move {
|
||||
forward(
|
||||
&http_client,
|
||||
&base_url,
|
||||
&path_template,
|
||||
&method_upper,
|
||||
&auth_scheme,
|
||||
&default_headers,
|
||||
&namespace,
|
||||
&error_status_codes,
|
||||
op_type,
|
||||
input,
|
||||
context,
|
||||
)
|
||||
.await
|
||||
}
|
||||
});
|
||||
HandlerKind::Once(once_handler)
|
||||
};
|
||||
|
||||
let capabilities = Capabilities::new();
|
||||
Ok(vec![HandlerRegistration::new(
|
||||
self.spec.clone(),
|
||||
handler,
|
||||
OperationProvenance::FromJsonSchema,
|
||||
None,
|
||||
None,
|
||||
capabilities,
|
||||
)])
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::adapters::forward::{build_request, HttpAuthScheme};
|
||||
use crate::client::HttpClientConfig;
|
||||
use alkcall::protocol::wire::ResponseEnvelope;
|
||||
use alkcall::registry::context::AbortPolicy;
|
||||
use alkcall::registry::env::OperationEnv;
|
||||
use alkcall::registry::spec::{AccessControl, ErrorDefinition, Visibility};
|
||||
use futures::StreamExt;
|
||||
use reqwest::header::AUTHORIZATION;
|
||||
use reqwest::Method;
|
||||
use std::collections::HashMap;
|
||||
use std::time::Duration;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpListener;
|
||||
|
||||
fn noop_context(request_id: &str, capabilities: Capabilities) -> OperationContext {
|
||||
struct NoopEnv;
|
||||
#[async_trait]
|
||||
impl OperationEnv for NoopEnv {
|
||||
async fn invoke_with_policy(
|
||||
&self,
|
||||
_ns: &str,
|
||||
_op: &str,
|
||||
_input: Value,
|
||||
parent: &OperationContext,
|
||||
_policy: AbortPolicy,
|
||||
) -> ResponseEnvelope {
|
||||
ResponseEnvelope::ok(parent.request_id.clone(), Value::Null)
|
||||
}
|
||||
fn contains(&self, _name: &str) -> bool {
|
||||
false
|
||||
}
|
||||
}
|
||||
OperationContext {
|
||||
request_id: request_id.to_string(),
|
||||
parent_request_id: None,
|
||||
identity: None,
|
||||
handler_identity: None,
|
||||
forwarded_for: None,
|
||||
capabilities,
|
||||
metadata: HashMap::new(),
|
||||
scoped_env: alkcall::registry::context::ScopedPeerEnv::empty(),
|
||||
env: Arc::new(NoopEnv),
|
||||
abort_policy: AbortPolicy::default(),
|
||||
deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
|
||||
internal: true,
|
||||
ownership: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn test_spec(name: &str, op_type: OperationType) -> OperationSpec {
|
||||
OperationSpec::new(
|
||||
name,
|
||||
op_type,
|
||||
Visibility::Internal,
|
||||
serde_json::json!({"type":"object","properties":{"id":{"type":"string"}}}),
|
||||
serde_json::json!({"type":"object"}),
|
||||
vec![],
|
||||
AccessControl::default(),
|
||||
None,
|
||||
)
|
||||
}
|
||||
|
||||
fn test_spec_with_errors(
|
||||
name: &str,
|
||||
op_type: OperationType,
|
||||
errors: Vec<ErrorDefinition>,
|
||||
) -> OperationSpec {
|
||||
OperationSpec::new(
|
||||
name,
|
||||
op_type,
|
||||
Visibility::Internal,
|
||||
serde_json::json!({"type":"object"}),
|
||||
serde_json::json!({"type":"object"}),
|
||||
errors,
|
||||
AccessControl::default(),
|
||||
None,
|
||||
)
|
||||
}
|
||||
|
||||
fn test_config(namespace: &str, base_url: &str) -> HttpServiceConfig {
|
||||
HttpServiceConfig {
|
||||
namespace: namespace.to_string(),
|
||||
base_url: base_url.to_string(),
|
||||
auth: None,
|
||||
default_headers: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn test_http_client() -> Arc<SharedHttpClient> {
|
||||
Arc::new(SharedHttpClient::new(HttpClientConfig::default()).unwrap())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn import_produces_one_handler_registration() {
|
||||
let adapter = FromJsonSchema::new(
|
||||
test_spec("svc/getWidget", OperationType::Query),
|
||||
test_config("svc", "https://api.example.com"),
|
||||
"/widgets".to_string(),
|
||||
"GET".to_string(),
|
||||
test_http_client(),
|
||||
);
|
||||
let bundles = adapter.import().await.unwrap();
|
||||
assert_eq!(bundles.len(), 1);
|
||||
assert_eq!(bundles[0].spec.name, "svc/getWidget");
|
||||
assert_eq!(bundles[0].provenance, OperationProvenance::FromJsonSchema);
|
||||
assert!(bundles[0].composition_authority.is_none());
|
||||
assert!(bundles[0].scoped_env.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn query_op_registration_is_handler_kind_once() {
|
||||
let adapter = FromJsonSchema::new(
|
||||
test_spec("svc/getWidget", OperationType::Query),
|
||||
test_config("svc", "https://api.example.com"),
|
||||
"/widgets".to_string(),
|
||||
"GET".to_string(),
|
||||
test_http_client(),
|
||||
);
|
||||
let bundles = adapter.import().await.unwrap();
|
||||
assert!(matches!(bundles[0].handler, HandlerKind::Once(_)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn sub_op_registration_is_handler_kind_stream() {
|
||||
let adapter = FromJsonSchema::new(
|
||||
test_spec("svc/stream", OperationType::Sub),
|
||||
test_config("svc", "https://api.example.com"),
|
||||
"/stream".to_string(),
|
||||
"POST".to_string(),
|
||||
test_http_client(),
|
||||
);
|
||||
let bundles = adapter.import().await.unwrap();
|
||||
assert!(matches!(bundles[0].handler, HandlerKind::Stream(_)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mutation_op_registration_is_handler_kind_once() {
|
||||
let adapter = FromJsonSchema::new(
|
||||
test_spec("svc/createWidget", OperationType::Mutation),
|
||||
test_config("svc", "https://api.example.com"),
|
||||
"/widgets".to_string(),
|
||||
"POST".to_string(),
|
||||
test_http_client(),
|
||||
);
|
||||
let bundles = adapter.import().await.unwrap();
|
||||
assert!(matches!(bundles[0].handler, HandlerKind::Once(_)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_request_injects_bearer_from_capabilities() {
|
||||
let caps = Capabilities::new().with_http_token("github", "tok-123".to_string());
|
||||
let ctx = noop_context("req-1", caps);
|
||||
let (method, url, _body, headers) = build_request(
|
||||
"https://api.github.com",
|
||||
"/repos/{owner}/{repo}/issues",
|
||||
"GET",
|
||||
&Some(HttpAuthScheme::Bearer),
|
||||
&HashMap::new(),
|
||||
"github",
|
||||
&serde_json::json!({"owner":"a","repo":"b"}),
|
||||
&ctx,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(method, Method::GET);
|
||||
assert_eq!(url.path(), "/repos/a/b/issues");
|
||||
assert_eq!(url.host_str(), Some("api.github.com"));
|
||||
let auth = headers.get(AUTHORIZATION).unwrap();
|
||||
assert_eq!(auth.to_str().unwrap(), "Bearer tok-123");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_request_path_and_query_split() {
|
||||
let ctx = noop_context("req-2", Capabilities::new());
|
||||
let (_, url, _, _) = build_request(
|
||||
"https://api.example.com",
|
||||
"/widgets/{id}",
|
||||
"GET",
|
||||
&None,
|
||||
&HashMap::new(),
|
||||
"svc",
|
||||
&serde_json::json!({"id":42,"filter":"active"}),
|
||||
&ctx,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(url.path(), "/widgets/42");
|
||||
assert_eq!(url.query().unwrap(), "filter=active");
|
||||
}
|
||||
|
||||
async fn spawn_echo_server(
|
||||
status: u16,
|
||||
body: &'static str,
|
||||
content_type: &'static str,
|
||||
) -> String {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let (mut sock, _) = match listener.accept().await {
|
||||
Ok(pair) => pair,
|
||||
Err(_) => break,
|
||||
};
|
||||
let status_line = match status {
|
||||
200 => "200 OK",
|
||||
201 => "201 Created",
|
||||
404 => "404 Not Found",
|
||||
500 => "500 Internal Server Error",
|
||||
_ => "200 OK",
|
||||
};
|
||||
let body_bytes = body.as_bytes();
|
||||
let response = format!(
|
||||
"HTTP/1.1 {status_line}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
|
||||
body_bytes.len(),
|
||||
body
|
||||
);
|
||||
let mut buf = [0u8; 4096];
|
||||
let _ = sock.read(&mut buf).await;
|
||||
sock.write_all(response.as_bytes()).await.unwrap();
|
||||
sock.flush().await.unwrap();
|
||||
}
|
||||
});
|
||||
format!("http://{addr}")
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn integration_forwarding_handler_calls_external_endpoint() {
|
||||
let base = spawn_echo_server(200, r#"{"ok":true}"#, "application/json").await;
|
||||
let adapter = FromJsonSchema::new(
|
||||
test_spec("svc/data", OperationType::Query),
|
||||
test_config("svc", &base),
|
||||
"/data".to_string(),
|
||||
"GET".to_string(),
|
||||
test_http_client(),
|
||||
);
|
||||
let bundles = adapter.import().await.unwrap();
|
||||
let registration = &bundles[0];
|
||||
let ctx = noop_context("req-10", Capabilities::new());
|
||||
let response = match ®istration.handler {
|
||||
HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
|
||||
_ => panic!("expected Once handler"),
|
||||
};
|
||||
assert_eq!(response.request_id, "req-10");
|
||||
match response.result {
|
||||
Ok(v) => assert_eq!(v, serde_json::json!({"ok":true})),
|
||||
Err(e) => panic!("expected Ok, got {e:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn integration_non_2xx_returns_declared_error() {
|
||||
let base = spawn_echo_server(404, r#"{"error":"missing"}"#, "application/json").await;
|
||||
let errors = vec![ErrorDefinition {
|
||||
code: "HTTP_404".to_string(),
|
||||
description: "Not found".to_string(),
|
||||
schema: serde_json::json!({"type":"object"}),
|
||||
http_status: Some(404),
|
||||
}];
|
||||
let adapter = FromJsonSchema::new(
|
||||
test_spec_with_errors("svc/missing", OperationType::Query, errors),
|
||||
test_config("svc", &base),
|
||||
"/missing".to_string(),
|
||||
"GET".to_string(),
|
||||
test_http_client(),
|
||||
);
|
||||
let bundles = adapter.import().await.unwrap();
|
||||
let registration = &bundles[0];
|
||||
let ctx = noop_context("req-11", Capabilities::new());
|
||||
let response = match ®istration.handler {
|
||||
HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
|
||||
_ => panic!("expected Once handler"),
|
||||
};
|
||||
match response.result {
|
||||
Err(e) => {
|
||||
assert_eq!(e.code, "HTTP_404");
|
||||
assert!(!e.retryable);
|
||||
}
|
||||
other => panic!("expected HTTP_404 error, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn integration_undeclared_error_status_returns_http_status_code() {
|
||||
let base = spawn_echo_server(500, "boom", "text/plain").await;
|
||||
let adapter = FromJsonSchema::new(
|
||||
test_spec("svc/x", OperationType::Query),
|
||||
test_config("svc", &base),
|
||||
"/x".to_string(),
|
||||
"GET".to_string(),
|
||||
test_http_client(),
|
||||
);
|
||||
let bundles = adapter.import().await.unwrap();
|
||||
let registration = &bundles[0];
|
||||
let ctx = noop_context("req-12", Capabilities::new());
|
||||
let response = match ®istration.handler {
|
||||
HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
|
||||
_ => panic!("expected Once handler"),
|
||||
};
|
||||
match response.result {
|
||||
Err(e) => assert_eq!(e.code, "HTTP_500"),
|
||||
other => panic!("expected HTTP_500, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn integration_sse_subscription_streams_responded_events() {
|
||||
let sse_body = "data: {\"n\":1}\n\ndata: {\"n\":2}\n\n";
|
||||
let base = spawn_echo_server(200, sse_body, "text/event-stream").await;
|
||||
let adapter = FromJsonSchema::new(
|
||||
test_spec("svc/stream", OperationType::Sub),
|
||||
test_config("svc", &base),
|
||||
"/stream".to_string(),
|
||||
"POST".to_string(),
|
||||
test_http_client(),
|
||||
);
|
||||
let bundles = adapter.import().await.unwrap();
|
||||
let registration = &bundles[0];
|
||||
let ctx = noop_context("req-13", Capabilities::new());
|
||||
let stream = match ®istration.handler {
|
||||
HandlerKind::Stream(h) => h(serde_json::json!({}), ctx),
|
||||
_ => panic!("expected Stream handler"),
|
||||
};
|
||||
let collected: Vec<ResponseEnvelope> = stream.collect().await;
|
||||
assert_eq!(collected.len(), 2);
|
||||
assert_eq!(collected[0].result, Ok(serde_json::json!({"n":1})));
|
||||
assert_eq!(collected[1].result, Ok(serde_json::json!({"n":2})));
|
||||
assert_eq!(collected[0].request_id, "req-13");
|
||||
assert_eq!(collected[1].request_id, "req-13");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_env_vars_read_in_build_request() {
|
||||
std::env::set_var("OPENAI_API_KEY", "should-not-be-used");
|
||||
let ctx = noop_context("req-14", Capabilities::new());
|
||||
let (_, _, _, headers) = build_request(
|
||||
"https://api.openai.com",
|
||||
"/v1/chat",
|
||||
"POST",
|
||||
&Some(HttpAuthScheme::Bearer),
|
||||
&HashMap::new(),
|
||||
"openai",
|
||||
&serde_json::json!({"body":{"prompt":"hi"}}),
|
||||
&ctx,
|
||||
)
|
||||
.unwrap();
|
||||
assert!(
|
||||
headers.get(AUTHORIZATION).is_none(),
|
||||
"no auth header when capabilities absent"
|
||||
);
|
||||
std::env::remove_var("OPENAI_API_KEY");
|
||||
}
|
||||
}
|
||||
@@ -1 +1,10 @@
|
||||
//! HTTP-backed call-protocol adapters: `from_openapi` / `from_jsonschema`
|
||||
//! (import external HTTP APIs as operations), `to_openapi` / `to_mcp`
|
||||
//! (project local operations onto HTTP surfaces), `from_mcp`, and
|
||||
//! `from_wss` (feature `wss`).
|
||||
|
||||
pub mod forward;
|
||||
pub mod from_jsonschema;
|
||||
|
||||
pub use forward::{HttpAuthScheme, HttpServiceConfig};
|
||||
pub use from_jsonschema::FromJsonSchema;
|
||||
|
||||
Reference in New Issue
Block a user