Switch the HTTP engine to a binary protocol (#1751)

lib/src/api/engine/any/native.rs

@@ -6,6 +6,8 @@ use crate::api::conn::Router;
 #[allow(unused_imports)] // used by the DB engines
 use crate::api::engine;
 use crate::api::engine::any::Any;
+#[cfg(feature = "protocol-http")]
+use crate::api::engine::remote::http;
 use crate::api::err::Error;
 use crate::api::opt::from_value;
 use crate::api::opt::Endpoint;
@@ -21,12 +23,6 @@ use crate::api::Surreal;
 use flume::Receiver;
 use once_cell::sync::OnceCell;
 #[cfg(feature = "protocol-http")]
-use reqwest::header::HeaderMap;
-#[cfg(feature = "protocol-http")]
-use reqwest::header::HeaderValue;
-#[cfg(feature = "protocol-http")]
-use reqwest::header::ACCEPT;
-#[cfg(feature = "protocol-http")]
 use reqwest::ClientBuilder;
 use serde::de::DeserializeOwned;
 use std::collections::HashSet;
@@ -105,8 +101,7 @@ impl Connection for Any {
 				"http" | "https" => {
 					features.insert(ExtraFeatures::Auth);
 					features.insert(ExtraFeatures::Backup);
-					let mut headers = HeaderMap::new();
-					headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
+					let headers = http::default_headers();
 					#[allow(unused_mut)]
 					let mut builder = ClientBuilder::new().default_headers(headers);
 					#[cfg(any(feature = "native-tls", feature = "rustls"))]

lib/src/api/engine/remote/http/mod.rs

@@ -23,8 +23,6 @@ use crate::api::Response as QueryResponse;
 use crate::api::Result;
 use crate::api::Surreal;
 use crate::opt::IntoEndpoint;
-use crate::sql;
-use crate::sql::to_value;
 use crate::sql::Array;
 use crate::sql::Strand;
 use crate::sql::Value;
@@ -33,7 +31,6 @@ use futures::TryStreamExt;
 use indexmap::IndexMap;
 use reqwest::header::HeaderMap;
 use reqwest::header::HeaderValue;
-#[cfg(not(target_arch = "wasm32"))]
 use reqwest::header::ACCEPT;
 #[cfg(not(target_arch = "wasm32"))]
 use reqwest::header::CONTENT_TYPE;
@@ -105,6 +102,12 @@ impl Surreal<Client> {
 	}
 }
 
+pub(crate) fn default_headers() -> HeaderMap {
+	let mut headers = HeaderMap::new();
+	headers.insert(ACCEPT, HeaderValue::from_static("application/cork"));
+	headers
+}
+
 #[derive(Debug)]
 enum Auth {
 	Basic {
@@ -136,10 +139,14 @@ impl Authenticate for RequestBuilder {
 }
 
 #[derive(Debug, Deserialize)]
+#[allow(dead_code)]
 struct HttpQueryResponse {
+	time: String,
 	status: Status,
-	result: Option<serde_json::Value>,
-	detail: Option<String>,
+	#[serde(default)]
+	result: Value,
+	#[serde(default)]
+	detail: String,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -149,42 +156,46 @@ struct Root {
 }
 
 #[derive(Debug, Deserialize)]
+#[allow(dead_code)]
 struct AuthResponse {
+	code: u16,
+	details: String,
+	#[serde(default)]
 	token: Option<String>,
 }
 
 async fn submit_auth(request: RequestBuilder) -> Result<Value> {
 	let response = request.send().await?.error_for_status()?;
-	let text = response.text().await?;
-	info!(target: LOG, "Response {text}");
-	let value = sql::json(&text)?;
-	let response: AuthResponse = from_value(value)?;
+	let bytes = response.bytes().await?;
+	let response: AuthResponse =
+		msgpack::from_slice(&bytes).map_err(|error| Error::ResponseFromBinary {
+			binary: bytes.to_vec(),
+			error,
+		})?;
 	Ok(response.token.into())
 }
 
 async fn query(request: RequestBuilder) -> Result<QueryResponse> {
 	info!(target: LOG, "{request:?}");
 	let response = request.send().await?.error_for_status()?;
-	let text = response.text().await?;
-	info!(target: LOG, "Response {text}");
-	let value = sql::json(&text)?;
-	let responses: Vec<HttpQueryResponse> = from_value(value)?;
+	let bytes = response.bytes().await?;
+	let responses: Vec<HttpQueryResponse> =
+		msgpack::from_slice(&bytes).map_err(|error| Error::ResponseFromBinary {
+			binary: bytes.to_vec(),
+			error,
+		})?;
 	let mut map = IndexMap::<usize, QueryResult>::with_capacity(responses.len());
 	for (index, response) in responses.into_iter().enumerate() {
 		match response.status {
 			Status::Ok => {
-				if let Some(value) = response.result {
-					match to_value(value)? {
-						Value::Array(Array(array)) => map.insert(index, Ok(array)),
-						Value::None | Value::Null => map.insert(index, Ok(vec![])),
-						value => map.insert(index, Ok(vec![value])),
-					};
-				}
+				match response.result {
+					Value::Array(Array(array)) => map.insert(index, Ok(array)),
+					Value::None | Value::Null => map.insert(index, Ok(vec![])),
+					value => map.insert(index, Ok(vec![value])),
+				};
 			}
 			Status::Err => {
-				if let Some(error) = response.detail {
-					map.insert(index, Err(Error::Query(error).into()));
-				}
+				map.insert(index, Err(Error::Query(response.detail).into()));
 			}
 		}
 	}
@@ -281,10 +292,15 @@ async fn import(request: RequestBuilder, path: PathBuf) -> Result<Value> {
 		}
 		.into());
 	}
-	// ideally we should pass `file` directly into the body
-	// but currently that results in
-	// "HTTP status client error (405 Method Not Allowed) for url"
-	request.body(contents).send().await?.error_for_status()?;
+	request
+		.header(ACCEPT, "application/octet-stream")
+		// ideally we should pass `file` directly into the body
+		// but currently that results in
+		// "HTTP status client error (405 Method Not Allowed) for url"
+		.body(contents)
+		.send()
+		.await?
+		.error_for_status()?;
 	Ok(Value::None)
 }
 

lib/src/api/engine/remote/http/native.rs

@@ -19,8 +19,6 @@ use futures::StreamExt;
 use indexmap::IndexMap;
 use once_cell::sync::OnceCell;
 use reqwest::header::HeaderMap;
-use reqwest::header::HeaderValue;
-use reqwest::header::ACCEPT;
 use reqwest::ClientBuilder;
 use serde::de::DeserializeOwned;
 use std::collections::HashSet;
@@ -45,8 +43,7 @@ impl Connection for Client {
 		capacity: usize,
 	) -> Pin<Box<dyn Future<Output = Result<Surreal<Self>>> + Send + Sync + 'static>> {
 		Box::pin(async move {
-			let mut headers = HeaderMap::new();
-			headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
+			let headers = super::default_headers();
 
 			#[allow(unused_mut)]
 			let mut builder = ClientBuilder::new().default_headers(headers);

lib/src/api/engine/remote/http/wasm.rs

@@ -18,8 +18,6 @@ use futures::StreamExt;
 use indexmap::IndexMap;
 use once_cell::sync::OnceCell;
 use reqwest::header::HeaderMap;
-use reqwest::header::HeaderValue;
-use reqwest::header::ACCEPT;
 use reqwest::ClientBuilder;
 use serde::de::DeserializeOwned;
 use std::collections::HashSet;
@@ -122,8 +120,7 @@ impl Connection for Client {
 }
 
 async fn client(base_url: &Url) -> Result<reqwest::Client> {
-	let mut headers = HeaderMap::new();
-	headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
+	let headers = super::default_headers();
 	let builder = ClientBuilder::new().default_headers(headers);
 	let client = builder.build()?;
 	let health = base_url.join(Method::Health.as_str())?;