bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

community: remove premium protocols and streaming surface

Browse Source
This commit is contained in:
a.tolmachev
2026-05-07 20:44:58 +00:00
parent 202c886793
commit 29f971c22d
78 changed files with 21 additions and 8838 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/crank-adapter-graphql/Cargo.toml
-18
View File
@@ -1,18 +0,0 @@
[package]
name = "crank-adapter-graphql"
edition.workspace = true
license.workspace = true
rust-version.workspace = true
version.workspace = true
[dependencies]
crank-core = { path = "../crank-core" }
crank-mapping = { path = "../crank-mapping" }
reqwest.workspace = true
serde.workspace = true
serde_json.workspace = true
thiserror.workspace = true
[dev-dependencies]
axum.workspace = true
tokio.workspace = true
crates/crank-adapter-graphql/src/client.rs
-264
View File
@@ -1,264 +0,0 @@
use std::{collections::BTreeMap, time::Duration};
use crank_core::GraphqlTarget;
use crank_mapping::JsonPath;
use reqwest::{
Client,
header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue},
};
use serde_json::{Map, Value, json};
use crate::{GraphqlAdapterError, GraphqlRequest, GraphqlResponse};
#[derive(Clone, Debug)]
pub struct GraphqlAdapter {
client: Client,
}
impl Default for GraphqlAdapter {
fn default() -> Self {
Self::new()
}
}
impl GraphqlAdapter {
pub fn new() -> Self {
Self {
client: Client::new(),
}
}
pub async fn execute(
&self,
target: &GraphqlTarget,
request: &GraphqlRequest,
) -> Result<GraphqlResponse, GraphqlAdapterError> {
let endpoint = reqwest::Url::parse(&target.endpoint).map_err(|_| {
GraphqlAdapterError::InvalidEndpoint {
url: target.endpoint.clone(),
}
})?;
let headers = build_headers(request)?;
let variables = normalize_variables(request.variables.clone())?;
let payload = json!({
"query": target.query_template,
"operationName": target.operation_name,
"variables": variables
});
let response = self
.client
.post(endpoint)
.headers(headers)
.timeout(Duration::from_millis(request.timeout_ms))
.json(&payload)
.send()
.await?;
let status = response.status();
let headers = normalize_headers(response.headers());
let body = decode_body(response).await?;
if !status.is_success() {
return Err(GraphqlAdapterError::UnexpectedStatus {
status: status.as_u16(),
body,
});
}
if let Some(errors) = body.get("errors").cloned() {
return Err(GraphqlAdapterError::OperationErrors {
data: body.get("data").cloned(),
errors,
});
}
let response_path = JsonPath::parse(&target.response_path).map_err(|_| {
GraphqlAdapterError::InvalidResponsePath {
path: target.response_path.clone(),
}
})?;
let response_context = json!({
"response": {
"body": body
}
});
let data = response_path
.read(&response_context)
.cloned()
.ok_or_else(|| GraphqlAdapterError::ResponsePathNotFound {
path: target.response_path.clone(),
})?;
let body = response_context["response"]["body"].clone();
Ok(GraphqlResponse {
status_code: status.as_u16(),
headers,
body,
data,
})
}
}
fn build_headers(request: &GraphqlRequest) -> Result<HeaderMap, GraphqlAdapterError> {
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
for (name, value) in &request.headers {
let header_name = HeaderName::try_from(name.as_str()).map_err(|_| {
GraphqlAdapterError::InvalidHeaderName {
header: name.clone(),
}
})?;
let header_value = HeaderValue::try_from(value.as_str()).map_err(|_| {
GraphqlAdapterError::InvalidHeaderValue {
header: name.clone(),
}
})?;
headers.insert(header_name, header_value);
}
Ok(headers)
}
fn normalize_variables(variables: Option<Value>) -> Result<Value, GraphqlAdapterError> {
match variables {
None | Some(Value::Null) => Ok(Value::Object(Map::new())),
Some(Value::Object(values)) => Ok(Value::Object(values)),
Some(_) => Err(GraphqlAdapterError::InvalidVariablesShape),
}
}
async fn decode_body(response: reqwest::Response) -> Result<Value, GraphqlAdapterError> {
let bytes = response.bytes().await?;
if bytes.is_empty() {
return Ok(Value::Null);
}
match serde_json::from_slice::<Value>(&bytes) {
Ok(value) => Ok(value),
Err(_) => Ok(Value::String(
String::from_utf8_lossy(&bytes).trim().to_owned(),
)),
}
}
fn normalize_headers(headers: &HeaderMap) -> BTreeMap<String, String> {
headers
.iter()
.filter_map(|(name, value)| {
value
.to_str()
.ok()
.map(|value| (name.as_str().to_owned(), value.to_owned()))
})
.collect()
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use axum::{Json, Router, routing::post};
use crank_core::{GraphqlOperationType, GraphqlTarget};
use serde_json::{Value, json};
use tokio::net::TcpListener;
use crate::{GraphqlAdapter, GraphqlAdapterError, GraphqlRequest};
#[tokio::test]
async fn executes_graphql_request_and_extracts_response_path() {
let endpoint = spawn_graphql_server().await;
let adapter = GraphqlAdapter::new();
let target = GraphqlTarget {
endpoint,
operation_type: GraphqlOperationType::Mutation,
operation_name: "CreateLead".to_owned(),
query_template:
"mutation CreateLead($email: String!) { createLead(email: $email) { id status } }"
.to_owned(),
response_path: "$.response.body.data.createLead".to_owned(),
};
let request = GraphqlRequest {
headers: BTreeMap::from([("x-trace-id".to_owned(), "trace-123".to_owned())]),
variables: Some(json!({ "email": "user@example.com" })),
timeout_ms: 1_000,
};
let response = adapter.execute(&target, &request).await.unwrap();
assert_eq!(response.status_code, 200);
assert_eq!(
response.data,
json!({ "id": "lead_123", "status": "created" })
);
}
#[tokio::test]
async fn returns_operation_errors_from_graphql_body() {
let endpoint = spawn_graphql_server().await;
let adapter = GraphqlAdapter::new();
let target = GraphqlTarget {
endpoint,
operation_type: GraphqlOperationType::Mutation,
operation_name: "CreateLead".to_owned(),
query_template:
"mutation CreateLead($email: String!) { createLead(email: $email) { id status } }"
.to_owned(),
response_path: "$.response.body.data.createLead".to_owned(),
};
let request = GraphqlRequest {
headers: BTreeMap::new(),
variables: Some(json!({ "email": "fail@example.com" })),
timeout_ms: 1_000,
};
let error = adapter.execute(&target, &request).await.unwrap_err();
assert!(matches!(
error,
GraphqlAdapterError::OperationErrors {
errors: Value::Array(_),
..
}
));
}
async fn spawn_graphql_server() -> String {
let app = Router::new().route("/", post(graphql_handler));
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
format!("http://{}", address)
}
async fn graphql_handler(Json(payload): Json<Value>) -> Json<Value> {
let email = payload
.get("variables")
.and_then(Value::as_object)
.and_then(|variables| variables.get("email"))
.and_then(Value::as_str)
.unwrap_or_default();
if email == "fail@example.com" {
return Json(json!({
"data": { "createLead": null },
"errors": [{ "message": "lead creation failed" }]
}));
}
Json(json!({
"data": {
"createLead": {
"id": "lead_123",
"status": "created"
}
}
}))
}
}
crates/crank-adapter-graphql/src/error.rs
-24
View File
@@ -1,24 +0,0 @@
use serde_json::Value;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum GraphqlAdapterError {
#[error("invalid graphql endpoint: {url}")]
InvalidEndpoint { url: String },
#[error("invalid header name {header}")]
InvalidHeaderName { header: String },
#[error("invalid header value for {header}")]
InvalidHeaderValue { header: String },
#[error("graphql variables must be a JSON object")]
InvalidVariablesShape,
#[error("invalid graphql response path: {path}")]
InvalidResponsePath { path: String },
#[error("graphql response path did not match any value: {path}")]
ResponsePathNotFound { path: String },
#[error("request failed")]
Transport(#[from] reqwest::Error),
#[error("graphql endpoint returned status {status}")]
UnexpectedStatus { status: u16, body: Value },
#[error("graphql operation returned errors")]
OperationErrors { errors: Value, data: Option<Value> },
}
crates/crank-adapter-graphql/src/lib.rs
-7
View File
@@ -1,7 +0,0 @@
mod client;
mod error;
mod model;
pub use client::GraphqlAdapter;
pub use error::GraphqlAdapterError;
pub use model::{GraphqlRequest, GraphqlResponse};
crates/crank-adapter-graphql/src/model.rs
-22
View File
@@ -1,22 +0,0 @@
use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
pub struct GraphqlRequest {
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub headers: BTreeMap<String, String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub variables: Option<Value>,
pub timeout_ms: u64,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct GraphqlResponse {
pub status_code: u16,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub headers: BTreeMap<String, String>,
pub body: Value,
pub data: Value,
}
crates/crank-adapter-grpc/Cargo.toml
-30
View File
@@ -1,30 +0,0 @@
[package]
name = "crank-adapter-grpc"
edition.workspace = true
license.workspace = true
rust-version.workspace = true
version.workspace = true
[features]
test-support = []
[dependencies]
base64.workspace = true
crank-core = { path = "../crank-core" }
crank-proto = { path = "../crank-proto" }
futures-util = "0.3"
prost.workspace = true
prost-reflect.workspace = true
serde.workspace = true
serde_json.workspace = true
thiserror.workspace = true
tonic.workspace = true
tonic-prost.workspace = true
tokio.workspace = true
[dev-dependencies]
prost-types.workspace = true
[build-dependencies]
protoc-bin-vendored.workspace = true
tonic-prost-build.workspace = true
crates/crank-adapter-grpc/build.rs
-17
View File
@@ -1,17 +0,0 @@
use std::{env, path::PathBuf};
fn main() {
let protoc = protoc_bin_vendored::protoc_bin_path().expect("vendored protoc must exist");
unsafe {
env::set_var("PROTOC", protoc);
}
let out_dir = PathBuf::from(env::var("OUT_DIR").expect("OUT_DIR must exist"));
let descriptor_path = out_dir.join("echo_descriptor.bin");
tonic_prost_build::configure()
.build_server(true)
.build_client(true)
.file_descriptor_set_path(descriptor_path)
.compile_protos(&["proto/echo.proto"], &["proto"])
.expect("echo proto must compile");
}
crates/crank-adapter-grpc/examples/stream_fixture.rs
-22
View File
@@ -1,22 +0,0 @@
use crank_adapter_grpc::test_support::{EchoServiceImpl, echo};
use tokio::net::TcpListener;
use tonic::transport::Server;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let bind = std::env::var("CRANK_E2E_GRPC_FIXTURE_BIND")
.unwrap_or_else(|_| "127.0.0.1:3311".to_owned());
let listener = TcpListener::bind(&bind).await?;
let incoming = tonic::transport::server::TcpIncoming::from(listener);
println!("gRPC stream fixture listening on http://{bind}");
Server::builder()
.add_service(echo::echo_service_server::EchoServiceServer::new(
EchoServiceImpl,
))
.serve_with_incoming(incoming)
.await?;
Ok(())
}
crates/crank-adapter-grpc/proto/echo.proto
-16
View File
@@ -1,16 +0,0 @@
syntax = "proto3";
package echo;
service EchoService {
rpc UnaryEcho(EchoRequest) returns (EchoResponse);
rpc ServerEcho(EchoRequest) returns (stream EchoResponse);
}
message EchoRequest {
string message = 1;
}
message EchoResponse {
string message = 1;
}
crates/crank-adapter-grpc/src/client.rs
-314
View File
@@ -1,314 +0,0 @@
use std::{collections::BTreeMap, str::FromStr, time::Duration};
use base64::{Engine as _, engine::general_purpose::STANDARD};
use crank_core::GrpcTarget;
use futures_util::StreamExt;
use prost::Message;
use prost_reflect::{DescriptorPool, MethodDescriptor, prost_types::FileDescriptorSet};
use serde_json::json;
use tokio::time::{Instant, timeout_at};
use tonic::{
Request,
client::Grpc,
metadata::{KeyAndValueRef, MetadataKey, MetadataValue},
transport::Endpoint,
};
use crate::{
GrpcAdapterError, GrpcRequest, GrpcResponse, GrpcWindowRequest, GrpcWindowResponse,
codec::JsonCodec,
};
#[derive(Clone, Debug, Default)]
pub struct GrpcAdapter;
impl GrpcAdapter {
pub fn new() -> Self {
Self
}
pub async fn execute(
&self,
target: &GrpcTarget,
request: &GrpcRequest,
) -> Result<GrpcResponse, GrpcAdapterError> {
let method = resolve_method(target)?;
if method.is_client_streaming() || method.is_server_streaming() {
return Err(GrpcAdapterError::UnsupportedMethodKind {
service: format!("{}.{}", target.package, target.service),
method: target.method.clone(),
});
}
let mut grpc = connect(target, request.timeout_ms).await?;
let path = build_path(&method)?;
let codec = JsonCodec::new(method.input(), method.output());
let mut tonic_request = Request::new(request.body.clone());
apply_headers(&mut tonic_request, &request.headers)?;
let response = grpc.unary(tonic_request, path, codec).await?;
let headers = normalize_headers(response.metadata());
let body = response.into_inner();
Ok(GrpcResponse {
status_code: 200,
headers,
body,
})
}
pub async fn execute_window(
&self,
target: &GrpcTarget,
request: &GrpcWindowRequest,
) -> Result<GrpcWindowResponse, GrpcAdapterError> {
let method = resolve_method(target)?;
if method.is_client_streaming() || !method.is_server_streaming() {
return Err(GrpcAdapterError::UnsupportedStreamingMethodKind {
service: format!("{}.{}", target.package, target.service),
method: target.method.clone(),
});
}
let mut grpc = connect(target, request.request.timeout_ms).await?;
let path = build_path(&method)?;
let codec = JsonCodec::new(method.input(), method.output());
let mut tonic_request = Request::new(request.request.body.clone());
apply_headers(&mut tonic_request, &request.request.headers)?;
let response = grpc.server_streaming(tonic_request, path, codec).await?;
let headers = normalize_headers(response.metadata());
let mut stream = response.into_inner();
let deadline = Instant::now() + Duration::from_millis(request.window_duration_ms);
let mut items = Vec::new();
let mut done = true;
loop {
if request
.max_items
.is_some_and(|limit| items.len() >= limit as usize)
{
done = false;
break;
}
let next_message = match timeout_at(deadline, stream.next()).await {
Ok(next_message) => next_message,
Err(_) => break,
};
let Some(next_message) = next_message else {
break;
};
let message = next_message?;
items.push(message);
}
Ok(GrpcWindowResponse {
status_code: 200,
headers,
body: json!({
"items": items,
"done": done,
}),
})
}
}
async fn connect(
target: &GrpcTarget,
timeout_ms: u64,
) -> Result<Grpc<tonic::transport::Channel>, GrpcAdapterError> {
let endpoint = Endpoint::from_shared(target.server_addr.clone())?
.timeout(Duration::from_millis(timeout_ms));
let channel = endpoint.connect().await?;
let mut grpc = Grpc::new(channel);
grpc.ready()
.await
.map_err(|error| GrpcAdapterError::Status {
code: tonic::Code::Unavailable,
message: error.to_string(),
})?;
Ok(grpc)
}
fn build_path(
method: &MethodDescriptor,
) -> Result<tonic::codegen::http::uri::PathAndQuery, GrpcAdapterError> {
let service_name = method.parent_service().full_name().to_owned();
let method_name = method.name().to_owned();
tonic::codegen::http::uri::PathAndQuery::from_str(&format!("/{service_name}/{method_name}"))
.map_err(|_| GrpcAdapterError::InvalidMethodPath {
service: service_name,
method: method_name,
})
}
fn resolve_method(target: &GrpcTarget) -> Result<MethodDescriptor, GrpcAdapterError> {
let bytes = STANDARD
.decode(&target.descriptor_set_b64)
.map_err(|_| GrpcAdapterError::InvalidDescriptorEncoding)?;
let descriptor_set = FileDescriptorSet::decode(bytes.as_slice())
.map_err(|_| GrpcAdapterError::InvalidDescriptorSet)?;
let pool = DescriptorPool::from_file_descriptor_set(descriptor_set)
.map_err(|_| GrpcAdapterError::InvalidDescriptorPool)?;
let full_service_name = if target.package.is_empty() {
target.service.clone()
} else {
format!("{}.{}", target.package, target.service)
};
let service = pool
.get_service_by_name(&full_service_name)
.ok_or_else(|| GrpcAdapterError::ServiceNotFound {
service: full_service_name.clone(),
})?;
service
.methods()
.find(|method| method.name() == target.method)
.ok_or_else(|| GrpcAdapterError::MethodNotFound {
service: full_service_name,
method: target.method.clone(),
})
}
fn apply_headers(
request: &mut Request<serde_json::Value>,
headers: &BTreeMap<String, String>,
) -> Result<(), GrpcAdapterError> {
for (key, value) in headers {
let metadata_key =
MetadataKey::from_str(key).map_err(|source| GrpcAdapterError::InvalidMetadataKey {
key: key.clone(),
source,
})?;
let metadata_value = MetadataValue::from_str(value).map_err(|source| {
GrpcAdapterError::InvalidMetadataValue {
key: key.clone(),
source,
}
})?;
request.metadata_mut().insert(metadata_key, metadata_value);
}
Ok(())
}
fn normalize_headers(metadata: &tonic::metadata::MetadataMap) -> BTreeMap<String, String> {
metadata
.iter()
.filter_map(|entry| match entry {
KeyAndValueRef::Ascii(key, value) => {
Some((key.as_str().to_owned(), value.to_str().ok()?.to_owned()))
}
KeyAndValueRef::Binary(_, _) => None,
})
.collect()
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use crank_core::{DescriptorId, GrpcTarget};
use serde_json::json;
use crate::{GrpcAdapter, GrpcAdapterError, GrpcRequest, GrpcWindowRequest, test_support};
#[tokio::test]
async fn executes_unary_grpc_request() {
let server_addr = test_support::spawn_unary_echo_server().await;
let adapter = GrpcAdapter::new();
let target = GrpcTarget {
server_addr,
package: "echo".to_owned(),
service: "EchoService".to_owned(),
method: "UnaryEcho".to_owned(),
read_only: false,
descriptor_ref: DescriptorId::new("desc_echo"),
descriptor_set_b64: test_support::descriptor_set_b64(),
};
let request = GrpcRequest {
headers: BTreeMap::new(),
body: json!({ "message": "hello" }),
timeout_ms: 1_000,
};
let response = adapter.execute(&target, &request).await.unwrap();
assert_eq!(response.body, json!({ "message": "hello" }));
}
#[tokio::test]
async fn collects_server_stream_messages_with_window_bounds() {
let server_addr = test_support::spawn_unary_echo_server().await;
let adapter = GrpcAdapter::new();
let target = GrpcTarget {
server_addr,
package: "echo".to_owned(),
service: "EchoService".to_owned(),
method: "ServerEcho".to_owned(),
read_only: false,
descriptor_ref: DescriptorId::new("desc_echo"),
descriptor_set_b64: test_support::descriptor_set_b64(),
};
let request = GrpcWindowRequest {
request: GrpcRequest {
headers: BTreeMap::new(),
body: json!({ "message": "hello" }),
timeout_ms: 1_000,
},
window_duration_ms: 1_000,
max_items: Some(2),
};
let response = adapter.execute_window(&target, &request).await.unwrap();
assert_eq!(
response.body,
json!({
"items": [
{ "message": "hello-one" },
{ "message": "hello-two" }
],
"done": false
})
);
}
#[tokio::test]
async fn rejects_unary_method_in_window_mode() {
let server_addr = test_support::spawn_unary_echo_server().await;
let adapter = GrpcAdapter::new();
let target = GrpcTarget {
server_addr,
package: "echo".to_owned(),
service: "EchoService".to_owned(),
method: "UnaryEcho".to_owned(),
read_only: false,
descriptor_ref: DescriptorId::new("desc_echo"),
descriptor_set_b64: test_support::descriptor_set_b64(),
};
let request = GrpcWindowRequest {
request: GrpcRequest {
headers: BTreeMap::new(),
body: json!({ "message": "hello" }),
timeout_ms: 1_000,
},
window_duration_ms: 1_000,
max_items: Some(10),
};
let error = adapter.execute_window(&target, &request).await.unwrap_err();
assert!(matches!(
error,
GrpcAdapterError::UnsupportedStreamingMethodKind { .. }
));
}
}
crates/crank-adapter-grpc/src/codec.rs
-72
View File
@@ -1,72 +0,0 @@
use prost::Message;
use prost_reflect::{DynamicMessage, MessageDescriptor};
use tonic::{
Status,
codec::{Codec, DecodeBuf, Decoder, EncodeBuf, Encoder},
};
pub struct JsonCodec {
request_descriptor: MessageDescriptor,
response_descriptor: MessageDescriptor,
}
impl JsonCodec {
pub fn new(
request_descriptor: MessageDescriptor,
response_descriptor: MessageDescriptor,
) -> Self {
Self {
request_descriptor,
response_descriptor,
}
}
}
impl Codec for JsonCodec {
type Encode = serde_json::Value;
type Decode = serde_json::Value;
type Encoder = JsonEncoder;
type Decoder = JsonDecoder;
fn encoder(&mut self) -> Self::Encoder {
JsonEncoder(self.request_descriptor.clone())
}
fn decoder(&mut self) -> Self::Decoder {
JsonDecoder(self.response_descriptor.clone())
}
}
pub struct JsonEncoder(MessageDescriptor);
impl Encoder for JsonEncoder {
type Item = serde_json::Value;
type Error = Status;
fn encode(&mut self, item: Self::Item, dst: &mut EncodeBuf<'_>) -> Result<(), Self::Error> {
let message = DynamicMessage::deserialize(self.0.clone(), item)
.map_err(|error| Status::invalid_argument(error.to_string()))?;
message.encode_raw(dst);
Ok(())
}
}
pub struct JsonDecoder(MessageDescriptor);
impl Decoder for JsonDecoder {
type Item = serde_json::Value;
type Error = Status;
fn decode(&mut self, src: &mut DecodeBuf<'_>) -> Result<Option<Self::Item>, Self::Error> {
let mut message = DynamicMessage::new(self.0.clone());
message
.merge(src)
.map_err(|error| Status::internal(error.to_string()))?;
serde_json::to_value(&message)
.map(Some)
.map_err(|error| Status::internal(error.to_string()))
}
}
crates/crank-adapter-grpc/src/error.rs
-52
View File
@@ -1,52 +0,0 @@
use thiserror::Error;
use tonic::{
Code, Status,
metadata::errors::{InvalidMetadataKey, InvalidMetadataValue},
};
#[derive(Debug, Error)]
pub enum GrpcAdapterError {
#[error("invalid descriptor set encoding")]
InvalidDescriptorEncoding,
#[error("invalid descriptor set")]
InvalidDescriptorSet,
#[error("invalid descriptor pool")]
InvalidDescriptorPool,
#[error("service {service} was not found in descriptor set")]
ServiceNotFound { service: String },
#[error("method {method} was not found in service {service}")]
MethodNotFound { service: String, method: String },
#[error("grpc method path is invalid for service {service} and method {method}")]
InvalidMethodPath { service: String, method: String },
#[error("grpc method {service}/{method} is not supported for unary execution")]
UnsupportedMethodKind { service: String, method: String },
#[error("grpc method {service}/{method} does not support server-stream execution")]
UnsupportedStreamingMethodKind { service: String, method: String },
#[error("invalid metadata key {key}")]
InvalidMetadataKey {
key: String,
#[source]
source: InvalidMetadataKey,
},
#[error("invalid metadata value for key {key}")]
InvalidMetadataValue {
key: String,
#[source]
source: InvalidMetadataValue,
},
#[error("transport endpoint is invalid")]
InvalidEndpoint(#[from] tonic::transport::Error),
#[error("stream collection window expired before grpc stream completed")]
WindowExpired,
#[error("grpc status {code}: {message}")]
Status { code: Code, message: String },
}
impl From<Status> for GrpcAdapterError {
fn from(value: Status) -> Self {
Self::Status {
code: value.code(),
message: value.message().to_owned(),
}
}
}
crates/crank-adapter-grpc/src/lib.rs
-11
View File
@@ -1,11 +0,0 @@
mod client;
mod codec;
mod error;
mod model;
pub use client::GrpcAdapter;
pub use error::GrpcAdapterError;
pub use model::{GrpcRequest, GrpcResponse, GrpcWindowRequest, GrpcWindowResponse};
#[cfg(any(test, feature = "test-support"))]
pub mod test_support;
crates/crank-adapter-grpc/src/model.rs
-37
View File
@@ -1,37 +0,0 @@
use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
pub struct GrpcRequest {
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub headers: BTreeMap<String, String>,
pub body: Value,
pub timeout_ms: u64,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct GrpcResponse {
pub status_code: u16,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub headers: BTreeMap<String, String>,
pub body: Value,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
pub struct GrpcWindowRequest {
#[serde(flatten)]
pub request: GrpcRequest,
pub window_duration_ms: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub max_items: Option<u32>,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct GrpcWindowResponse {
pub status_code: u16,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub headers: BTreeMap<String, String>,
pub body: Value,
}
crates/crank-adapter-grpc/src/test_support.rs
-178
View File
@@ -1,178 +0,0 @@
use base64::{Engine as _, engine::general_purpose::STANDARD};
use futures_util::stream;
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
use tokio::net::TcpListener;
use tonic::{Request, Response, Status, transport::Server};
pub mod echo {
tonic::include_proto!("echo");
pub const FILE_DESCRIPTOR_SET: &[u8] = tonic::include_file_descriptor_set!("echo_descriptor");
}
#[derive(Default)]
pub struct EchoServiceImpl;
#[tonic::async_trait]
impl echo::echo_service_server::EchoService for EchoServiceImpl {
async fn unary_echo(
&self,
request: Request<echo::EchoRequest>,
) -> Result<Response<echo::EchoResponse>, Status> {
Ok(Response::new(echo::EchoResponse {
message: request.into_inner().message,
}))
}
type ServerEchoStream = std::pin::Pin<
Box<dyn futures_util::Stream<Item = Result<echo::EchoResponse, Status>> + Send>,
>;
async fn server_echo(
&self,
request: Request<echo::EchoRequest>,
) -> Result<Response<Self::ServerEchoStream>, Status> {
let message = request.into_inner().message;
let events = vec![
Ok(echo::EchoResponse {
message: format!("{message}-one"),
}),
Ok(echo::EchoResponse {
message: format!("{message}-two"),
}),
Ok(echo::EchoResponse {
message: format!("{message}-three"),
}),
];
Ok(Response::new(Box::pin(stream::iter(events))))
}
}
pub async fn spawn_unary_echo_server() -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let incoming = tonic::transport::server::TcpIncoming::from(listener);
tokio::spawn(async move {
Server::builder()
.add_service(echo::echo_service_server::EchoServiceServer::new(
EchoServiceImpl,
))
.serve_with_incoming(incoming)
.await
.unwrap();
});
format!("http://{}", address)
}
pub async fn spawn_metadata_echo_server() -> String {
#[derive(Default)]
struct MetadataEchoServiceImpl;
#[tonic::async_trait]
impl echo::echo_service_server::EchoService for MetadataEchoServiceImpl {
async fn unary_echo(
&self,
request: Request<echo::EchoRequest>,
) -> Result<Response<echo::EchoResponse>, Status> {
let message = request.get_ref().message.clone();
let request_id = request
.metadata()
.get("x-request-id")
.and_then(|value| value.to_str().ok())
.unwrap_or_default();
let correlation_id = request
.metadata()
.get("x-correlation-id")
.and_then(|value| value.to_str().ok())
.unwrap_or_default();
Ok(Response::new(echo::EchoResponse {
message: format!("{}|{}|{}", message, request_id, correlation_id),
}))
}
type ServerEchoStream = std::pin::Pin<
Box<dyn futures_util::Stream<Item = Result<echo::EchoResponse, Status>> + Send>,
>;
async fn server_echo(
&self,
_request: Request<echo::EchoRequest>,
) -> Result<Response<Self::ServerEchoStream>, Status> {
Ok(Response::new(Box::pin(stream::empty())))
}
}
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let incoming = tonic::transport::server::TcpIncoming::from(listener);
tokio::spawn(async move {
Server::builder()
.add_service(echo::echo_service_server::EchoServiceServer::new(
MetadataEchoServiceImpl,
))
.serve_with_incoming(incoming)
.await
.unwrap();
});
format!("http://{}", address)
}
pub async fn spawn_counting_unary_echo_server(request_count: Arc<AtomicUsize>) -> String {
struct CountingEchoServiceImpl {
request_count: Arc<AtomicUsize>,
}
#[tonic::async_trait]
impl echo::echo_service_server::EchoService for CountingEchoServiceImpl {
async fn unary_echo(
&self,
request: Request<echo::EchoRequest>,
) -> Result<Response<echo::EchoResponse>, Status> {
let message = request.into_inner().message;
let count = self.request_count.fetch_add(1, Ordering::SeqCst) + 1;
Ok(Response::new(echo::EchoResponse {
message: format!("{message}-{count}"),
}))
}
type ServerEchoStream = std::pin::Pin<
Box<dyn futures_util::Stream<Item = Result<echo::EchoResponse, Status>> + Send>,
>;
async fn server_echo(
&self,
_request: Request<echo::EchoRequest>,
) -> Result<Response<Self::ServerEchoStream>, Status> {
Ok(Response::new(Box::pin(stream::empty())))
}
}
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let incoming = tonic::transport::server::TcpIncoming::from(listener);
tokio::spawn(async move {
Server::builder()
.add_service(echo::echo_service_server::EchoServiceServer::new(
CountingEchoServiceImpl { request_count },
))
.serve_with_incoming(incoming)
.await
.unwrap();
});
format!("http://{}", address)
}
pub fn descriptor_set_b64() -> String {
STANDARD.encode(echo::FILE_DESCRIPTOR_SET)
}
crates/crank-adapter-soap/Cargo.toml
-19
View File
@@ -1,19 +0,0 @@
[package]
name = "crank-adapter-soap"
edition.workspace = true
license.workspace = true
rust-version.workspace = true
version.workspace = true
[dependencies]
crank-core = { path = "../crank-core" }
crank-mapping = { path = "../crank-mapping" }
reqwest.workspace = true
roxmltree.workspace = true
serde.workspace = true
serde_json.workspace = true
thiserror.workspace = true
tokio.workspace = true
[dev-dependencies]
axum.workspace = true
crates/crank-adapter-soap/src/client.rs
-326
View File
@@ -1,326 +0,0 @@
use std::{collections::BTreeMap, time::Duration};
use crank_core::{SoapTarget, SoapVersion};
use crank_mapping::JsonPath;
use reqwest::{
Client,
header::{HeaderMap, HeaderName, HeaderValue},
};
use serde_json::{Value, json};
use crate::{SoapAdapterError, SoapRequest, SoapResponse, xml};
#[derive(Clone, Debug)]
pub struct SoapAdapter {
client: Client,
}
impl Default for SoapAdapter {
fn default() -> Self {
Self::new()
}
}
impl SoapAdapter {
pub fn new() -> Self {
Self {
client: Client::new(),
}
}
pub async fn execute(
&self,
target: &SoapTarget,
request: &SoapRequest,
) -> Result<SoapResponse, SoapAdapterError> {
let endpoint = target
.endpoint_override
.as_deref()
.ok_or(SoapAdapterError::MissingEndpoint)?;
let headers = build_headers(target, request)?;
let envelope_headers = resolve_envelope_headers(target, request)?;
let envelope = xml::build_envelope(
&target.operation_name,
target.metadata.namespaces.first().map(String::as_str),
&request.body,
envelope_namespace(target.soap_version),
target.binding_style,
&envelope_headers,
);
let response = self
.client
.post(endpoint)
.headers(headers)
.body(envelope)
.timeout(Duration::from_millis(request.timeout_ms))
.send()
.await?;
let status = response.status();
let headers = normalize_headers(response.headers());
let body_text = response.text().await?;
let body = xml::decode_envelope(&body_text)?;
if !status.is_success() {
return Err(SoapAdapterError::UnexpectedStatus {
status: status.as_u16(),
body,
});
}
Ok(SoapResponse {
status_code: status.as_u16(),
headers,
body,
})
}
}
fn resolve_envelope_headers(
target: &SoapTarget,
request: &SoapRequest,
) -> Result<Vec<xml::SoapEnvelopeHeader>, SoapAdapterError> {
let context = json!({
"request": {
"body": request.body.clone(),
},
"mcp": request.body.clone(),
});
let mut rendered = Vec::new();
for header in &target.headers {
let Some(value) = resolve_header_value(header, &context, &request.body)? else {
if header.required {
return Err(SoapAdapterError::MissingRequiredHeader {
name: header.name.clone(),
});
}
continue;
};
rendered.push(xml::SoapEnvelopeHeader {
name: header.name.clone(),
namespace_uri: header.namespace_uri.clone(),
value,
});
}
Ok(rendered)
}
fn resolve_header_value(
header: &crank_core::SoapHeaderConfig,
context: &Value,
body: &Value,
) -> Result<Option<Value>, SoapAdapterError> {
if let Some(path) = header.value_path.as_deref() {
let path = JsonPath::parse(path).map_err(|_| SoapAdapterError::InvalidHeaderValuePath {
path: path.to_owned(),
})?;
return Ok(path.read(context).cloned());
}
Ok(match body {
Value::Object(map) => map.get(&header.name).cloned(),
_ => None,
})
}
fn build_headers(
target: &SoapTarget,
request: &SoapRequest,
) -> Result<HeaderMap, SoapAdapterError> {
let mut headers = HeaderMap::new();
let content_type = match target.soap_version {
SoapVersion::Soap11 => "text/xml; charset=utf-8".to_owned(),
SoapVersion::Soap12 => match target.soap_action.as_deref() {
Some(action) => format!(r#"application/soap+xml; charset=utf-8; action="{action}""#),
None => "application/soap+xml; charset=utf-8".to_owned(),
},
};
headers.insert(
reqwest::header::CONTENT_TYPE,
HeaderValue::from_str(&content_type).map_err(|_| SoapAdapterError::MissingEndpoint)?,
);
headers.insert(
reqwest::header::ACCEPT,
HeaderValue::from_static("application/soap+xml, text/xml, application/xml"),
);
if matches!(target.soap_version, SoapVersion::Soap11) {
if let Some(action) = target.soap_action.as_deref() {
headers.insert(
HeaderName::from_static("soapaction"),
HeaderValue::from_str(action).map_err(|_| SoapAdapterError::MissingEndpoint)?,
);
}
}
for (name, value) in &request.headers {
let header_name =
HeaderName::try_from(name.as_str()).map_err(|_| SoapAdapterError::MissingEndpoint)?;
let header_value =
HeaderValue::try_from(value).map_err(|_| SoapAdapterError::MissingEndpoint)?;
headers.insert(header_name, header_value);
}
Ok(headers)
}
fn envelope_namespace(version: SoapVersion) -> &'static str {
match version {
SoapVersion::Soap11 => "http://schemas.xmlsoap.org/soap/envelope/",
SoapVersion::Soap12 => "http://www.w3.org/2003/05/soap-envelope",
}
}
fn normalize_headers(headers: &HeaderMap) -> BTreeMap<String, String> {
headers
.iter()
.filter_map(|(name, value)| {
value
.to_str()
.ok()
.map(|value| (name.as_str().to_owned(), value.to_owned()))
})
.collect()
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use axum::{Router, body::Bytes, routing::post};
use crank_core::{SoapBindingStyle, SoapOperationMetadata, SoapTarget, SoapVersion, Target};
use serde_json::json;
use tokio::net::TcpListener;
use crate::{SoapAdapter, SoapRequest};
async fn spawn_server() -> String {
async fn handler(body: Bytes) -> String {
let text = String::from_utf8_lossy(&body);
assert!(text.contains("<email>user@example.com</email>"));
r#"<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<CreateLeadResponse>
<id>lead_123</id>
<status>created</status>
</CreateLeadResponse>
</soap:Body>
</soap:Envelope>"#
.to_owned()
}
let app = Router::new().route("/", post(handler));
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
format!("http://{}", address)
}
fn test_target(endpoint: String) -> SoapTarget {
match Target::Soap(SoapTarget {
wsdl_ref: "sample_wsdl".into(),
service_name: "LeadService".to_owned(),
port_name: "LeadPort".to_owned(),
operation_name: "CreateLead".to_owned(),
endpoint_override: Some(endpoint),
soap_version: SoapVersion::Soap11,
soap_action: Some("urn:createLead".to_owned()),
binding_style: SoapBindingStyle::DocumentLiteral,
headers: Vec::new(),
fault_contract: None,
metadata: SoapOperationMetadata {
input_part_names: vec!["CreateLeadRequest".to_owned()],
output_part_names: vec!["CreateLeadResponse".to_owned()],
namespaces: vec!["urn:crm".to_owned()],
},
}) {
Target::Soap(target) => target,
_ => unreachable!(),
}
}
#[tokio::test]
async fn executes_soap_request_response() {
let endpoint = spawn_server().await;
let adapter = SoapAdapter::new();
let response = adapter
.execute(
&test_target(endpoint),
&SoapRequest {
headers: BTreeMap::new(),
body: json!({ "email": "user@example.com" }),
timeout_ms: 1_000,
},
)
.await
.unwrap();
assert_eq!(response.status_code, 200);
assert_eq!(response.body["id"], "lead_123");
}
#[tokio::test]
async fn renders_soap_headers_and_rpc_literal_children() {
async fn handler(body: Bytes) -> String {
let text = String::from_utf8_lossy(&body);
assert!(text.contains("<soap:Header>"));
assert!(
text.contains(
r#"<h:CorrelationId xmlns:h="urn:headers">corr-123</h:CorrelationId>"#
)
);
assert!(text.contains("<m:CreateLead"));
assert!(text.contains("<m:email>user@example.com</m:email>"));
r#"<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<CreateLeadResponse>
<id>lead_rpc</id>
</CreateLeadResponse>
</soap:Body>
</soap:Envelope>"#
.to_owned()
}
let app = Router::new().route("/", post(handler));
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let target = SoapTarget {
binding_style: SoapBindingStyle::RpcLiteral,
headers: vec![crank_core::SoapHeaderConfig {
name: "CorrelationId".to_owned(),
namespace_uri: Some("urn:headers".to_owned()),
required: true,
value_path: Some("$.request.body.correlation_id".to_owned()),
}],
..test_target(format!("http://{}", address))
};
let adapter = SoapAdapter::new();
let response = adapter
.execute(
&target,
&SoapRequest {
headers: BTreeMap::new(),
body: json!({
"email": "user@example.com",
"correlation_id": "corr-123"
}),
timeout_ms: 1_000,
},
)
.await
.unwrap();
assert_eq!(response.body["id"], "lead_rpc");
}
}
crates/crank-adapter-soap/src/error.rs
-28
View File
@@ -1,28 +0,0 @@
use serde_json::Value;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum SoapAdapterError {
#[error("soap endpoint is missing")]
MissingEndpoint,
#[error("invalid SOAP header value path {path}")]
InvalidHeaderValuePath { path: String },
#[error("required SOAP header {name} is missing")]
MissingRequiredHeader { name: String },
#[error("request failed")]
Transport(#[from] reqwest::Error),
#[error("invalid xml payload")]
InvalidXml(#[from] roxmltree::Error),
#[error("soap envelope body was not found")]
MissingBody,
#[error("soap response body was empty")]
EmptyBody,
#[error("soap endpoint returned status {status}")]
UnexpectedStatus { status: u16, body: Value },
#[error("soap fault: {message}")]
SoapFault {
code: Option<String>,
message: String,
detail: Option<Value>,
},
}
crates/crank-adapter-soap/src/lib.rs
-12
View File
@@ -1,12 +0,0 @@
mod client;
mod error;
mod model;
mod wsdl;
mod xml;
pub use client::SoapAdapter;
pub use error::SoapAdapterError;
pub use model::{
SoapOperationSummary, SoapPortSummary, SoapRequest, SoapResponse, SoapServiceSummary,
};
pub use wsdl::inspect_wsdl;
crates/crank-adapter-soap/src/model.rs
-40
View File
@@ -1,40 +0,0 @@
use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SoapRequest {
pub headers: BTreeMap<String, String>,
pub body: Value,
pub timeout_ms: u64,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct SoapResponse {
pub status_code: u16,
pub headers: BTreeMap<String, String>,
pub body: Value,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SoapOperationSummary {
pub operation_name: String,
pub soap_action: Option<String>,
pub binding_style: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SoapPortSummary {
pub port_name: String,
pub binding_name: String,
pub endpoint: Option<String>,
pub soap_version: String,
pub operations: Vec<SoapOperationSummary>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SoapServiceSummary {
pub service_name: String,
pub ports: Vec<SoapPortSummary>,
}
crates/crank-adapter-soap/src/wsdl.rs
-165
View File
@@ -1,165 +0,0 @@
use std::collections::BTreeMap;
use roxmltree::Document;
use crate::{SoapAdapterError, SoapOperationSummary, SoapPortSummary, SoapServiceSummary};
#[derive(Clone, Debug)]
struct BindingSummary {
soap_version: String,
operations: Vec<SoapOperationSummary>,
}
pub fn inspect_wsdl(payload: &[u8]) -> Result<Vec<SoapServiceSummary>, SoapAdapterError> {
let text = std::str::from_utf8(payload).map_err(|_| SoapAdapterError::EmptyBody)?;
let document = Document::parse(text)?;
let bindings = collect_bindings(&document);
let services = document
.descendants()
.filter(|node| node.is_element() && node.tag_name().name() == "service")
.map(|service| {
let service_name = service.attribute("name").unwrap_or_default().to_owned();
let ports = service
.children()
.filter(|node| node.is_element() && node.tag_name().name() == "port")
.map(|port| {
let port_name = port.attribute("name").unwrap_or_default().to_owned();
let binding_name = local_name(port.attribute("binding").unwrap_or_default());
let endpoint = port
.descendants()
.find(|node| node.is_element() && node.tag_name().name() == "address")
.and_then(|node| node.attribute("location"))
.map(ToOwned::to_owned);
let binding = bindings
.get(&binding_name)
.cloned()
.unwrap_or(BindingSummary {
soap_version: "soap_11".to_owned(),
operations: Vec::new(),
});
SoapPortSummary {
port_name,
binding_name,
endpoint,
soap_version: binding.soap_version,
operations: binding.operations,
}
})
.collect();
SoapServiceSummary {
service_name,
ports,
}
})
.collect();
Ok(services)
}
fn collect_bindings(document: &Document<'_>) -> BTreeMap<String, BindingSummary> {
document
.descendants()
.filter(|node| node.is_element() && node.tag_name().name() == "binding")
.map(|binding| {
let name = binding.attribute("name").unwrap_or_default().to_owned();
let soap_binding = binding
.children()
.find(|node| node.is_element() && node.tag_name().name() == "binding");
let soap_namespace = soap_binding
.and_then(|node| node.tag_name().namespace())
.unwrap_or("http://schemas.xmlsoap.org/wsdl/soap/");
let soap_version = if soap_namespace.contains("soap12") {
"soap_12"
} else {
"soap_11"
}
.to_owned();
let binding_style = soap_binding
.and_then(|node| node.attribute("style"))
.map(|value| match value {
"rpc" => "rpc_literal",
_ => "document_literal",
})
.unwrap_or("document_literal")
.to_owned();
let operations = binding
.children()
.filter(|node| node.is_element() && node.tag_name().name() == "operation")
.map(|operation| SoapOperationSummary {
operation_name: operation.attribute("name").unwrap_or_default().to_owned(),
soap_action: operation
.children()
.find(|node| node.is_element() && node.tag_name().name() == "operation")
.and_then(|node| node.attribute("soapAction"))
.map(ToOwned::to_owned),
binding_style: operation
.children()
.find(|node| node.is_element() && node.tag_name().name() == "operation")
.and_then(|node| node.attribute("style"))
.map(|value| match value {
"rpc" => "rpc_literal",
_ => "document_literal",
})
.unwrap_or(binding_style.as_str())
.to_owned(),
})
.collect();
(
name,
BindingSummary {
soap_version,
operations,
},
)
})
.collect()
}
fn local_name(value: &str) -> String {
value.rsplit(':').next().unwrap_or(value).to_owned()
}
#[cfg(test)]
mod tests {
use super::inspect_wsdl;
#[test]
fn parses_services_ports_and_operations_from_wsdl() {
let services = inspect_wsdl(
br#"<?xml version="1.0"?>
<definitions xmlns="http://schemas.xmlsoap.org/wsdl/"
xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
xmlns:tns="urn:crm"
targetNamespace="urn:crm">
<binding name="LeadBinding" type="tns:LeadPortType">
<soap:binding transport="http://schemas.xmlsoap.org/soap/http" style="document"/>
<operation name="CreateLead">
<soap:operation soapAction="urn:createLead"/>
</operation>
</binding>
<service name="LeadService">
<port name="LeadPort" binding="tns:LeadBinding">
<soap:address location="https://soap.example.com/lead"/>
</port>
</service>
</definitions>"#,
)
.unwrap();
assert_eq!(services[0].service_name, "LeadService");
assert_eq!(services[0].ports[0].port_name, "LeadPort");
assert_eq!(services[0].ports[0].soap_version, "soap_11");
assert_eq!(
services[0].ports[0].operations[0].operation_name,
"CreateLead"
);
assert_eq!(
services[0].ports[0].operations[0].soap_action.as_deref(),
Some("urn:createLead")
);
}
}
crates/crank-adapter-soap/src/xml.rs
-319
View File
@@ -1,319 +0,0 @@
use crank_core::SoapBindingStyle;
use roxmltree::{Document, Node};
use serde_json::{Map, Value};
use crate::SoapAdapterError;
#[derive(Clone, Debug, PartialEq)]
pub struct SoapEnvelopeHeader {
pub name: String,
pub namespace_uri: Option<String>,
pub value: Value,
}
pub fn build_envelope(
operation_name: &str,
namespace: Option<&str>,
body: &Value,
envelope_namespace: &str,
binding_style: SoapBindingStyle,
headers: &[SoapEnvelopeHeader],
) -> String {
let mut xml = String::new();
xml.push_str(&format!(
r#"<soap:Envelope xmlns:soap="{envelope_namespace}">"#
));
if !headers.is_empty() {
xml.push_str("<soap:Header>");
for header in headers {
append_header(&mut xml, header);
}
xml.push_str("</soap:Header>");
}
xml.push_str("<soap:Body>");
match namespace {
Some(namespace) => xml.push_str(&format!(r#"<m:{operation_name}" xmlns:m="{namespace}">"#)),
None => xml.push_str(&format!("<{operation_name}>")),
}
let child_prefix = match binding_style {
SoapBindingStyle::DocumentLiteral => None,
SoapBindingStyle::RpcLiteral => namespace.map(|_| "m"),
};
append_value_children(&mut xml, body, child_prefix);
match namespace {
Some(_) => xml.push_str(&format!("</m:{operation_name}>")),
None => xml.push_str(&format!("</{operation_name}>")),
}
xml.push_str("</soap:Body></soap:Envelope>");
xml
}
pub fn decode_envelope(payload: &str) -> Result<Value, SoapAdapterError> {
let document = Document::parse(payload)?;
let body = document
.descendants()
.find(|node| node.is_element() && node.tag_name().name() == "Body")
.ok_or(SoapAdapterError::MissingBody)?;
let body_child = body
.children()
.find(|node| node.is_element())
.ok_or(SoapAdapterError::EmptyBody)?;
if body_child.tag_name().name() == "Fault" {
return Err(parse_fault(body_child));
}
Ok(node_to_json(body_child, true))
}
fn parse_fault(fault: Node<'_, '_>) -> SoapAdapterError {
let code = find_nested_text(fault, &["Code", "Value"])
.or_else(|| find_nested_text(fault, &["faultcode"]));
let message = find_nested_text(fault, &["Reason", "Text"])
.or_else(|| find_nested_text(fault, &["faultstring"]))
.unwrap_or_else(|| "SOAP fault".to_owned());
let detail = fault
.children()
.find(|node| node.is_element() && node.tag_name().name() == "Detail")
.map(|node| node_to_json(node, false));
SoapAdapterError::SoapFault {
code,
message,
detail,
}
}
fn append_header(xml: &mut String, header: &SoapEnvelopeHeader) {
match header.namespace_uri.as_deref() {
Some(namespace_uri) => {
xml.push_str(&format!(
r#"<h:{} xmlns:h="{}">"#,
header.name, namespace_uri
));
append_header_value(xml, &header.value);
xml.push_str(&format!("</h:{}>", header.name));
}
None => {
xml.push('<');
xml.push_str(&header.name);
xml.push('>');
append_header_value(xml, &header.value);
xml.push_str("</");
xml.push_str(&header.name);
xml.push('>');
}
}
}
fn append_header_value(xml: &mut String, value: &Value) {
match value {
Value::Null => {}
Value::Bool(value) => xml.push_str(if *value { "true" } else { "false" }),
Value::Number(value) => xml.push_str(&value.to_string()),
Value::String(value) => xml.push_str(&escape_xml(value)),
Value::Array(_) | Value::Object(_) => append_value_children(xml, value, None),
}
}
fn find_nested_text(node: Node<'_, '_>, path: &[&str]) -> Option<String> {
let mut current = node;
for segment in path {
current = current
.children()
.find(|child| child.is_element() && child.tag_name().name() == *segment)?;
}
current
.text()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn append_value_children(xml: &mut String, value: &Value, namespace_prefix: Option<&str>) {
match value {
Value::Object(object) => {
for (key, value) in object {
append_named_value(xml, key, value, namespace_prefix);
}
}
other => append_named_value(xml, "value", other, namespace_prefix),
}
}
fn append_named_value(xml: &mut String, name: &str, value: &Value, namespace_prefix: Option<&str>) {
match value {
Value::Array(items) => {
for item in items {
append_named_value(xml, name, item, namespace_prefix);
}
}
Value::Object(object) => {
push_start_tag(xml, name, namespace_prefix);
for (child_name, child_value) in object {
append_named_value(xml, child_name, child_value, namespace_prefix);
}
push_end_tag(xml, name, namespace_prefix);
}
Value::Null => {
push_empty_tag(xml, name, namespace_prefix);
}
Value::Bool(value) => {
push_start_tag(xml, name, namespace_prefix);
xml.push_str(if *value { "true" } else { "false" });
push_end_tag(xml, name, namespace_prefix);
}
Value::Number(value) => {
push_start_tag(xml, name, namespace_prefix);
xml.push_str(&value.to_string());
push_end_tag(xml, name, namespace_prefix);
}
Value::String(value) => {
push_start_tag(xml, name, namespace_prefix);
xml.push_str(&escape_xml(value));
push_end_tag(xml, name, namespace_prefix);
}
}
}
fn push_start_tag(xml: &mut String, name: &str, namespace_prefix: Option<&str>) {
xml.push('<');
if let Some(prefix) = namespace_prefix {
xml.push_str(prefix);
xml.push(':');
}
xml.push_str(name);
xml.push('>');
}
fn push_end_tag(xml: &mut String, name: &str, namespace_prefix: Option<&str>) {
xml.push_str("</");
if let Some(prefix) = namespace_prefix {
xml.push_str(prefix);
xml.push(':');
}
xml.push_str(name);
xml.push('>');
}
fn push_empty_tag(xml: &mut String, name: &str, namespace_prefix: Option<&str>) {
xml.push('<');
if let Some(prefix) = namespace_prefix {
xml.push_str(prefix);
xml.push(':');
}
xml.push_str(name);
xml.push_str("/>");
}
fn escape_xml(value: &str) -> String {
value
.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;")
.replace('\'', "&apos;")
}
fn node_to_json(node: Node<'_, '_>, _unwrap_root: bool) -> Value {
let children = node
.children()
.filter(|child| child.is_element())
.collect::<Vec<_>>();
if children.is_empty() {
return node
.text()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| Value::String(value.to_owned()))
.unwrap_or(Value::Null);
}
let mut object = Map::new();
for child in children {
let key = child.tag_name().name().to_owned();
let value = node_to_json(child, false);
match object.get_mut(&key) {
Some(existing) => match existing {
Value::Array(array) => array.push(value),
other => {
let previous = other.take();
*other = Value::Array(vec![previous, value]);
}
},
None => {
object.insert(key, value);
}
}
}
Value::Object(object)
}
#[cfg(test)]
mod tests {
use serde_json::{Value, json};
use super::{SoapEnvelopeHeader, build_envelope, decode_envelope};
use crank_core::SoapBindingStyle;
#[test]
fn builds_document_literal_envelope() {
let xml = build_envelope(
"CreateLead",
Some("urn:crm"),
&json!({"email":"user@example.com","source":"mcp"}),
"http://schemas.xmlsoap.org/soap/envelope/",
SoapBindingStyle::DocumentLiteral,
&[],
);
assert!(xml.contains("<soap:Envelope"));
assert!(xml.contains("<m:CreateLead"));
assert!(xml.contains(r#"xmlns:m="urn:crm""#));
assert!(xml.contains("<email>user@example.com</email>"));
}
#[test]
fn decodes_wrapped_response_body() {
let value = decode_envelope(
r#"<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<CreateLeadResponse>
<id>lead_123</id>
<status>created</status>
</CreateLeadResponse>
</soap:Body>
</soap:Envelope>"#,
)
.unwrap();
assert_eq!(value["id"], "lead_123");
assert_eq!(value["status"], "created");
}
#[test]
fn builds_rpc_literal_envelope_with_headers() {
let xml = build_envelope(
"CreateLead",
Some("urn:crm"),
&json!({"email":"user@example.com"}),
"http://schemas.xmlsoap.org/soap/envelope/",
SoapBindingStyle::RpcLiteral,
&[SoapEnvelopeHeader {
name: "CorrelationId".to_owned(),
namespace_uri: Some("urn:headers".to_owned()),
value: Value::String("corr-123".to_owned()),
}],
);
assert!(xml.contains("<soap:Header>"));
assert!(
xml.contains(r#"<h:CorrelationId xmlns:h="urn:headers">corr-123</h:CorrelationId>"#)
);
assert!(xml.contains("<m:email>user@example.com</m:email>"));
}
}
crates/crank-adapter-websocket/Cargo.toml
-19
View File
@@ -1,19 +0,0 @@
[package]
name = "crank-adapter-websocket"
edition.workspace = true
license.workspace = true
rust-version.workspace = true
version.workspace = true
[dependencies]
crank-core = { path = "../crank-core" }
futures-util = "0.3"
reqwest.workspace = true
serde.workspace = true
serde_json.workspace = true
thiserror.workspace = true
tokio = { workspace = true, features = ["net", "time"] }
tokio-tungstenite.workspace = true
[dev-dependencies]
tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread", "time"] }
crates/crank-adapter-websocket/src/client.rs
-537
View File
@@ -1,537 +0,0 @@
use std::{collections::BTreeMap, time::Duration};
use crank_core::WebsocketTarget;
use futures_util::{SinkExt, StreamExt};
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
use serde_json::Value;
use tokio::time::{Instant, sleep};
use tokio_tungstenite::{
MaybeTlsStream, WebSocketStream, connect_async,
tungstenite::{
Error as TungsteniteError, Message,
client::IntoClientRequest,
protocol::{CloseFrame, frame::coding::CloseCode},
},
};
use crate::{
HeartbeatPolicy, ReconnectPolicy, WebsocketAdapterError, WebsocketWindowRequest,
WebsocketWindowResponse,
};
enum WindowCollectionStatus {
WindowExpired,
MaxItemsReached,
}
#[derive(Clone, Debug)]
pub struct WebsocketAdapter;
impl Default for WebsocketAdapter {
fn default() -> Self {
Self::new()
}
}
impl WebsocketAdapter {
pub fn new() -> Self {
Self
}
pub async fn execute_window(
&self,
target: &WebsocketTarget,
request: &WebsocketWindowRequest,
) -> Result<WebsocketWindowResponse, WebsocketAdapterError> {
let started_at = Instant::now();
let deadline = started_at + Duration::from_millis(request.window_duration_ms);
let heartbeat = request
.heartbeat_interval_ms
.map(Duration::from_millis)
.map(|interval| crate::HeartbeatPolicy { interval });
let reconnect = ReconnectPolicy {
max_attempts: request.reconnect_max_attempts,
backoff: Duration::from_millis(request.reconnect_backoff_ms),
};
let mut attempts = 0_u32;
let mut items = Vec::new();
let mut connected_headers = BTreeMap::new();
loop {
let (mut stream, headers) = connect_websocket(target, request).await?;
if connected_headers.is_empty() {
connected_headers = headers;
}
send_subscribe_message(&mut stream, target).await?;
let status = match collect_window(
&mut stream,
request.max_items,
deadline,
heartbeat.as_ref(),
&mut items,
)
.await
{
Ok(status) => status,
Err(WebsocketAdapterError::ClosedEarly) => {
if attempts >= reconnect.max_attempts {
return Err(WebsocketAdapterError::ReconnectExhausted);
}
attempts = attempts.saturating_add(1);
reconnect_if_needed(&reconnect, attempts).await;
continue;
}
Err(error) => return Err(error),
};
match status {
WindowCollectionStatus::WindowExpired => {
send_unsubscribe_message(&mut stream, target).await?;
return Ok(WebsocketWindowResponse {
status_code: 101,
headers: connected_headers,
body: serde_json::json!({
"items": items,
"done": false,
}),
});
}
WindowCollectionStatus::MaxItemsReached => {
send_unsubscribe_message(&mut stream, target).await?;
return Ok(WebsocketWindowResponse {
status_code: 101,
headers: connected_headers,
body: serde_json::json!({
"items": items,
"done": true,
}),
});
}
}
}
}
}
type WsStream = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
pub async fn connect_websocket(
target: &WebsocketTarget,
request: &WebsocketWindowRequest,
) -> Result<(WsStream, BTreeMap<String, String>), WebsocketAdapterError> {
let mut client_request = target.url.as_str().into_client_request().map_err(|_| {
WebsocketAdapterError::InvalidUrl {
url: target.url.clone(),
}
})?;
let headers = build_headers(target, request)?;
for (name, value) in &headers {
client_request
.headers_mut()
.insert(name.clone(), value.clone());
}
if !target.subprotocols.is_empty() {
let value = target.subprotocols.join(", ");
let header_value = HeaderValue::from_str(&value)
.map_err(|_| WebsocketAdapterError::InvalidSubprotocol { value })?;
client_request
.headers_mut()
.insert("Sec-WebSocket-Protocol", header_value);
}
let (stream, response) = connect_async(client_request).await?;
let response_headers = response
.headers()
.iter()
.filter_map(|(name, value)| {
value
.to_str()
.ok()
.map(|value| (name.as_str().to_owned(), value.to_owned()))
})
.collect();
Ok((stream, response_headers))
}
pub async fn send_subscribe_message(
stream: &mut WsStream,
target: &WebsocketTarget,
) -> Result<(), WebsocketAdapterError> {
let Some(payload) = target.subscribe_message_template.as_ref() else {
return Ok(());
};
let message = serde_json::to_string(payload)
.map_err(|_| WebsocketAdapterError::InvalidSubscribePayload)?;
stream.send(Message::Text(message.into())).await?;
Ok(())
}
pub async fn send_unsubscribe_message(
stream: &mut WsStream,
target: &WebsocketTarget,
) -> Result<(), WebsocketAdapterError> {
let Some(payload) = target.unsubscribe_message_template.as_ref() else {
let _ = stream
.close(Some(CloseFrame {
code: CloseCode::Normal,
reason: "completed".into(),
}))
.await;
return Ok(());
};
let message = serde_json::to_string(payload)
.map_err(|_| WebsocketAdapterError::InvalidUnsubscribePayload)?;
if let Err(error) = stream.send(Message::Text(message.into())).await {
if !matches!(
error,
TungsteniteError::ConnectionClosed
| TungsteniteError::AlreadyClosed
| TungsteniteError::Protocol(
tokio_tungstenite::tungstenite::error::ProtocolError::SendAfterClosing
)
) {
return Err(error.into());
}
}
let _ = stream
.close(Some(CloseFrame {
code: CloseCode::Normal,
reason: "completed".into(),
}))
.await;
Ok(())
}
async fn collect_window(
stream: &mut WsStream,
max_items: Option<u32>,
deadline: Instant,
heartbeat: Option<&HeartbeatPolicy>,
items: &mut Vec<Value>,
) -> Result<WindowCollectionStatus, WebsocketAdapterError> {
let mut heartbeat_deadline = heartbeat.map(|policy| Instant::now() + policy.interval);
loop {
if Instant::now() >= deadline {
return Ok(WindowCollectionStatus::WindowExpired);
}
let now = Instant::now();
let next_tick = heartbeat_deadline.unwrap_or(deadline);
let sleep_until = std::cmp::min(next_tick, deadline);
let wait = sleep_until.saturating_duration_since(now);
let timer = sleep(wait);
tokio::pin!(timer);
tokio::select! {
_ = &mut timer => {
if heartbeat_deadline.is_some_and(|value| value <= Instant::now()) {
heartbeat_tick(stream).await?;
heartbeat_deadline = heartbeat.map(|policy| Instant::now() + policy.interval);
continue;
}
return Ok(WindowCollectionStatus::WindowExpired);
}
frame = read_next_frame(stream) => {
match frame? {
Some(value) => {
items.push(value);
if max_items.is_some_and(|limit| items.len() as u32 >= limit) {
return Ok(WindowCollectionStatus::MaxItemsReached);
}
}
None => return Err(WebsocketAdapterError::ClosedEarly),
}
heartbeat_deadline = heartbeat.map(|policy| Instant::now() + policy.interval);
}
}
}
}
pub async fn read_next_frame(
stream: &mut WsStream,
) -> Result<Option<Value>, WebsocketAdapterError> {
loop {
let Some(frame) = stream.next().await else {
return Ok(None);
};
match frame? {
Message::Text(text) => return Ok(Some(decode_text_frame(text.as_ref())?)),
Message::Binary(_) => return Err(WebsocketAdapterError::InvalidFramePayload),
Message::Ping(payload) => {
stream.send(Message::Pong(payload)).await?;
}
Message::Pong(_) => {}
Message::Frame(_) => {}
Message::Close(_) => return Ok(None),
}
}
}
pub fn decode_text_frame(text: &str) -> Result<Value, WebsocketAdapterError> {
serde_json::from_str(text).or_else(|_| Ok(Value::String(text.to_owned())))
}
pub async fn heartbeat_tick(stream: &mut WsStream) -> Result<(), WebsocketAdapterError> {
stream.send(Message::Ping(Vec::new().into())).await?;
Ok(())
}
pub async fn reconnect_if_needed(policy: &ReconnectPolicy, attempts: u32) {
if attempts == 0 || policy.backoff.is_zero() {
return;
}
sleep(policy.backoff).await;
}
fn build_headers(
target: &WebsocketTarget,
request: &WebsocketWindowRequest,
) -> Result<HeaderMap, WebsocketAdapterError> {
let mut headers = HeaderMap::new();
for (name, value) in target.static_headers.iter().chain(request.headers.iter()) {
let header_name = HeaderName::try_from(name.as_str()).map_err(|_| {
WebsocketAdapterError::InvalidHeaderName {
header: name.clone(),
}
})?;
let header_value = HeaderValue::try_from(value.as_str()).map_err(|_| {
WebsocketAdapterError::InvalidHeaderValue {
header: name.clone(),
}
})?;
headers.insert(header_name, header_value);
}
Ok(headers)
}
#[cfg(test)]
mod tests {
use std::{collections::BTreeMap, sync::Arc};
use futures_util::{SinkExt, StreamExt};
use serde_json::{Value, json};
use tokio::{net::TcpListener, sync::Mutex};
use tokio_tungstenite::{
accept_hdr_async,
tungstenite::handshake::server::{Request, Response},
};
use crate::{WebsocketAdapter, WebsocketWindowRequest};
use crank_core::WebsocketTarget;
#[tokio::test]
async fn collects_window_messages_and_sends_subscribe_payload() {
let received = Arc::new(Mutex::new(Vec::new()));
let target_url = spawn_server(received.clone(), false).await;
let adapter = WebsocketAdapter::new();
let target = WebsocketTarget {
url: target_url,
subprotocols: vec!["events.v1".to_owned()],
subscribe_message_template: Some(json!({"type":"subscribe","topic":"metrics"})),
unsubscribe_message_template: Some(json!({"type":"unsubscribe"})),
static_headers: BTreeMap::from([("x-test-env".to_owned(), "ci".to_owned())]),
};
let response = adapter
.execute_window(
&target,
&WebsocketWindowRequest {
headers: BTreeMap::new(),
window_duration_ms: 1_000,
max_items: Some(3),
heartbeat_interval_ms: None,
reconnect_max_attempts: 0,
reconnect_backoff_ms: 0,
},
)
.await
.unwrap();
assert_eq!(response.status_code, 101);
assert_eq!(response.body["items"].as_array().unwrap().len(), 3);
assert_eq!(response.body["items"][0]["seq"], 1);
let received = received.lock().await.clone();
assert!(
received
.iter()
.any(|value| value == &json!({"type":"subscribe","topic":"metrics"}))
);
}
#[tokio::test]
async fn reconnects_when_socket_closes_early() {
let received = Arc::new(Mutex::new(Vec::new()));
let target_url = spawn_server(received, true).await;
let adapter = WebsocketAdapter::new();
let target = WebsocketTarget {
url: target_url,
subprotocols: Vec::new(),
subscribe_message_template: None,
unsubscribe_message_template: None,
static_headers: BTreeMap::new(),
};
let response = adapter
.execute_window(
&target,
&WebsocketWindowRequest {
headers: BTreeMap::new(),
window_duration_ms: 1_000,
max_items: Some(3),
heartbeat_interval_ms: None,
reconnect_max_attempts: 2,
reconnect_backoff_ms: 10,
},
)
.await
.unwrap();
assert_eq!(response.body["items"].as_array().unwrap().len(), 3);
assert_eq!(response.body["items"][2]["seq"], 3);
assert_eq!(response.body["done"], true);
}
#[tokio::test]
async fn reconnects_after_partial_close_without_marking_done_early() {
let target_url = spawn_partial_close_server().await;
let adapter = WebsocketAdapter::new();
let target = WebsocketTarget {
url: target_url,
subprotocols: Vec::new(),
subscribe_message_template: None,
unsubscribe_message_template: None,
static_headers: BTreeMap::new(),
};
let response = adapter
.execute_window(
&target,
&WebsocketWindowRequest {
headers: BTreeMap::new(),
window_duration_ms: 1_000,
max_items: Some(3),
heartbeat_interval_ms: None,
reconnect_max_attempts: 2,
reconnect_backoff_ms: 10,
},
)
.await
.unwrap();
assert_eq!(response.body["items"].as_array().unwrap().len(), 3);
assert_eq!(response.body["items"][0]["seq"], 1);
assert_eq!(response.body["items"][2]["seq"], 3);
assert_eq!(response.body["done"], true);
}
async fn spawn_server(received: Arc<Mutex<Vec<Value>>>, close_early: bool) -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
let mut accepted = 0_u32;
loop {
let (stream, _) = listener.accept().await.unwrap();
let received = received.clone();
accepted = accepted.saturating_add(1);
tokio::spawn(async move {
let websocket =
accept_hdr_async(stream, |request: &Request, mut response: Response| {
if let Some(value) = request.headers().get("x-test-env") {
assert_eq!(value, "ci");
}
if let Some(value) = request.headers().get("sec-websocket-protocol") {
response
.headers_mut()
.insert("sec-websocket-protocol", value.clone());
}
Ok(response)
})
.await
.unwrap();
let (mut sink, mut source) = websocket.split();
let mut sent = 0_u32;
if !close_early {
if let Some(message) = source.next().await {
let message = message.unwrap();
if let tokio_tungstenite::tungstenite::Message::Text(text) = message {
if let Ok(value) = serde_json::from_str::<Value>(&text) {
received.lock().await.push(value);
}
}
}
}
let payloads = if close_early && accepted == 1 {
Vec::new()
} else {
vec![json!({"seq": 1}), json!({"seq": 2}), json!({"seq": 3})]
};
for payload in payloads {
sink.send(tokio_tungstenite::tungstenite::Message::Text(
payload.to_string().into(),
))
.await
.unwrap();
sent += 1;
}
if sent < 3 {
let _ = sink.close().await;
}
});
}
});
format!("ws://{}", addr)
}
async fn spawn_partial_close_server() -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
let mut accepted = 0_u32;
loop {
let (stream, _) = listener.accept().await.unwrap();
accepted = accepted.saturating_add(1);
tokio::spawn(async move {
let websocket =
accept_hdr_async(stream, |_request: &Request, response: Response| {
Ok(response)
})
.await
.unwrap();
let (mut sink, _source) = websocket.split();
let payloads = if accepted == 1 {
vec![json!({"seq": 1})]
} else {
vec![json!({"seq": 2}), json!({"seq": 3})]
};
for payload in payloads {
sink.send(tokio_tungstenite::tungstenite::Message::Text(
payload.to_string().into(),
))
.await
.unwrap();
}
let _ = sink.close().await;
});
}
});
format!("ws://{}", addr)
}
}
crates/crank-adapter-websocket/src/error.rs
-30
View File
@@ -1,30 +0,0 @@
use serde_json::Value;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum WebsocketAdapterError {
#[error("invalid websocket url: {url}")]
InvalidUrl { url: String },
#[error("invalid websocket header name {header}")]
InvalidHeaderName { header: String },
#[error("invalid websocket header value for {header}")]
InvalidHeaderValue { header: String },
#[error("invalid websocket subprotocol {value}")]
InvalidSubprotocol { value: String },
#[error("websocket connect failed")]
Connect(#[from] tokio_tungstenite::tungstenite::Error),
#[error("websocket window expired before collecting any items")]
WindowExpired,
#[error("websocket stream produced malformed frame payload")]
InvalidFramePayload,
#[error("websocket endpoint returned close frame before collection completed")]
ClosedEarly,
#[error("websocket reconnect policy exhausted")]
ReconnectExhausted,
#[error("websocket upstream returned invalid subscribe payload")]
InvalidSubscribePayload,
#[error("websocket upstream returned invalid unsubscribe payload")]
InvalidUnsubscribePayload,
#[error("websocket upstream status {status}")]
UnexpectedStatus { status: u16, body: Value },
}
crates/crank-adapter-websocket/src/lib.rs
-35
View File
@@ -1,35 +0,0 @@
mod client;
mod error;
mod session;
use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use serde_json::Value;
pub use client::WebsocketAdapter;
pub use error::WebsocketAdapterError;
pub use session::{HeartbeatPolicy, ReconnectPolicy};
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
pub struct WebsocketWindowRequest {
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub headers: BTreeMap<String, String>,
pub window_duration_ms: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub max_items: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub heartbeat_interval_ms: Option<u64>,
#[serde(default)]
pub reconnect_max_attempts: u32,
#[serde(default)]
pub reconnect_backoff_ms: u64,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct WebsocketWindowResponse {
pub status_code: u16,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub headers: BTreeMap<String, String>,
pub body: Value,
}
crates/crank-adapter-websocket/src/session.rs
-21
View File
@@ -1,21 +0,0 @@
use std::time::Duration;
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ReconnectPolicy {
pub max_attempts: u32,
pub backoff: Duration,
}
impl Default for ReconnectPolicy {
fn default() -> Self {
Self {
max_attempts: 0,
backoff: Duration::from_millis(0),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct HeartbeatPolicy {
pub interval: Duration,
}
crates/crank-proto/Cargo.toml
-15
View File
@@ -1,15 +0,0 @@
[package]
name = "crank-proto"
edition.workspace = true
license.workspace = true
rust-version.workspace = true
version.workspace = true
[dependencies]
crank-core = { path = "../crank-core" }
crank-schema = { path = "../crank-schema" }
prost.workspace = true
prost-reflect.workspace = true
prost-types.workspace = true
serde.workspace = true
thiserror.workspace = true
crates/crank-proto/src/convert/mod.rs
-1
View File
@@ -1 +0,0 @@
pub mod to_schema;
crates/crank-proto/src/convert/to_schema.rs
-381
View File
@@ -1,381 +0,0 @@
use std::collections::BTreeMap;
use crank_schema::{Schema, SchemaKind};
use crate::{
errors::ProtoError,
model::{
ProtoEnum, ProtoField, ProtoFieldCardinality, ProtoFieldType, ProtoMapField, ProtoMessage,
ProtoOneof, ProtoScalarKind,
},
};
pub fn message_to_schema(message: &ProtoMessage) -> Result<Schema, ProtoError> {
let mut fields = BTreeMap::new();
for field in &message.fields {
let (name, schema) = convert_field(field)?;
fields.insert(name, schema);
}
for oneof in &message.oneofs {
let (name, schema) = convert_oneof(oneof)?;
fields.insert(name, schema);
}
Ok(object_schema(fields))
}
fn convert_field(field: &ProtoField) -> Result<(String, Schema), ProtoError> {
let mut schema = convert_field_type(&field.field_type)?;
schema.required = matches!(field.cardinality, ProtoFieldCardinality::Required);
schema.nullable = false;
if matches!(field.cardinality, ProtoFieldCardinality::Repeated) {
schema = Schema {
kind: SchemaKind::Array,
description: None,
required: false,
nullable: false,
default_value: None,
fields: BTreeMap::new(),
items: Some(Box::new(schema)),
enum_values: Vec::new(),
variants: Vec::new(),
};
}
Ok((field.name.clone(), schema))
}
fn convert_field_type(field_type: &ProtoFieldType) -> Result<Schema, ProtoError> {
match field_type {
ProtoFieldType::Scalar { scalar } => Ok(scalar_schema(scalar)),
ProtoFieldType::Message { message } => message_to_schema(message),
ProtoFieldType::Enum { enumeration } => Ok(enum_schema(enumeration)),
ProtoFieldType::Map { map } => convert_map(map),
}
}
fn convert_map(map: &ProtoMapField) -> Result<Schema, ProtoError> {
let entry = object_schema(BTreeMap::from([
("key".to_owned(), scalar_schema(&map.key)),
("value".to_owned(), convert_field_type(&map.value)?),
]));
Ok(Schema {
kind: SchemaKind::Array,
description: None,
required: false,
nullable: false,
default_value: None,
fields: BTreeMap::new(),
items: Some(Box::new(entry)),
enum_values: Vec::new(),
variants: Vec::new(),
})
}
fn convert_oneof(oneof: &ProtoOneof) -> Result<(String, Schema), ProtoError> {
if oneof.variants.is_empty() {
return Err(ProtoError::EmptyOneof {
oneof_name: oneof.name.clone(),
});
}
let mut variants = Vec::with_capacity(oneof.variants.len());
for field in &oneof.variants {
let (field_name, field_schema) = convert_field(field)?;
variants.push(object_schema(BTreeMap::from([(field_name, field_schema)])));
}
Ok((
oneof.name.clone(),
Schema {
kind: SchemaKind::Oneof,
description: None,
required: false,
nullable: false,
default_value: None,
fields: BTreeMap::new(),
items: None,
enum_values: Vec::new(),
variants,
},
))
}
fn object_schema(fields: BTreeMap<String, Schema>) -> Schema {
Schema {
kind: SchemaKind::Object,
description: None,
required: true,
nullable: false,
default_value: None,
fields,
items: None,
enum_values: Vec::new(),
variants: Vec::new(),
}
}
fn enum_schema(enumeration: &ProtoEnum) -> Schema {
Schema {
kind: SchemaKind::Enum,
description: None,
required: false,
nullable: false,
default_value: None,
fields: BTreeMap::new(),
items: None,
enum_values: enumeration.values.clone(),
variants: Vec::new(),
}
}
fn scalar_schema(kind: &ProtoScalarKind) -> Schema {
let schema_kind = match kind {
ProtoScalarKind::String | ProtoScalarKind::Bytes => SchemaKind::String,
ProtoScalarKind::Bool => SchemaKind::Boolean,
ProtoScalarKind::Int32
| ProtoScalarKind::Int64
| ProtoScalarKind::Uint32
| ProtoScalarKind::Uint64 => SchemaKind::Integer,
ProtoScalarKind::Float | ProtoScalarKind::Double => SchemaKind::Number,
};
Schema {
kind: schema_kind,
description: None,
required: false,
nullable: false,
default_value: None,
fields: BTreeMap::new(),
items: None,
enum_values: Vec::new(),
variants: Vec::new(),
}
}
#[cfg(test)]
mod tests {
use crank_schema::SchemaKind;
use crate::{
ProtoEnum, ProtoField, ProtoFieldCardinality, ProtoFieldType, ProtoMapField, ProtoMessage,
ProtoMethod, ProtoOneof, ProtoScalarKind, ProtoService, message_to_schema,
};
#[test]
fn service_filters_unary_methods() {
let unary = ProtoMethod {
name: "GetLead".to_owned(),
input: empty_message("GetLeadRequest"),
output: empty_message("GetLeadResponse"),
client_streaming: false,
server_streaming: false,
};
let streaming = ProtoMethod {
name: "StreamLeads".to_owned(),
input: empty_message("StreamLeadsRequest"),
output: empty_message("StreamLeadsResponse"),
client_streaming: false,
server_streaming: true,
};
let service = ProtoService {
package: "crm.v1".to_owned(),
name: "LeadService".to_owned(),
methods: vec![unary.clone(), streaming],
};
let unary_methods = service.unary_methods().collect::<Vec<_>>();
assert_eq!(unary_methods.len(), 1);
assert_eq!(unary_methods[0].name, unary.name);
}
#[test]
fn converts_nested_message_to_schema() {
let request = ProtoMessage {
name: "CreateLeadRequest".to_owned(),
fields: vec![ProtoField {
name: "lead".to_owned(),
json_name: "lead".to_owned(),
cardinality: ProtoFieldCardinality::Required,
field_type: ProtoFieldType::Message {
message: Box::new(ProtoMessage {
name: "Lead".to_owned(),
fields: vec![ProtoField {
name: "email".to_owned(),
json_name: "email".to_owned(),
cardinality: ProtoFieldCardinality::Required,
field_type: ProtoFieldType::Scalar {
scalar: ProtoScalarKind::String,
},
}],
oneofs: Vec::new(),
}),
},
}],
oneofs: Vec::new(),
};
let schema = message_to_schema(&request).unwrap();
assert_eq!(schema.kind, SchemaKind::Object);
assert_eq!(
schema.field("lead").map(|field| field.kind.clone()),
Some(SchemaKind::Object)
);
assert_eq!(
schema
.field("lead")
.and_then(|field| field.field("email"))
.map(|field| field.kind.clone()),
Some(SchemaKind::String)
);
assert_eq!(
schema
.field("lead")
.and_then(|field| field.field("email"))
.map(|field| field.required),
Some(true)
);
}
#[test]
fn converts_enum_map_and_oneof_contracts() {
let response = ProtoMessage {
name: "GetLeadResponse".to_owned(),
fields: vec![
ProtoField {
name: "status".to_owned(),
json_name: "status".to_owned(),
cardinality: ProtoFieldCardinality::Optional,
field_type: ProtoFieldType::Enum {
enumeration: ProtoEnum {
name: "LeadStatus".to_owned(),
values: vec!["OPEN".to_owned(), "CLOSED".to_owned()],
},
},
},
ProtoField {
name: "attributes".to_owned(),
json_name: "attributes".to_owned(),
cardinality: ProtoFieldCardinality::Optional,
field_type: ProtoFieldType::Map {
map: ProtoMapField {
key: ProtoScalarKind::String,
value: Box::new(ProtoFieldType::Scalar {
scalar: ProtoScalarKind::String,
}),
},
},
},
],
oneofs: vec![ProtoOneof {
name: "contact".to_owned(),
variants: vec![
ProtoField {
name: "email".to_owned(),
json_name: "email".to_owned(),
cardinality: ProtoFieldCardinality::Optional,
field_type: ProtoFieldType::Scalar {
scalar: ProtoScalarKind::String,
},
},
ProtoField {
name: "phone".to_owned(),
json_name: "phone".to_owned(),
cardinality: ProtoFieldCardinality::Optional,
field_type: ProtoFieldType::Scalar {
scalar: ProtoScalarKind::String,
},
},
],
}],
};
let schema = message_to_schema(&response).unwrap();
assert_eq!(
schema.field("status").map(|field| field.kind.clone()),
Some(SchemaKind::Enum)
);
assert_eq!(
schema
.field("status")
.map(|field| field.enum_values.clone())
.unwrap(),
vec!["OPEN".to_owned(), "CLOSED".to_owned()]
);
let attributes = schema.field("attributes").unwrap();
assert_eq!(attributes.kind, SchemaKind::Array);
assert_eq!(
attributes
.items
.as_deref()
.and_then(|entry| entry.field("key"))
.map(|field| field.kind.clone()),
Some(SchemaKind::String)
);
assert_eq!(
attributes
.items
.as_deref()
.and_then(|entry| entry.field("value"))
.map(|field| field.kind.clone()),
Some(SchemaKind::String)
);
let contact = schema.field("contact").unwrap();
assert_eq!(contact.kind, SchemaKind::Oneof);
assert_eq!(contact.variants.len(), 2);
assert_eq!(
contact.variants[0]
.field("email")
.map(|field| field.kind.clone()),
Some(SchemaKind::String)
);
assert_eq!(
contact.variants[1]
.field("phone")
.map(|field| field.kind.clone()),
Some(SchemaKind::String)
);
}
#[test]
fn rejects_empty_oneof() {
let message = ProtoMessage {
name: "BrokenMessage".to_owned(),
fields: Vec::new(),
oneofs: vec![ProtoOneof {
name: "selection".to_owned(),
variants: Vec::new(),
}],
};
let error = message_to_schema(&message).unwrap_err();
assert_eq!(
error,
crate::ProtoError::EmptyOneof {
oneof_name: "selection".to_owned(),
}
);
}
fn empty_message(name: &str) -> ProtoMessage {
ProtoMessage {
name: name.to_owned(),
fields: Vec::new(),
oneofs: Vec::new(),
}
}
}
crates/crank-proto/src/errors.rs
-11
View File
@@ -1,11 +0,0 @@
use thiserror::Error;
#[derive(Clone, Debug, PartialEq, Eq, Error)]
pub enum ProtoError {
#[error("oneof {oneof_name} must contain at least one variant")]
EmptyOneof { oneof_name: String },
#[error("descriptor set could not be decoded")]
InvalidDescriptorSet,
#[error("descriptor pool could not be built")]
InvalidDescriptorPool,
}
crates/crank-proto/src/lib.rs
-142
View File
@@ -1,142 +0,0 @@
pub mod convert;
pub mod errors;
pub mod model;
use prost::Message;
use prost_reflect::{
Cardinality, DescriptorPool, EnumDescriptor, FieldDescriptor, Kind, MessageDescriptor,
MethodDescriptor, ServiceDescriptor,
};
use prost_types::FileDescriptorSet;
pub use convert::to_schema::message_to_schema;
pub use errors::ProtoError;
pub use model::{
ProtoEnum, ProtoField, ProtoFieldCardinality, ProtoFieldType, ProtoMapField, ProtoMessage,
ProtoMethod, ProtoOneof, ProtoScalarKind, ProtoService,
};
pub fn services_from_descriptor_set_bytes(bytes: &[u8]) -> Result<Vec<ProtoService>, ProtoError> {
let descriptor_set =
FileDescriptorSet::decode(bytes).map_err(|_| ProtoError::InvalidDescriptorSet)?;
let pool = DescriptorPool::from_file_descriptor_set(descriptor_set)
.map_err(|_| ProtoError::InvalidDescriptorPool)?;
Ok(pool.services().map(service_from_descriptor).collect())
}
fn service_from_descriptor(service: ServiceDescriptor) -> ProtoService {
ProtoService {
package: service.parent_file().package_name().to_owned(),
name: service.name().to_owned(),
methods: service.methods().map(method_from_descriptor).collect(),
}
}
fn method_from_descriptor(method: MethodDescriptor) -> ProtoMethod {
ProtoMethod {
name: method.name().to_owned(),
input: message_from_descriptor(method.input()),
output: message_from_descriptor(method.output()),
client_streaming: method.is_client_streaming(),
server_streaming: method.is_server_streaming(),
}
}
fn message_from_descriptor(message: MessageDescriptor) -> ProtoMessage {
let fields = message
.fields()
.filter(|field| field.containing_oneof().is_none())
.map(field_from_descriptor)
.collect();
let oneofs = message
.oneofs()
.map(|oneof| ProtoOneof {
name: oneof.name().to_owned(),
variants: oneof.fields().map(field_from_descriptor).collect(),
})
.collect();
ProtoMessage {
name: message.name().to_owned(),
fields,
oneofs,
}
}
fn field_from_descriptor(field: FieldDescriptor) -> ProtoField {
let cardinality = if field.is_list() && !field.is_map() {
ProtoFieldCardinality::Repeated
} else {
match field.cardinality() {
Cardinality::Optional => ProtoFieldCardinality::Optional,
Cardinality::Required => ProtoFieldCardinality::Required,
Cardinality::Repeated => ProtoFieldCardinality::Repeated,
}
};
ProtoField {
name: field.name().to_owned(),
json_name: field.json_name().to_owned(),
cardinality,
field_type: field_type_from_descriptor(field),
}
}
fn field_type_from_descriptor(field: FieldDescriptor) -> ProtoFieldType {
if field.is_map() {
return map_field_from_descriptor(field);
}
match field.kind() {
Kind::Bool => scalar_field(ProtoScalarKind::Bool),
Kind::String => scalar_field(ProtoScalarKind::String),
Kind::Bytes => scalar_field(ProtoScalarKind::Bytes),
Kind::Int32 | Kind::Sint32 | Kind::Sfixed32 => scalar_field(ProtoScalarKind::Int32),
Kind::Int64 | Kind::Sint64 | Kind::Sfixed64 => scalar_field(ProtoScalarKind::Int64),
Kind::Uint32 | Kind::Fixed32 => scalar_field(ProtoScalarKind::Uint32),
Kind::Uint64 | Kind::Fixed64 => scalar_field(ProtoScalarKind::Uint64),
Kind::Float => scalar_field(ProtoScalarKind::Float),
Kind::Double => scalar_field(ProtoScalarKind::Double),
Kind::Message(message) => ProtoFieldType::Message {
message: Box::new(message_from_descriptor(message)),
},
Kind::Enum(enumeration) => ProtoFieldType::Enum {
enumeration: enum_from_descriptor(enumeration),
},
}
}
fn map_field_from_descriptor(field: FieldDescriptor) -> ProtoFieldType {
let message = match field.kind() {
Kind::Message(message) => message,
_ => unreachable!(),
};
let key_field = message.map_entry_key_field();
let value_field = message.map_entry_value_field();
let key = match field_type_from_descriptor(key_field) {
ProtoFieldType::Scalar { scalar } => scalar,
_ => ProtoScalarKind::String,
};
ProtoFieldType::Map {
map: ProtoMapField {
key,
value: Box::new(field_type_from_descriptor(value_field)),
},
}
}
fn scalar_field(scalar: ProtoScalarKind) -> ProtoFieldType {
ProtoFieldType::Scalar { scalar }
}
fn enum_from_descriptor(enumeration: EnumDescriptor) -> ProtoEnum {
ProtoEnum {
name: enumeration.name().to_owned(),
values: enumeration
.values()
.map(|value| value.name().to_owned())
.collect(),
}
}
crates/crank-proto/src/model/field.rs
-54
View File
@@ -1,54 +0,0 @@
use serde::{Deserialize, Serialize};
use crate::model::ProtoMessage;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ProtoScalarKind {
String,
Bool,
Int32,
Int64,
Uint32,
Uint64,
Float,
Double,
Bytes,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProtoEnum {
pub name: String,
pub values: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProtoMapField {
pub key: ProtoScalarKind,
pub value: Box<ProtoFieldType>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ProtoFieldType {
Scalar { scalar: ProtoScalarKind },
Message { message: Box<ProtoMessage> },
Enum { enumeration: ProtoEnum },
Map { map: ProtoMapField },
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ProtoFieldCardinality {
Optional,
Required,
Repeated,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProtoField {
pub name: String,
pub json_name: String,
pub cardinality: ProtoFieldCardinality,
pub field_type: ProtoFieldType,
}
crates/crank-proto/src/model/message.rs
-16
View File
@@ -1,16 +0,0 @@
use serde::{Deserialize, Serialize};
use crate::model::ProtoField;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProtoOneof {
pub name: String,
pub variants: Vec<ProtoField>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProtoMessage {
pub name: String,
pub fields: Vec<ProtoField>,
pub oneofs: Vec<ProtoOneof>,
}
crates/crank-proto/src/model/method.rs
-18
View File
@@ -1,18 +0,0 @@
use serde::{Deserialize, Serialize};
use crate::model::ProtoMessage;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProtoMethod {
pub name: String,
pub input: ProtoMessage,
pub output: ProtoMessage,
pub client_streaming: bool,
pub server_streaming: bool,
}
impl ProtoMethod {
pub fn is_unary(&self) -> bool {
!self.client_streaming && !self.server_streaming
}
}
crates/crank-proto/src/model/mod.rs
-11
View File
@@ -1,11 +0,0 @@
mod field;
mod message;
mod method;
mod service;
pub use field::{
ProtoEnum, ProtoField, ProtoFieldCardinality, ProtoFieldType, ProtoMapField, ProtoScalarKind,
};
pub use message::{ProtoMessage, ProtoOneof};
pub use method::ProtoMethod;
pub use service::ProtoService;
crates/crank-proto/src/model/service.rs
-16
View File
@@ -1,16 +0,0 @@
use serde::{Deserialize, Serialize};
use crate::model::ProtoMethod;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProtoService {
pub package: String,
pub name: String,
pub methods: Vec<ProtoMethod>,
}
impl ProtoService {
pub fn unary_methods(&self) -> impl Iterator<Item = &ProtoMethod> {
self.methods.iter().filter(|method| method.is_unary())
}
}