New example

Author: ardatan

lib/executor/src/lib.rs

@@ -12,4 +12,4 @@ pub mod variables;
 
 pub use execution::plan::execute_query_plan;
 pub use executors::map::SubgraphExecutorMap;
-pub use plugins::response_cache::*;
+pub use plugins::*;

lib/executor/src/plugins/examples/mod.rs

@@ -0,0 +1,2 @@
+pub mod response_cache;
+pub mod subgraph_response_cache;

lib/executor/src/plugins/response_cache.rs renamed to lib/executor/src/plugins/examples/response_cache.rs

@@ -3,10 +3,9 @@ use ntex::web::HttpResponse;
 use redis::Commands;
 
 use crate::{
-    plugins::traits::{
-        ControlFlow, OnExecutePayload, OnSchemaReloadPayload, RouterPlugin
-    },
-    utils::consts::TYPENAME_FIELD_NAME,
+    hooks::{on_execute::OnExecutePayload, on_schema_reload::OnSchemaReloadPayload}, plugins::plugin_trait::{
+        ControlFlow, RouterPlugin
+    }, utils::consts::TYPENAME_FIELD_NAME
 };
 
 pub struct ResponseCachePlugin {

lib/executor/src/plugins/examples/subgraph_response_cache.rs

@@ -0,0 +1,32 @@
+use dashmap::DashMap;
+
+use crate::{hooks::on_subgraph_execute::{OnSubgraphExecuteEndPayload, OnSubgraphExecuteStartPayload, SubgraphExecutorResponse, SubgraphResponse}, plugin_trait::{ControlFlow, RouterPlugin}};
+
+struct SubgraphResponseCachePlugin {
+    cache: DashMap<String, SubgraphResponse<'static>>,
+}
+
+impl RouterPlugin for SubgraphResponseCachePlugin {
+    fn on_subgraph_execute<'exec>(
+            &self, 
+            payload: OnSubgraphExecuteStartPayload<'exec>,
+        ) -> ControlFlow<'exec, OnSubgraphExecuteEndPayload<'exec>> {
+        let key = format!(
+            "subgraph_response_cache:{}:{}:{:?}",
+            payload.subgraph_name, payload.execution_request.operation_name.unwrap_or(""), payload.execution_request.variables
+        );
+        if let Some(cached_response) = self.cache.get(&key) {
+            *payload.response = Some(SubgraphExecutorResponse::RawResponse(cached_response));
+            // Return early with the cached response
+            return ControlFlow::Continue;
+        } else {
+            ControlFlow::OnEnd(Box::new(move |payload: OnSubgraphExecuteEndPayload| {
+                let cacheable = payload.response.errors.is_none_or(|errors| errors.is_empty());
+                if cacheable {
+                    self.cache.insert(key, *payload.response);
+                }
+                ControlFlow::Continue
+            }))
+        }
+    }
+}

lib/executor/src/plugins/hooks/mod.rs

@@ -0,0 +1,3 @@
+pub mod on_execute;
+pub mod on_schema_reload;
+pub mod on_subgraph_execute;

lib/executor/src/plugins/hooks/on_execute.rs

@@ -0,0 +1,33 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use hive_router_query_planner::planner::plan_nodes::QueryPlan;
+use ntex::web::HttpRequest;
+
+use crate::response::{value::Value};
+use crate::response::graphql_error::GraphQLError;
+
+pub struct OnExecutePayload<'exec> {
+    pub router_http_request: &'exec HttpRequest,
+    pub query_plan: Arc<QueryPlan>,
+
+    pub data: &'exec mut Value<'exec>,
+    pub errors: &'exec mut Vec<GraphQLError>,
+    pub extensions: &'exec mut HashMap<String, sonic_rs::Value>,
+
+    pub skip_execution: bool,
+
+    pub variable_values: &'exec Option<HashMap<String, sonic_rs::Value>>,
+}
+
+pub struct OnExecuteEndPayload<'exec> {
+    pub router_http_request: &'exec HttpRequest,
+    pub query_plan: Arc<QueryPlan>,
+
+    pub data: &'exec Value<'exec>,
+    pub errors: &'exec Vec<GraphQLError>,
+    pub extensions: &'exec mut HashMap<String, sonic_rs::Value>,
+
+    pub variable_values: &'exec Option<HashMap<String, sonic_rs::Value>>,
+}
+

lib/executor/src/plugins/hooks/on_schema_reload.rs

@@ -0,0 +1,6 @@
+use hive_router_query_planner::consumer_schema::ConsumerSchema;
+
+pub struct OnSchemaReloadPayload {
+    pub old_schema: &'static ConsumerSchema,
+    pub new_schema: &'static mut ConsumerSchema,
+}

lib/executor/src/plugins/hooks/on_subgraph_execute.rs

@@ -0,0 +1,47 @@
+use std::collections::HashMap;
+
+use bytes::Bytes;
+use hive_router_query_planner::ast::operation::SubgraphFetchOperation;
+use ntex::web::HttpRequest;
+
+use crate::{executors::dedupe::SharedResponse, response::{graphql_error::GraphQLError, value::Value}};
+
+
+
+pub struct OnSubgraphExecuteStartPayload<'exec> {
+    pub router_http_request: &'exec HttpRequest,
+	pub subgraph_name: &'exec str,
+	// The node that initiates this subgraph execution
+	pub execution_request: &'exec mut SubgraphExecutionRequest<'exec>,
+	// This will be tricky to implement with the current structure,
+	// but I'm sure we'll figure it out
+	pub response: &'exec mut Option<SubgraphResponse<'exec>>,
+}
+
+pub struct SubgraphExecutionRequest<'exec> {
+    pub query: &'exec str,
+    // We can add the original operation here too
+    pub operation: &'exec SubgraphFetchOperation,
+    
+    pub dedupe: bool,
+    pub operation_name: Option<&'exec str>,
+    pub variables: Option<HashMap<&'exec str, &'exec sonic_rs::Value>>,
+    pub extensions: Option<HashMap<String, sonic_rs::Value>>,
+    pub representations: Option<Vec<u8>>,
+}
+
+pub struct SubgraphResponse<'exec> {
+    pub data: Value<'exec>,
+    pub errors: Option<Vec<GraphQLError>>,
+    pub extensions: Option<HashMap<String, Value<'exec>>>,
+}
+
+pub struct OnSubgraphExecuteEndPayload<'exec> {
+    pub router_http_request: &'exec HttpRequest,
+    pub subgraph_name: &'exec str,
+    // The node that initiates this subgraph execution
+    pub execution_request: &'exec SubgraphExecutionRequest<'exec>,
+    // This will be tricky to implement with the current structure,
+    // but I'm sure we'll figure it out
+    pub response: &'exec SubgraphResponse<'exec>,
+}

lib/executor/src/plugins/mod.rs

@@ -1,2 +1,3 @@
-pub mod response_cache;
-pub mod traits;
+pub mod examples;
+pub mod plugin_trait;
+pub mod hooks;

lib/executor/src/plugins/plugin_trait.rs

@@ -0,0 +1,28 @@
+use ntex::web::HttpResponse;
+
+use crate::hooks::on_execute::OnExecutePayload;
+use crate::hooks::on_schema_reload::OnSchemaReloadPayload;
+use crate::hooks::on_subgraph_execute::OnSubgraphExecuteEndPayload;
+use crate::hooks::on_subgraph_execute::OnSubgraphExecuteStartPayload;
+
+pub enum ControlFlow<'a, TPayload> {
+    Continue,
+    Break(HttpResponse),
+    OnEnd(Box<dyn FnOnce(TPayload) -> ControlFlow<'a, ()> + Send + 'a>),
+}
+
+pub trait RouterPlugin {
+    fn on_execute<'exec>(
+        &self, 
+        _payload: OnExecutePayload<'exec>,
+    ) -> ControlFlow<'exec, OnExecutePayload<'exec>> {
+        ControlFlow::Continue
+    }
+    fn on_subgraph_execute<'exec>(
+        &self, 
+        _payload: OnSubgraphExecuteStartPayload<'exec>,
+    ) -> ControlFlow<'exec, OnSubgraphExecuteEndPayload<'exec>> {
+        ControlFlow::Continue
+    }
+    fn on_schema_reload(&self, _payload: OnSchemaReloadPayload) {}
+}