Implement graph binary protocol

.github/workflows/test.yml

@@ -52,29 +52,29 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: --manifest-path gremlin-client/Cargo.toml  --features=tokio-runtime
+          args: --manifest-path gremlin-client/Cargo.toml  --features=tokio-runtime,derive
       - name: Run cargo test with async-std
         if: matrix.gremlin-server == '3.5.7'
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: --manifest-path gremlin-client/Cargo.toml --features=async-std-runtime
+          args: --manifest-path gremlin-client/Cargo.toml --features=async-std-runtime,derive
       # MergeV as a step doesn't exist in 3.5.x, so selectively run those tests
       - name: Run cargo test with blocking client
         if: matrix.gremlin-server != '3.5.7'
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: --manifest-path gremlin-client/Cargo.toml  --features=merge_tests
+          args: --manifest-path gremlin-client/Cargo.toml  --features=merge_tests,derive
       - name: Run cargo test with tokio
         if: matrix.gremlin-server != '3.5.7'
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: --manifest-path gremlin-client/Cargo.toml  --features=tokio-runtime,merge_tests
+          args: --manifest-path gremlin-client/Cargo.toml  --features=tokio-runtime,merge_tests,derive
       - name: Run cargo test with async-std
         if: matrix.gremlin-server != '3.5.7'
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: --manifest-path gremlin-client/Cargo.toml --features=async-std-runtime,merge_tests
+          args: --manifest-path gremlin-client/Cargo.toml --features=async-std-runtime,merge_tests,derive

gremlin-cli/src/actions/connect.rs

@@ -1,6 +1,6 @@
 use crate::{actions::Action, command::Command, context::GremlinContext};
 use futures::FutureExt;
-use gremlin_client::{aio::GremlinClient, ConnectionOptions, GraphSON, TlsOptions};
+use gremlin_client::{aio::GremlinClient, ConnectionOptions, IoProtocol, TlsOptions};
 use std::str::FromStr;
 use structopt::StructOpt;
 
@@ -30,11 +30,11 @@ impl FromStr for Serializer {
     }
 }
 
-impl From<Serializer> for GraphSON {
+impl From<Serializer> for IoProtocol {
     fn from(serializer: Serializer) -> Self {
         match serializer {
-            Serializer::GraphSONV2 => GraphSON::V2,
-            Serializer::GraphSONV3 => GraphSON::V3,
+            Serializer::GraphSONV2 => IoProtocol::GraphSONV2,
+            Serializer::GraphSONV3 => IoProtocol::GraphSONV3,
         }
     }
 }

gremlin-client/Cargo.toml

@@ -69,8 +69,10 @@ tokio = { version = "1", optional=true, features = ["full"] }
 features = ["serde", "v4"]
 version = "1.1.2"
 
-
-
+[dev-dependencies]
+rstest = "0.23.0"
+rstest_reuse = "0.7.0"
+serial_test = "3.1.1"
 
 [[example]]
 name = "traversal_async"

gremlin-client/src/aio/client.rs

@@ -1,18 +1,14 @@
 use crate::aio::pool::GremlinConnectionManager;
 use crate::aio::GResultSet;
-use crate::io::GraphSON;
-use crate::message::{
-    message_with_args, message_with_args_and_uuid, message_with_args_v2, Message,
-};
 use crate::process::traversal::Bytecode;
 use crate::GValue;
 use crate::ToGValue;
 use crate::{ConnectionOptions, GremlinError, GremlinResult};
 use base64::encode;
 use futures::future::{BoxFuture, FutureExt};
 use mobc::{Connection, Pool};
-use serde::Serialize;
 use std::collections::{HashMap, VecDeque};
+use uuid::Uuid;
 
 pub type SessionedClient = GremlinClient;
 
@@ -21,18 +17,14 @@
         if let Some(session_name) = self.session.take() {
             let mut args = HashMap::new();
             args.insert(String::from("session"), GValue::from(session_name.clone()));
-            let args = self.options.serializer.write(&GValue::from(args))?;
-
-            let processor = "session".to_string();
-
-            let message = match self.options.serializer {
-                GraphSON::V2 => message_with_args_v2(String::from("close"), processor, args),
-                GraphSON::V3 => message_with_args(String::from("close"), processor, args),
-            };
+            let (id, message) = self
+                .options
+                .serializer
+                .build_message("close", "session", args, None)?;
 
             let conn = self.pool.get().await?;
 
-            self.send_message_new(conn, message).await
+            self.send_message_new(conn, id, message).await
         } else {
             Err(GremlinError::Generic("No session to close".to_string()))
         }
@@ -131,47 +123,38 @@
             args.insert(String::from("session"), GValue::from(session_name.clone()));
         }
 
-        let args = self.options.serializer.write(&GValue::from(args))?;
-
         let processor = if self.session.is_some() {
-            "session".to_string()
+            "session"
         } else {
-            String::default()
+            ""
         };
 
-        let message = match self.options.serializer {
-            GraphSON::V2 => message_with_args_v2(String::from("eval"), processor, args),
-            GraphSON::V3 => message_with_args(String::from("eval"), processor, args),
-        };
+        let (id, message) = self
+            .options
+            .serializer
+            .build_message("eval", processor, args, None)?;
 
         let conn = self.pool.get().await?;
 
-        self.send_message_new(conn, message).await
+        self.send_message_new(conn, id, message).await
     }
 
-    pub(crate) fn send_message_new<'a, T: Serialize>(
+    pub(crate) fn send_message_new<'a>(
         &'a self,
         mut conn: Connection<GremlinConnectionManager>,
-        msg: Message<T>,
+        id: Uuid,
+        binary: Vec<u8>,
     ) -> BoxFuture<'a, GremlinResult<GResultSet>> {
-        let id = msg.id().clone();
-        let message = self.build_message(msg).unwrap();
-
         async move {
-            let content_type = self.options.serializer.content_type();
-            let payload = String::from("") + content_type + &message;
-            let mut binary = payload.into_bytes();
-            binary.insert(0, content_type.len() as u8);
-
             let (response, receiver) = conn.send(id, binary).await?;
 
             let (response, results) = match response.status.code {
                 200 | 206 => {
-                    let results: VecDeque<GValue> = self
-                        .options
-                        .deserializer
-                        .read(&response.result.data)?
-                        .map(|v| v.into())
+                    let results: VecDeque<GValue> = response
+                        .result
+                        .data
+                        .clone()
+                        .map(Into::into)
                         .unwrap_or_else(VecDeque::new);
                     Ok((response, results))
                 }
@@ -182,18 +165,21 @@
 
                         args.insert(
                             String::from("sasl"),
                             GValue::String(encode(&format!("\0{}\0{}", c.username, c.password))),
                         );
 
-                        let args = self.options.serializer.write(&GValue::from(args))?;
-                        let message = message_with_args_and_uuid(
-                            String::from("authentication"),
-                            String::from("traversal"),
-                            response.request_id,
+                        let (id, message) = self.options.serializer.build_message(
+                            "authentication",
+                            "traversal",
                             args,
-                        );
-
-                        return self.send_message_new(conn, message).await;
+                            Some(
+                                response
+                                    .request_id
+                                    .expect("Auth challenge requires response id"),
+                            ),
+                        )?;
+
+                        return self.send_message_new(conn, id, message).await;
                     }
                     None => Err(GremlinError::Request((
                         response.status.code,
@@ -229,16 +215,13 @@
 
         args.insert(String::from("aliases"), GValue::from(aliases));
 
-        let args = self.options.serializer.write(&GValue::from(args))?;
-
-        let message = message_with_args(String::from("bytecode"), String::from("traversal"), args);
+        let (id, message) =
+            self.options
+                .serializer
+                .build_message("bytecode", "traversal", args, None)?;
 
         let conn = self.pool.get().await?;
 
-        self.send_message_new(conn, message).await
-    }
-
-    fn build_message<T: Serialize>(&self, msg: Message<T>) -> GremlinResult<String> {
-        serde_json::to_string(&msg).map_err(GremlinError::from)
+        self.send_message_new(conn, id, message).await
     }
 }

gremlin-client/src/aio/connection.rs

@@ -1,4 +1,4 @@
-use crate::{GremlinError, GremlinResult, WebSocketOptions};
+use crate::{GremlinError, GremlinResult, IoProtocol};
 
 use crate::connection::ConnectionOptions;
 
@@ -90,7 +90,7 @@
             _scts: &mut dyn Iterator<Item = &[u8]>,
             _ocsp_response: &[u8],
             _now: SystemTime,
         ) -> Result<rustls::client::ServerCertVerified, rustls::TLSError> {
             Ok(rustls::client::ServerCertVerified::assertion())
         }
     }
@@ -165,7 +165,7 @@
 
         sender_loop(sink, requests.clone(), receiver);
 
-        receiver_loop(stream, requests.clone(), sender.clone());
+        receiver_loop(stream, requests.clone(), sender.clone(), opts.deserializer);
 
         Ok(Conn {
             sender,
@@ -266,6 +266,7 @@
     mut stream: SplitStream<WSStream>,
     requests: Arc<Mutex<HashMap<Uuid, Sender<GremlinResult<Response>>>>>,
     mut sender: Sender<Cmd>,
+    deserializer: IoProtocol,
 ) {
     task::spawn(async move {
         loop {
@@ -283,10 +284,24 @@
                 }
                 Some(Ok(item)) => match item {
                     Message::Binary(data) => {
-                        let response: Response = serde_json::from_slice(&data).unwrap();
+                        let response = deserializer
+                            .read_response(data)
+                            .expect("Unable to parse message");
                         let mut guard = requests.lock().await;
+
+                        //GraphBinary permits a null response request id, so in lieu of a request id assume
+                        //a single entry in the requests to be the one we should respond to given connection
+                        //multiplexing isn't currently implemented
+                        let request_id = response.request_id.unwrap_or_else(|| {
+                            if guard.len() == 1 {
+                                guard.keys().next().expect("Should have had only 1 key").clone()
+                            } else {
+                                panic!("Request response without request id was received, but there isn't only 1 request currently submitted");
+                            }
+                        });
+
                         if response.status.code != 206 {
-                            let item = guard.remove(&response.request_id);
+                            let item = guard.remove(&request_id);
                             drop(guard);
                             if let Some(mut s) = item {
                                 match s.send(Ok(response)).await {
@@ -295,7 +310,7 @@
                                 };
                             }
                         } else {
-                            let item = guard.get_mut(&response.request_id);
+                            let item = guard.get_mut(&request_id);
                             if let Some(s) = item {
                                 match s.send(Ok(response)).await {
                                     Ok(_r) => {}

gremlin-client/src/aio/pool.rs

@@ -3,10 +3,9 @@
 use crate::aio::connection::Conn;
 use crate::connection::ConnectionOptions;
 use crate::error::GremlinError;
-use crate::message::{message_with_args, message_with_args_and_uuid, message_with_args_v2};
-use crate::{GValue, GraphSON};
+use crate::GValue;
 use async_trait::async_trait;
 use base64::encode;
 use std::collections::HashMap;
 
 #[derive(Debug)]
@@ -40,23 +39,13 @@
             String::from("language"),
             GValue::String(String::from("gremlin-groovy")),
         );
-        let args = self.options.serializer.write(&GValue::from(args))?;
 
-        let message = match self.options.serializer {
-            GraphSON::V2 => message_with_args_v2(String::from("eval"), String::default(), args),
-            GraphSON::V3 => message_with_args(String::from("eval"), String::default(), args),
-        };
+        let (id, message) = self
+            .options
+            .serializer
+            .build_message("eval", "", args, None)?;
 
-        let id = message.id().clone();
-        let msg = serde_json::to_string(&message).map_err(GremlinError::from)?;
-
-        let content_type = self.options.serializer.content_type();
-
-        let payload = String::from("") + content_type + &msg;
-        let mut binary = payload.into_bytes();
-        binary.insert(0, content_type.len() as u8);
-
-        let (response, _receiver) = conn.send(id, binary).await?;
+        let (response, _receiver) = conn.send(id, message).await?;
 
         match response.status.code {
             200 | 206 => Ok(conn),
@@ -70,25 +59,17 @@
                         GValue::String(encode(&format!("\0{}\0{}", c.username, c.password))),
                     );
 
-                    let args = self.options.serializer.write(&GValue::from(args))?;
-                    let message = message_with_args_and_uuid(
-                        String::from("authentication"),
-                        String::from("traversal"),
-                        response.request_id,
+                    let (id, message) = self.options.serializer.build_message(
+                        "authentication",
+                        "traversal",
                         args,
-                    );
-
-                    let id = message.id().clone();
-                    let msg = serde_json::to_string(&message).map_err(GremlinError::from)?;
-
-                    let content_type = self.options.serializer.content_type();
-                    let payload = String::from("") + content_type + &msg;
-
-                    let mut binary = payload.into_bytes();
-                    binary.insert(0, content_type.len() as u8);
-
-                    let (response, _receiver) = conn.send(id, binary).await?;
-
+                        Some(
+                            response
+                                .request_id
+                                .expect("Auth challenge requires response id"),
+                        ),
+                    )?;
+                    let (response, _receiver) = conn.send(id, message).await?;
                     match response.status.code {
                         200 | 206 => Ok(conn),
                         204 => Ok(conn),

gremlin-client/src/aio/result.rs

@@ -59,14 +59,12 @@ impl Stream for GResultSet {
                     if this.response.status.code == 206 {
                         match futures::ready!(this.receiver.as_mut().poll_next(cx)) {
                             Some(Ok(response)) => {
-                                let results: VecDeque<GValue> = this
-                                    .client
-                                    .options
-                                    .serializer
-                                    .read(&response.result.data)?
-                                    .map(|v| v.into())
+                                let results: VecDeque<GValue> = response
+                                    .result
+                                    .data
+                                    .clone()
+                                    .map(Into::into)
                                     .unwrap_or_else(VecDeque::new);
-
                                 *this.results = results;
                                 *this.response = response;
                             }