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

docker-compose/docker-compose.yaml

@@ -26,6 +26,8 @@ services:
       - JANUS_PROPS_TEMPLATE=inmemory
     ports:
       - "8184:8182"
+    volumes:
+      - ./janusgraph_config.yaml:/opt/janusgraph/conf/janusgraph-server.yaml
     healthcheck:
       test: ["CMD", "bin/gremlin.sh", "-e", "scripts/remote-connect.groovy"]
       interval: 10s

docker-compose/janusgraph_config.yaml

@@ -0,0 +1,54 @@
+# Copyright 2019 JanusGraph Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+host: 0.0.0.0
+port: 8182
+evaluationTimeout: 30000
+channelizer: org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer
+graphManager: org.janusgraph.graphdb.management.JanusGraphManager
+graphs: {
+  graph: conf/janusgraph-inmemory.properties
+}
+scriptEngines: {
+  gremlin-groovy: {
+    plugins: { org.janusgraph.graphdb.tinkerpop.plugin.JanusGraphGremlinPlugin: {},
+               org.apache.tinkerpop.gremlin.server.jsr223.GremlinServerGremlinPlugin: {},
+               org.apache.tinkerpop.gremlin.tinkergraph.jsr223.TinkerGraphGremlinPlugin: {},
+               org.apache.tinkerpop.gremlin.jsr223.ImportGremlinPlugin: {classImports: [java.lang.Math], methodImports: [java.lang.Math#*]},
+               org.apache.tinkerpop.gremlin.jsr223.ScriptFileGremlinPlugin: {files: [scripts/empty-sample.groovy]}}}}
+serializers:
+  - { className: org.apache.tinkerpop.gremlin.util.ser.GraphBinaryMessageSerializerV1, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
+  - { className: org.apache.tinkerpop.gremlin.util.ser.GraphBinaryMessageSerializerV1, config: { serializeResultToString: true }}
+  - { className: org.apache.tinkerpop.gremlin.util.ser.GraphSONMessageSerializerV3, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
+  # Older serialization versions for backwards compatibility:
+  - { className: org.apache.tinkerpop.gremlin.util.ser.GraphSONMessageSerializerV2, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
+  - { className: org.apache.tinkerpop.gremlin.util.ser.GraphSONMessageSerializerV1, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistryV1d0] }}
+  - { className: org.apache.tinkerpop.gremlin.util.ser.GraphSONUntypedMessageSerializerV1, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistryV1d0] }}
+processors:
+  - { className: org.apache.tinkerpop.gremlin.server.op.session.SessionOpProcessor, config: { sessionTimeout: 28800000 }}
+  - { className: org.apache.tinkerpop.gremlin.server.op.traversal.TraversalOpProcessor, config: { cacheExpirationTime: 600000, cacheMaxSize: 1000 }}
+metrics: {
+  consoleReporter: {enabled: true, interval: 180000},
+  csvReporter: {enabled: true, interval: 180000, fileName: /tmp/gremlin-server-metrics.csv},
+  jmxReporter: {enabled: true},
+  slf4jReporter: {enabled: true, interval: 180000},
+  graphiteReporter: {enabled: false, interval: 180000}}
+maxInitialLineLength: 4096
+maxHeaderSize: 8192
+maxChunkSize: 8192
+maxContentLength: 65536
+maxAccumulationBufferComponents: 1024
+resultIterationBatchSize: 64
+writeBufferLowWaterMark: 32768
+writeBufferHighWaterMark: 65536

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) => {}