the-guild-org · dotansimha · Aug 20, 2023 · Aug 18, 2023 · Aug 18, 2023 · Aug 20, 2023
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -21,7 +21,6 @@ tracing-subscriber = "0.3.17"
 async-trait = "0.1.73"
 hyper-tls = "0.5.0"
 axum-macros = "0.3.8"
-sync_wrapper = "0.1.2"
 tokio-util = { version = "0.7.8", features = ["io", "compat"] }
 
 [dev-dependencies]

diff --git a/src/config/mod.rs b/src/config/mod.rs
@@ -7,6 +7,7 @@ pub struct ConductorConfig {
     pub logger: LoggerConfig,
     pub sources: Vec<SourceDefinition>,
     pub endpoints: Vec<EndpointDefinition>,
+    pub global_plugins: Option<Vec<PluginDefinition>>,
 }
 
 #[derive(Deserialize, Debug, Clone)]
@@ -23,6 +24,9 @@ pub struct EndpointDefinition {
 pub enum PluginDefinition {
     #[serde(rename = "verbose_logging")]
     VerboseLogging,
+
+    #[serde(rename = "json_content_type_response")]
+    JSONContentTypeResponse,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -91,7 +95,7 @@ pub struct GraphQLSourceConfig {
     pub endpoint: String,
 }
 
-#[tracing::instrument]
+#[tracing::instrument(level = "trace")]
 pub async fn load_config(file_path: &String) -> ConductorConfig {
     let path = Path::new(file_path);
     let contents = read_to_string(file_path).expect("Failed to read config file");

diff --git a/src/endpoint/endpoint_runtime.rs b/src/endpoint/endpoint_runtime.rs
@@ -5,7 +5,7 @@
 };
 use axum::body::Body;
 use serde_json::json;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 
 pub type EndpointResponse = hyper::Response<Body>;
 
@@ -39,7 +39,7 @@
 pub struct EndpointRuntime {
     pub config: EndpointDefinition,
     pub plugin_manager: Arc<PluginManager>,
-    pub upstream: Arc<Mutex<dyn SourceService + Send>>,
+    pub upstream: Arc<Box<dyn SourceService + Send>>,
 }
 
 impl EndpointRuntime {
@@ -50,7 +50,7 @@
     ) -> Self {
         Self {
             config: endpoint_config,
-            upstream: Arc::new(Mutex::new(source)),
+            upstream: Arc::new(Box::new(source)),
             plugin_manager,
         }
     }
@@ -63,16 +63,12 @@
             Some(source_request) => {
                 // DOTAN: Can we avoid cloning here?
                 let upstream_request = SourceRequest::from_parts(
-                    source_request.operation_name.clone(),
-                    source_request.query.clone(),
-                    Some(source_request.variables.clone()),
+                    source_request.operation_name.as_ref().map(|e| e.as_str()),
+                    source_request.query.as_ref(),
+                    Some(&source_request.variables),
                 );
 
-                let source_result = self
-                    .upstream
-                    .lock()
-                    .expect("upstream service lock coudln't be acquired")
-                    .call(upstream_request);
+                let source_result = self.upstream.call(upstream_request);
 
                 // DOTAN: We probably need some kind of handling for network-related errors here,
                 // I guess some kind of static "upstream is not healthy" error response?

diff --git a/src/lib.rs b/src/lib.rs
@@ -24,7 +24,6 @@ use crate::config::SourceDefinition;
 use crate::plugins::plugin_manager::PluginManager;
 use crate::source::graphql_source::GraphQLSourceService;
 
-#[derive(Clone)]
 pub struct RouterState {
     pub plugin_manager: Arc<PluginManager>,
 }
@@ -37,7 +36,7 @@ pub async fn serve_graphiql_ide(req: Request<Body>) -> impl IntoResponse {
 pub async fn handle_post(
     Extension(endpoint): Extension<EndpointRuntime>,
     headers: HeaderMap,
-    body_stream: BodyStream,
+    mut body_stream: BodyStream,
 ) -> impl IntoResponse {
     // This represents the main context that is shared across the execution.
     // We'll use this struct to pass data between the plugins, endpoint and source.
@@ -50,7 +49,9 @@ pub async fn handle_post(
     // Execute plugins on the HTTP request.
     // Later we need to think how to expose things like the request body to the
     // plugins, if needed,without having to read it twice.
-    flow_ctx = endpoint.plugin_manager.on_downstream_http_request(flow_ctx);
+    endpoint
+        .plugin_manager
+        .on_downstream_http_request(&mut flow_ctx);
 
     // In case the response was set by one of the plugins at this stage, just short-circuit and return it.
     if let Some(sc_response) = flow_ctx.short_circuit_response {
@@ -60,30 +61,33 @@ pub async fn handle_post(
     // In case the GraphQL operation was not extracted from the HTTP request yet, do it now.
     // This is done in order to allow plugins to override/set the GraphQL operation, for use-cases like persisted operations.
     if flow_ctx.downstream_graphql_request.is_none() {
-        flow_ctx = flow_ctx
-            .extract_graphql_request_from_http_request(body_stream)
+        flow_ctx
+            .extract_graphql_request_from_http_request(&mut body_stream)
             .await;
     }
 
     // Execute plugins on the GraphQL request.
-    flow_ctx = endpoint
+    endpoint
         .plugin_manager
-        .on_downstream_graphql_request(flow_ctx);
+        .on_downstream_graphql_request(&mut flow_ctx);
 
     // In case the response was set by one of the plugins at this stage, just short-circuit and return it.
     if let Some(sc_response) = flow_ctx.short_circuit_response {
         return sc_response;
     }
 
     // Run the actual endpoint handler and get the response.
-    let endpoint_response = endpoint.handle_request(flow_ctx).await;
-    flow_ctx = endpoint_response.0;
+    let (mut flow_ctx, mut endpoint_response) = endpoint.handle_request(flow_ctx).await;
+
+    endpoint
+        .plugin_manager
+        .on_downstream_http_response(&mut flow_ctx);
 
     endpoint
         .plugin_manager
-        .on_downstream_http_response(flow_ctx);
+        .on_upstream_graphql_response(&mut endpoint_response);
 
-    match endpoint_response.1 {
+    match endpoint_response {
         Ok(response) => response,
         Err(e) => e.into(),
     }
@@ -104,8 +108,18 @@ pub async fn run_services(config_file_path: String) {
     let server_config = config_object.server.clone();
     let mut http_router = Router::new();
 
+    let global_plugins = &config_object.global_plugins;
+
     for endpoint_config in config_object.endpoints.into_iter() {
-        let plugin_manager = Arc::new(PluginManager::new(&endpoint_config.plugins));
+        let combined_plugins = global_plugins
+            .iter()
+            .chain(&endpoint_config.plugins)
+            .flat_map(|vec| vec.iter())
+            .cloned()
+            .collect::<Vec<_>>();
+
+        let plugin_manager = Arc::new(PluginManager::new(&Some(combined_plugins)));
+
         let upstream_source = config_object
             .sources
             .iter()

diff --git a/src/plugins/core.rs b/src/plugins/core.rs
@@ -1,21 +1,20 @@
 use std::fmt::Debug;
 
-use crate::source::base_source::SourceRequest;
+use hyper::Body;
+
+use crate::{endpoint::endpoint_runtime::EndpointError, source::base_source::SourceRequest};
 
 use super::flow_context::FlowContext;
 
 pub trait Plugin: Sync + Send {
-    fn on_downstream_http_request(&self, mut _ctx: FlowContext) -> FlowContext {
-        _ctx
-    }
-    fn on_downstream_http_response(&self, mut _ctx: FlowContext) -> FlowContext {
-        _ctx
-    }
-    fn on_downstream_graphql_request(&self, mut _ctx: FlowContext) -> FlowContext {
-        _ctx
-    }
-    fn on_upstream_graphql_request(&self, mut _req: SourceRequest) -> SourceRequest {
-        _req
+    fn on_downstream_http_request(&self, _ctx: &mut FlowContext) {}
+    fn on_downstream_http_response(&self, _ctx: &mut FlowContext) {}
+    fn on_downstream_graphql_request(&self, _ctx: &mut FlowContext) {}
+    fn on_upstream_graphql_request<'a>(&self, _req: &mut SourceRequest<'a>) {}
+    fn on_upstream_graphql_response<'a>(
+        &self,
+        _response: &mut Result<hyper::Response<Body>, EndpointError>,
+    ) {
     }
 }
 

diff --git a/src/plugins/flow_context.rs b/src/plugins/flow_context.rs
@@ -14,11 +14,11 @@ pub struct FlowContext {
 }
 
 impl FlowContext {
-    #[tracing::instrument]
+    #[tracing::instrument(level = "trace")]
     pub async fn extract_graphql_request_from_http_request(
-        mut self,
-        body_stream: BodyStream,
-    ) -> Self {
+        &mut self,
+        body_stream: &mut BodyStream,
+    ) {
         let content_type = self
             .downstream_headers
             .get(axum::http::header::CONTENT_TYPE)
@@ -41,6 +41,5 @@ impl FlowContext {
                 .unwrap();
 
         self.downstream_graphql_request = Some(graphql_request);
-        self
     }
 }
diff --git a/src/plugins/json_content_type_response_plugin.rs b/src/plugins/json_content_type_response_plugin.rs
@@ -0,0 +1,18 @@
+use crate::endpoint::endpoint_runtime::EndpointError;
+
+use super::core::Plugin;
+use hyper::{
+    header::{HeaderValue, CONTENT_TYPE},
+    Body,
+};
+
+pub struct JSONContentTypePlugin {}
+
+impl Plugin for JSONContentTypePlugin {
+    fn on_upstream_graphql_response(&self, req: &mut Result<hyper::Response<Body>, EndpointError>) {
+        if let Ok(res) = req {
+            let headers = res.headers_mut();
+            headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        }
+    }
+}
diff --git a/src/plugins/mod.rs b/src/plugins/mod.rs
@@ -1,4 +1,5 @@
 pub mod core;
 pub mod flow_context;
+pub mod json_content_type_response_plugin;
 pub mod plugin_manager;
 pub mod verbose_logging_plugin;