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

runtime: gate premium protocol adapters by feature

Browse Source
This commit is contained in:
a.tolmachev
2026-05-07 19:03:39 +00:00
parent c4ac6e85c0
commit 978ed3a02e
8 changed files with 301 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/crank-runtime/src/error.rs
+45 -12
View File
@@ -1,29 +1,34 @@
use crank_adapter_graphql::GraphqlAdapterError;
use crank_adapter_grpc::GrpcAdapterError;
use crank_adapter_rest::RestAdapterError;
use crank_adapter_soap::SoapAdapterError;
use crank_adapter_websocket::WebsocketAdapterError;
use crank_core::{ExecutionMode, Protocol};
use crank_mapping::MappingError;
use crank_schema::SchemaError;
use thiserror::Error;
#[cfg(feature = "protocol-graphql")]
use crank_adapter_graphql::GraphqlAdapterError;
#[cfg(feature = "protocol-grpc")]
use crank_adapter_grpc::GrpcAdapterError;
#[cfg(feature = "protocol-soap")]
use crank_adapter_soap::SoapAdapterError;
#[cfg(feature = "protocol-websocket")]
use crank_adapter_websocket::WebsocketAdapterError;
#[derive(Debug, Error)]
pub enum RuntimeError {
#[error(transparent)]
Schema(#[from] SchemaError),
#[error(transparent)]
Mapping(#[from] MappingError),
#[error(transparent)]
GraphqlAdapter(#[from] GraphqlAdapterError),
#[error(transparent)]
GrpcAdapter(#[from] GrpcAdapterError),
#[error("{0}")]
GraphqlAdapter(String),
#[error("{0}")]
GrpcAdapter(String),
#[error(transparent)]
RestAdapter(#[from] RestAdapterError),
#[error(transparent)]
SoapAdapter(#[from] SoapAdapterError),
#[error(transparent)]
WebsocketAdapter(#[from] WebsocketAdapterError),
#[error("{0}")]
SoapAdapter(String),
#[error("{0}")]
WebsocketAdapter(String),
#[error("protocol {protocol:?} is not supported by runtime")]
UnsupportedProtocol { protocol: Protocol },
#[error("operation {operation_id} does not define streaming config")]
@@ -53,3 +58,31 @@ pub enum RuntimeError {
details: String,
},
}
#[cfg(feature = "protocol-graphql")]
impl From<GraphqlAdapterError> for RuntimeError {
fn from(value: GraphqlAdapterError) -> Self {
Self::GraphqlAdapter(value.to_string())
}
}
#[cfg(feature = "protocol-grpc")]
impl From<GrpcAdapterError> for RuntimeError {
fn from(value: GrpcAdapterError) -> Self {
Self::GrpcAdapter(value.to_string())
}
}
#[cfg(feature = "protocol-soap")]
impl From<SoapAdapterError> for RuntimeError {
fn from(value: SoapAdapterError) -> Self {
Self::SoapAdapter(value.to_string())
}
}
#[cfg(feature = "protocol-websocket")]
impl From<WebsocketAdapterError> for RuntimeError {
fn from(value: WebsocketAdapterError) -> Self {
Self::WebsocketAdapter(value.to_string())
}
}
crates/crank-runtime/src/executor.rs
+189 -126
View File
@@ -3,11 +3,7 @@ use std::sync::Arc;
use std::time::Duration;
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use crank_adapter_graphql::{GraphqlAdapter, GraphqlRequest};
use crank_adapter_grpc::{GrpcAdapter, GrpcRequest, GrpcWindowRequest};
use crank_adapter_rest::{RestAdapter, RestRequest, RestWindowRequest};
use crank_adapter_soap::{SoapAdapter, SoapRequest};
use crank_adapter_websocket::{WebsocketAdapter, WebsocketWindowRequest};
use crank_core::{
CachedHeader, CachedResponse, ExecutionMode, HttpMethod, ResponseCacheStore, Target,
TransportBehavior,
@@ -22,12 +18,25 @@ use crate::{
RuntimeRequestContext, WindowExecutionResult,
};
#[cfg(feature = "protocol-graphql")]
use crank_adapter_graphql::{GraphqlAdapter, GraphqlRequest};
#[cfg(feature = "protocol-grpc")]
use crank_adapter_grpc::{GrpcAdapter, GrpcRequest, GrpcWindowRequest};
#[cfg(feature = "protocol-soap")]
use crank_adapter_soap::{SoapAdapter, SoapRequest};
#[cfg(feature = "protocol-websocket")]
use crank_adapter_websocket::{WebsocketAdapter, WebsocketWindowRequest};
#[derive(Clone)]
pub struct RuntimeExecutor {
#[cfg(feature = "protocol-graphql")]
graphql_adapter: GraphqlAdapter,
#[cfg(feature = "protocol-grpc")]
grpc_adapter: GrpcAdapter,
rest_adapter: RestAdapter,
#[cfg(feature = "protocol-soap")]
soap_adapter: SoapAdapter,
#[cfg(feature = "protocol-websocket")]
websocket_adapter: WebsocketAdapter,
limits: RuntimeLimits,
unary_limiter: Arc<Semaphore>,
@@ -50,10 +59,14 @@ impl RuntimeExecutor {
pub fn with_limits(limits: RuntimeLimits) -> Self {
Self {
#[cfg(feature = "protocol-graphql")]
graphql_adapter: GraphqlAdapter::new(),
#[cfg(feature = "protocol-grpc")]
grpc_adapter: GrpcAdapter::new(),
rest_adapter: RestAdapter::new(),
#[cfg(feature = "protocol-soap")]
soap_adapter: SoapAdapter::new(),
#[cfg(feature = "protocol-websocket")]
websocket_adapter: WebsocketAdapter::new(),
unary_limiter: Arc::new(Semaphore::new(limits.max_concurrent_unary)),
window_limiter: Arc::new(Semaphore::new(limits.max_concurrent_window)),
@@ -312,47 +325,67 @@ impl RuntimeExecutor {
let context_headers = runtime_context_headers(request_context);
match &operation.target {
Target::Grpc(target) => {
let request = GrpcRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request
.grpc
.clone()
.unwrap_or(Value::Object(Map::new())),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.grpc_adapter.execute(target, &request).await?;
#[cfg(not(feature = "protocol-grpc"))]
{
let _ = target;
return Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
});
}
#[cfg(feature = "protocol-grpc")]
{
let request = GrpcRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request
.grpc
.clone()
.unwrap_or(Value::Object(Map::new())),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.grpc_adapter.execute(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body.clone(),
data: response.body,
})
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body.clone(),
data: response.body,
})
}
}
Target::Graphql(target) => {
let request = GraphqlRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
variables: prepared_request.variables.clone(),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.graphql_adapter.execute(target, &request).await?;
#[cfg(not(feature = "protocol-graphql"))]
{
let _ = target;
return Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
});
}
#[cfg(feature = "protocol-graphql")]
{
let request = GraphqlRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
variables: prepared_request.variables.clone(),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.graphql_adapter.execute(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: response.data,
})
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: response.data,
})
}
}
Target::Rest(target) => {
let request = RestRequest {
@@ -377,27 +410,37 @@ impl RuntimeExecutor {
})
}
Target::Soap(target) => {
let request = SoapRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request
.body
.clone()
.unwrap_or(Value::Object(Map::new())),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.soap_adapter.execute(target, &request).await?;
#[cfg(not(feature = "protocol-soap"))]
{
let _ = target;
Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
})
}
#[cfg(feature = "protocol-soap")]
{
let request = SoapRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request
.body
.clone()
.unwrap_or(Value::Object(Map::new())),
timeout_ms: operation.execution_config.timeout_ms,
};
let response = self.soap_adapter.execute(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
}
}
Target::Websocket(_) => Err(RuntimeError::UnsupportedExecutionMode {
operation_id: operation.operation_id.as_str().to_owned(),
@@ -449,79 +492,99 @@ impl RuntimeExecutor {
})
}
Target::Grpc(target) => {
let Some(streaming) = operation.execution_config.streaming.as_ref() else {
return Err(RuntimeError::MissingStreamingConfig {
operation_id: operation.operation_id.as_str().to_owned(),
#[cfg(not(feature = "protocol-grpc"))]
{
let _ = target;
return Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
});
};
let request = GrpcWindowRequest {
request: GrpcRequest {
}
#[cfg(feature = "protocol-grpc")]
{
let Some(streaming) = operation.execution_config.streaming.as_ref() else {
return Err(RuntimeError::MissingStreamingConfig {
operation_id: operation.operation_id.as_str().to_owned(),
});
};
let request = GrpcWindowRequest {
request: GrpcRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request
.grpc
.clone()
.unwrap_or(Value::Object(Map::new())),
timeout_ms: streaming
.upstream_timeout_ms
.unwrap_or(operation.execution_config.timeout_ms),
},
window_duration_ms: streaming.window_duration_ms.unwrap_or_default(),
max_items: streaming.max_items.map(|value| value.saturating_add(1)),
};
let response = self.grpc_adapter.execute_window(target, &request).await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
}
}
Target::Websocket(target) => {
#[cfg(not(feature = "protocol-websocket"))]
{
let _ = target;
Err(RuntimeError::UnsupportedProtocol {
protocol: operation.protocol,
})
}
#[cfg(feature = "protocol-websocket")]
{
let Some(streaming) = operation.execution_config.streaming.as_ref() else {
return Err(RuntimeError::MissingStreamingConfig {
operation_id: operation.operation_id.as_str().to_owned(),
});
};
let websocket_options = operation
.execution_config
.protocol_options
.as_ref()
.and_then(|options| options.websocket.as_ref());
let request = WebsocketWindowRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
body: prepared_request
.grpc
.clone()
.unwrap_or(Value::Object(Map::new())),
timeout_ms: streaming
.upstream_timeout_ms
.unwrap_or(operation.execution_config.timeout_ms),
},
window_duration_ms: streaming.window_duration_ms.unwrap_or_default(),
max_items: streaming.max_items.map(|value| value.saturating_add(1)),
};
let response = self.grpc_adapter.execute_window(target, &request).await?;
window_duration_ms: streaming.window_duration_ms.unwrap_or_default(),
max_items: streaming.max_items.map(|value| value.saturating_add(1)),
heartbeat_interval_ms: websocket_options
.and_then(|options| options.heartbeat_interval_ms),
reconnect_max_attempts: websocket_options
.and_then(|options| options.reconnect_max_attempts)
.unwrap_or_default(),
reconnect_backoff_ms: websocket_options
.and_then(|options| options.reconnect_backoff_ms)
.unwrap_or_default(),
};
let response = self
.websocket_adapter
.execute_window(target, &request)
.await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
}
Target::Websocket(target) => {
let Some(streaming) = operation.execution_config.streaming.as_ref() else {
return Err(RuntimeError::MissingStreamingConfig {
operation_id: operation.operation_id.as_str().to_owned(),
});
};
let websocket_options = operation
.execution_config
.protocol_options
.as_ref()
.and_then(|options| options.websocket.as_ref());
let request = WebsocketWindowRequest {
headers: merge_headers(
&BTreeMap::new(),
&operation.execution_config.headers,
&prepared_request.headers,
&context_headers,
),
window_duration_ms: streaming.window_duration_ms.unwrap_or_default(),
max_items: streaming.max_items.map(|value| value.saturating_add(1)),
heartbeat_interval_ms: websocket_options
.and_then(|options| options.heartbeat_interval_ms),
reconnect_max_attempts: websocket_options
.and_then(|options| options.reconnect_max_attempts)
.unwrap_or_default(),
reconnect_backoff_ms: websocket_options
.and_then(|options| options.reconnect_backoff_ms)
.unwrap_or_default(),
};
let response = self
.websocket_adapter
.execute_window(target, &request)
.await?;
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
Ok(AdapterResponse {
status_code: response.status_code,
headers: response.headers,
body: response.body,
data: Value::Null,
})
}
}
Target::Soap(_) => Err(RuntimeError::UnsupportedExecutionMode {
operation_id: operation.operation_id.as_str().to_owned(),