server: Refactor away GraphQLResponse

Author: leoyvens

Cargo.lock

@@ -23,6 +23,7 @@ reqwest = "0.10"
 # ethabi, but long term we want to find a way to drop our fork.
 ethabi = { git = "https://github.com/graphprotocol/ethabi.git", branch = "master" }
 hex = "0.4.2"
+http = "0.2"
 futures = "0.1.21"
 graphql-parser = "0.2.3"
 ipfs-api = { version = "0.7.1", features = ["hyper-tls"] }

graph/Cargo.toml

@@ -1,19 +1,15 @@
 use futures::prelude::*;
 
 use crate::data::graphql::effort::LoadManager;
-use crate::data::query::{Query, QueryError, QueryResult};
+use crate::data::query::{Query, QueryResult};
 use crate::data::subscription::{Subscription, SubscriptionError, SubscriptionResult};
 
 use async_trait::async_trait;
 use failure::format_err;
 use failure::Error;
-use futures03::compat::Future01CompatExt;
 use graphql_parser::query as q;
 use std::sync::Arc;
 
-/// Future for query results.
-pub type QueryResultFuture = Box<dyn Future<Item = Arc<QueryResult>, Error = QueryError> + Send>;
-
 /// Future for subscription results.
 pub type SubscriptionResultFuture =
     Box<dyn Future<Item = SubscriptionResult, Error = SubscriptionError> + Send>;
@@ -22,34 +18,32 @@ pub type SubscriptionResultFuture =
 #[async_trait]
 pub trait GraphQlRunner: Send + Sync + 'static {
     /// Runs a GraphQL query and returns its result.
-    fn run_query(&self, query: Query) -> QueryResultFuture;
+    async fn run_query(self: Arc<Self>, query: Query) -> Arc<QueryResult>;
 
     /// Runs a GraphqL query up to the given complexity. Overrides the global complexity limit.
-    fn run_query_with_complexity(
+    async fn run_query_with_complexity(
         &self,
         query: Query,
         max_complexity: Option<u64>,
         max_depth: Option<u8>,
         max_first: Option<u32>,
-    ) -> QueryResultFuture;
+    ) -> Arc<QueryResult>;
 
     /// Runs a GraphQL subscription and returns a stream of results.
     fn run_subscription(&self, subscription: Subscription) -> SubscriptionResultFuture;
 
     async fn query_metadata(&self, query: Query) -> Result<q::Value, Error> {
-        self.run_query_with_complexity(query, None, None, None)
-            .compat()
-            .await
-            .map_err(move |e| format_err!("Failed to query metadata: {}", e))
-            .and_then(move |result| {
-                // Metadata queries are not cached.
-                let result = Arc::try_unwrap(result).unwrap();
-                if result.errors.is_some() {
-                    Err(format_err!("Failed to query metadata: {:?}", result.errors))
-                } else {
-                    result.data.ok_or_else(|| format_err!("No metadata found"))
-                }
-            })
+        let result = self
+            .run_query_with_complexity(query, None, None, None)
+            .await;
+
+        // Metadata queries are not cached.
+        let result = Arc::try_unwrap(result).unwrap();
+        if result.errors.is_some() {
+            Err(format_err!("Failed to query metadata: {:?}", result.errors))
+        } else {
+            result.data.ok_or_else(|| format_err!("No metadata found"))
+        }
     }
 
     fn load_manager(&self) -> Arc<LoadManager>;

graph/src/components/graphql.rs

@@ -1,49 +1,26 @@
 use crate::data::query::QueryError;
 use futures::prelude::*;
-use futures::sync::oneshot::Canceled;
 use serde::ser::*;
 use std::error::Error;
 use std::fmt;
 
 /// Errors that can occur while processing incoming requests.
 #[derive(Debug)]
 pub enum GraphQLServerError {
-    Canceled(Canceled),
     ClientError(String),
     QueryError(QueryError),
     InternalError(String),
 }
 
-impl From<Canceled> for GraphQLServerError {
-    fn from(e: Canceled) -> Self {
-        GraphQLServerError::Canceled(e)
-    }
-}
-
 impl From<QueryError> for GraphQLServerError {
     fn from(e: QueryError) -> Self {
         GraphQLServerError::QueryError(e)
     }
 }
 
-impl From<&'static str> for GraphQLServerError {
-    fn from(s: &'static str) -> Self {
-        GraphQLServerError::InternalError(String::from(s))
-    }
-}
-
-impl From<String> for GraphQLServerError {
-    fn from(s: String) -> Self {
-        GraphQLServerError::InternalError(s)
-    }
-}
-
 impl fmt::Display for GraphQLServerError {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match *self {
-            GraphQLServerError::Canceled(_) => {
-                write!(f, "GraphQL server error (query was canceled)")
-            }
             GraphQLServerError::ClientError(ref s) => {
                 write!(f, "GraphQL server error (client error): {}", s)
             }
@@ -64,7 +41,6 @@ impl Error for GraphQLServerError {
 
     fn cause(&self) -> Option<&dyn Error> {
         match *self {
-            GraphQLServerError::Canceled(ref e) => Some(e),
             GraphQLServerError::ClientError(_) => None,
             GraphQLServerError::QueryError(ref e) => Some(e),
             GraphQLServerError::InternalError(_) => None,

graph/src/components/server/query.rs

@@ -78,6 +78,20 @@ impl QueryResult {
             _ => self.errors = self.errors.take().or(other.errors),
         }
     }
+
+    pub fn as_http_response<T: From<String>>(&self) -> http::Response<T> {
+        let status_code = http::StatusCode::OK;
+        let json =
+            serde_json::to_string(&self).expect("Failed to serialize GraphQL response to JSON");
+        http::Response::builder()
+            .status(status_code)
+            .header("Access-Control-Allow-Origin", "*")
+            .header("Access-Control-Allow-Headers", "Content-Type, User-Agent")
+            .header("Access-Control-Allow-Methods", "GET, OPTIONS, POST")
+            .header("Content-Type", "application/json")
+            .body(T::from(json))
+            .unwrap()
+    }
 }
 
 impl From<QueryExecutionError> for QueryResult {
@@ -88,6 +102,16 @@ impl From<QueryExecutionError> for QueryResult {
     }
 }
 
+impl From<QueryError> for QueryResult {
+    fn from(e: QueryError) -> Self {
+        QueryResult {
+            data: None,
+            errors: Some(vec![e]),
+            extensions: None,
+        }
+    }
+}
+
 impl From<Vec<QueryExecutionError>> for QueryResult {
     fn from(e: Vec<QueryExecutionError>) -> Self {
         QueryResult {

graph/src/data/query/result.rs

@@ -83,9 +83,7 @@ pub mod prelude {
         EthereumNetworkIdentifier, EthereumTransactionData, EthereumTrigger, LightEthereumBlock,
         LightEthereumBlockExt, ProviderEthRpcMetrics, SubgraphEthRpcMetrics,
     };
-    pub use crate::components::graphql::{
-        GraphQlRunner, QueryResultFuture, SubscriptionResultFuture,
-    };
+    pub use crate::components::graphql::{GraphQlRunner, SubscriptionResultFuture};
     pub use crate::components::link_resolver::{JsonStreamValue, JsonValueStream, LinkResolver};
     pub use crate::components::metrics::{
         aggregate::Aggregate, stopwatch::StopwatchMetrics, Collector, Counter, CounterVec, Gauge,

graph/src/lib.rs

@@ -13,8 +13,8 @@ use crate::query::execute_query;
 use crate::subscription::execute_prepared_subscription;
 use graph::data::graphql::effort::LoadManager;
 use graph::prelude::{
-    o, EthereumBlockPointer, GraphQlRunner as GraphQlRunnerTrait, Logger, Query,
-    QueryExecutionError, QueryResult, QueryResultFuture, Store, StoreError, SubgraphDeploymentId,
+    async_trait, o, EthereumBlockPointer, GraphQlRunner as GraphQlRunnerTrait, Logger, Query,
+    QueryExecutionError, QueryResult, Store, StoreError, SubgraphDeploymentId,
     SubgraphDeploymentStore, Subscription, SubscriptionError, SubscriptionResultFuture,
 };
 
@@ -107,7 +107,7 @@ where
         Ok(exts)
     }
 
-    pub fn execute(
+    fn execute(
         &self,
         query: Query,
         max_complexity: Option<u64>,
@@ -166,30 +166,30 @@ where
     }
 }
 
+#[async_trait]
 impl<S> GraphQlRunnerTrait for GraphQlRunner<S>
 where
     S: Store + SubgraphDeploymentStore,
 {
-    fn run_query(&self, query: Query) -> QueryResultFuture {
+    async fn run_query(self: Arc<Self>, query: Query) -> Arc<QueryResult> {
         self.run_query_with_complexity(
             query,
             *GRAPHQL_MAX_COMPLEXITY,
             Some(*GRAPHQL_MAX_DEPTH),
             Some(*GRAPHQL_MAX_FIRST),
         )
+        .await
     }
 
-    fn run_query_with_complexity(
+    async fn run_query_with_complexity(
         &self,
         query: Query,
         max_complexity: Option<u64>,
         max_depth: Option<u8>,
         max_first: Option<u32>,
-    ) -> QueryResultFuture {
-        Box::new(future::ok(
-            self.execute(query, max_complexity, max_depth, max_first)
-                .unwrap_or_else(|e| Arc::new(e)),
-        ))
+    ) -> Arc<QueryResult> {
+        self.execute(query, max_complexity, max_depth, max_first)
+            .unwrap_or_else(|e| Arc::new(e))
     }
 
     fn run_subscription(&self, subscription: Subscription) -> SubscriptionResultFuture {

graphql/src/runner.rs

@@ -10,14 +10,14 @@ use std::time::{Duration, Instant};
 
 use graph::prelude::{
     futures03::stream::StreamExt, futures03::FutureExt, futures03::TryFutureExt, o, slog, tokio,
-    Entity, EntityKey, EntityOperation, EthereumBlockPointer, FutureExtension, Logger, Query,
-    QueryError, QueryExecutionError, QueryResult, QueryVariables, Schema, Store,
+    Entity, EntityKey, EntityOperation, EthereumBlockPointer, FutureExtension, GraphQlRunner as _,
+    Logger, Query, QueryError, QueryExecutionError, QueryResult, QueryVariables, Schema, Store,
     SubgraphDeploymentEntity, SubgraphDeploymentId, SubgraphDeploymentStore, SubgraphManifest,
     Subscription, SubscriptionError, Value,
 };
 use graph_graphql::prelude::*;
 use test_store::{
-    execute_subgraph_query_with_complexity, execute_subgraph_query_with_deadline, return_err,
+    execute_subgraph_query_with_complexity, execute_subgraph_query_with_deadline,
     transact_entity_operations, BLOCK_ONE, GENESIS_PTR, LOAD_MANAGER, LOGGER, STORE,
 };
 
@@ -233,9 +233,11 @@ fn execute_query_document_with_variables(
     let runner = GraphQlRunner::new(&*LOGGER, STORE.clone(), LOAD_MANAGER.clone());
     let query = Query::new(Arc::new(api_test_schema()), query, variables, None);
 
-    return_err!(runner.execute(query, None, None, None))
-        .as_ref()
-        .clone()
+    graph::prelude::futures03::executor::block_on(
+        runner.run_query_with_complexity(query, None, None, None),
+    )
+    .as_ref()
+    .clone()
 }
 
 #[test]

graphql/tests/query.rs

@@ -9,12 +9,10 @@ extern crate hyper;
 extern crate serde;
 
 mod request;
-mod response;
 mod server;
 mod service;
 
 pub use self::request::GraphQLRequest;
-pub use self::response::GraphQLResponse;
 pub use self::server::GraphQLServer;
 pub use self::service::{GraphQLService, GraphQLServiceResponse};
 

server/http/src/lib.rs

@@ -43,13 +43,13 @@ impl Future for GraphQLRequest {
         // Ensure the JSON data has a "query" field
         let query_value = obj.get("query").ok_or_else(|| {
             GraphQLServerError::ClientError(String::from(
-                "The \"query\" field missing in request data",
+                "The \"query\" field is missing in request data",
             ))
         })?;
 
         // Ensure the "query" field is a string
         let query_string = query_value.as_str().ok_or_else(|| {
-            GraphQLServerError::ClientError(String::from("The\"query\" field is not a string"))
+            GraphQLServerError::ClientError(String::from("The \"query\" field is not a string"))
         })?;
 
         // Parse the "query" field of the JSON body