MergeV, MergeE, & Option Steps

.github/workflows/coverage.yml

@@ -23,7 +23,7 @@ jobs:
       - uses: actions/checkout@v2
       - name: Starting Gremlin Servers
         run: |
-          docker-compose -f ./docker-compose/docker-compose.yaml up -d 
+          docker compose -f ./docker-compose/docker-compose.yaml up -d 
         env: 
           GREMLIN_SERVER: ${{ matrix.gremlin-server }}
 

.github/workflows/test.yml

@@ -23,7 +23,7 @@ jobs:
       - uses: actions/checkout@v2
       - name: Starting Gremlin Servers
         run: |
-          docker-compose -f ./docker-compose/docker-compose.yaml up -d 
+          docker compose -f ./docker-compose/docker-compose.yaml up -d --wait --wait-timeout 90
         env: 
           GREMLIN_SERVER: ${{ matrix.gremlin-server }}
 
@@ -42,12 +42,27 @@ jobs:
           command: fmt
           args: --all -- --check
       - 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
       - 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
+      # MergeV as a step doesn't exist in 3.5.x, so selectively run those tests
+      - 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
+      - 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

docker-compose/docker-compose.yaml

@@ -18,3 +18,16 @@ services:
     command : ["conf/gremlin-server-credentials.yaml"]
     ports:
       - "8183:8182"
+  janusgraph:
+    image: janusgraph/janusgraph:latest
+    environment:
+      - janusgraph.graph.set-vertex-id=true
+      - janusgraph.graph.allow-custom-vid-types=true
+      - JANUS_PROPS_TEMPLATE=inmemory
+    ports:
+      - "8184:8182"
+    healthcheck:
+      test: ["CMD", "bin/gremlin.sh", "-e", "scripts/remote-connect.groovy"]
+      interval: 10s
+      timeout: 30s
+      retries: 3

gremlin-client/Cargo.toml

@@ -15,14 +15,15 @@ readme = "README.md"
 [features]
 
 default = []
+merge_tests = []
 
 
 
 async_gremlin = ["futures","mobc","async-tungstenite","async-trait","url","pin-project-lite"]
 
 async_std = ["async-std-runtime"]
-tokio-runtime = ["async_gremlin","tokio","mobc/tokio","async-tungstenite/tokio-runtime","async-tungstenite/tokio-native-tls","tokio-native-tls","tokio-stream"]
-async-std-runtime = ["async_gremlin","async-std","async-tungstenite/async-std-runtime","async-tungstenite/async-tls","mobc/async-std","async-tls","rustls","webpki"]
+tokio-runtime = ["async_gremlin","tokio","async-tungstenite/tokio-runtime","async-tungstenite/tokio-native-tls","tokio-native-tls","tokio-stream"]
+async-std-runtime = ["async_gremlin","async-std","async-tungstenite/async-std-runtime","async-tungstenite/async-tls","tokio/sync", "mobc/async-std","async-tls","rustls","webpki"]
 
 derive = ["gremlin-derive"] 
 
@@ -57,7 +58,7 @@ thiserror = "1.0.20"
 
 
 
-mobc = {version = "0.7", optional = true, default-features=false, features = ["unstable"] }
+mobc = {version = "0.8", optional = true }
 url =  {version = "2.1.0", optional = true}
 futures = { version = "0.3.1", optional = true}
 pin-project-lite = { version = "0.2", optional = true}

gremlin-client/src/aio/client.rs

@@ -8,7 +8,7 @@
 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;
@@ -59,6 +59,9 @@
         let pool = Pool::builder()
             .get_timeout(opts.pool_get_connection_timeout)
             .max_open(pool_size as u64)
+            .health_check_interval(opts.pool_healthcheck_interval)
+            //Makes max idle connections equal to max open, matching the behavior of the sync pool r2d2
+            .max_idle(0)
             .build(manager);
 
         Ok(GremlinClient {
@@ -179,7 +182,7 @@
 
                         args.insert(
                             String::from("sasl"),
                             GValue::String(encode(&format!("\0{}\0{}", c.username, c.password))),
                         );
 
                         let args = self.options.serializer.write(&GValue::from(args))?;

gremlin-client/src/aio/connection.rs

@@ -1,4 +1,4 @@
 use crate::{GremlinError, GremlinResult, WebSocketOptions};
 
 use crate::connection::ConnectionOptions;
 
@@ -21,6 +21,7 @@
     pub use tokio_native_tls::TlsStream;
 }
 
+use futures::TryFutureExt;
 #[cfg(feature = "tokio-runtime")]
 use tokio_use::*;
 
@@ -89,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())
         }
     }
@@ -144,6 +145,7 @@
                 tls::connector(&opts),
                 websocket_config,
             )
+            .map_err(|e| Arc::new(e))
             .await?
         };
         #[cfg(feature = "tokio-runtime")]
@@ -153,6 +155,7 @@
                 tls::connector(&opts),
                 websocket_config,
             )
+            .map_err(|e| Arc::new(e))
             .await?
         };
 
@@ -190,6 +193,18 @@
             .await
             .expect("It should contain the response")
             .map(|r| (r, receiver))
+            .map_err(|e| {
+                //If there's been an websocket layer error, mark the connection as invalid
+                match e {
+                    GremlinError::WebSocket(_)
+                    | GremlinError::WebSocketAsync(_)
+                    | GremlinError::WebSocketPoolAsync(_) => {
+                        self.valid = false;
+                    }
+                    _ => {}
+                }
+                e
+            })
     }
 
     pub fn is_valid(&self) -> bool {
@@ -222,7 +237,7 @@
                         if let Err(e) = sink.send(Message::Binary(msg.2)).await {
                             let mut sender = guard.remove(&msg.1).unwrap();
                             sender
-                                .send(Err(GremlinError::from(e)))
+                                .send(Err(GremlinError::from(Arc::new(e))))
                                 .await
                                 .expect("Failed to send error");
                         }
@@ -257,8 +272,9 @@
             match stream.next().await {
                 Some(Err(error)) => {
                     let mut guard = requests.lock().await;
+                    let error = Arc::new(error);
                     for s in guard.values_mut() {
-                        match s.send(Err(GremlinError::from(&error))).await {
+                        match s.send(Err(error.clone().into())).await {
                             Ok(_r) => {}
                             Err(_e) => {}
                         }

gremlin-client/src/aio/mod.rs

@@ -3,7 +3,6 @@ pub(crate) mod connection;
 pub(crate) mod pool;
 mod result;
 
-mod error;
 pub(crate) mod process;
 pub use client::GremlinClient;
 pub use process::traversal::AsyncTerminator;

gremlin-client/src/aio/process/traversal/mod.rs

@@ -6,6 +6,7 @@ use crate::GremlinResult;
 use core::task::Context;
 use core::task::Poll;
 use futures::Stream;
+use futures::StreamExt;
 use std::marker::PhantomData;
 use std::pin::Pin;
 
@@ -29,6 +30,17 @@ impl<T> RemoteTraversalStream<T> {
         }
     }
 }
+
+impl RemoteTraversalStream<crate::structure::Null> {
+    pub async fn iterate(&mut self) -> GremlinResult<()> {
+        while let Some(response) = self.next().await {
+            //consume the entire stream, returning any errors
+            response?;
+        }
+        Ok(())
+    }
+}
+
 impl<T: FromGValue> Stream for RemoteTraversalStream<T> {
     type Item = GremlinResult<T>;
 

gremlin-client/src/connection.rs

@@ -1,4 +1,4 @@
-use std::{net::TcpStream, time::Duration};
+use std::{net::TcpStream, sync::Arc, time::Duration};
 
 use crate::{GraphSON, GremlinError, GremlinResult};
 use native_tls::TlsConnector;
@@ -62,12 +62,13 @@
 
     fn send(&mut self, payload: Vec<u8>) -> GremlinResult<()> {
         self.0
             .write_message(Message::Binary(payload))
+            .map_err(|e| Arc::new(e))
             .map_err(GremlinError::from)
     }
 
     fn recv(&mut self) -> GremlinResult<Vec<u8>> {
-        match self.0.read_message()? {
+        match self.0.read_message().map_err(|e| Arc::new(e))? {
             Message::Binary(binary) => Ok(binary),
             _ => unimplemented!(),
         }
@@ -120,6 +121,16 @@
         self
     }
 
+    /// Only applicable to async client. By default a connection is checked on each return to the pool (None)
+    /// This allows setting an interval of how often it is checked on return.
+    pub fn pool_healthcheck_interval(
+        mut self,
+        pool_healthcheck_interval: Option<Duration>,
+    ) -> Self {
+        self.0.pool_healthcheck_interval = pool_healthcheck_interval;
+        self
+    }
+
     /// Both the sync and async pool providers use a default of 30 seconds,
     /// Async pool interprets `None` as no timeout. Sync pool maps `None` to the default value
     pub fn pool_connection_timeout(mut self, pool_connection_timeout: Option<Duration>) -> Self {
@@ -170,6 +181,7 @@
     pub(crate) host: String,
     pub(crate) port: u16,
     pub(crate) pool_size: u32,
+    pub(crate) pool_healthcheck_interval: Option<Duration>,
     pub(crate) pool_get_connection_timeout: Option<Duration>,
     pub(crate) credentials: Option<Credentials>,
     pub(crate) ssl: bool,
@@ -254,6 +266,7 @@
             port: 8182,
             pool_size: 10,
             pool_get_connection_timeout: Some(Duration::from_secs(30)),
+            pool_healthcheck_interval: None,
             credentials: None,
             ssl: false,
             tls_options: None,

gremlin-client/src/conversion.rs

@@ -1,6 +1,6 @@
 use crate::{
     process::traversal::Bytecode,
-    structure::{TextP, P as Predicate},
+    structure::{Null, TextP, P as Predicate},
     Edge, GKey, GValue, GremlinError, GremlinResult, IntermediateRepr, List, Map, Metric, Path,
     Property, Token, TraversalExplanation, TraversalMetrics, Vertex, VertexProperty, GID,
 };
@@ -134,9 +134,23 @@ impl_from_gvalue!(IntermediateRepr, GValue::IntermediateRepr);
 impl_from_gvalue!(chrono::DateTime<chrono::Utc>, GValue::Date);
 impl_from_gvalue!(Traverser, GValue::Traverser);
 
+impl FromGValue for Null {
+    fn from_gvalue(v: GValue) -> GremlinResult<Self> {
+        match v {
+            GValue::Null => Ok(crate::structure::Null {}),
+            _ => Err(GremlinError::Cast(format!(
+                "Cannot convert {:?} to {}",
+                v,
+                stringify!($t)
+            ))),
+        }
+    }
+}
+
 impl FromGValue for GKey {
     fn from_gvalue(v: GValue) -> GremlinResult<GKey> {
         match v {
+            GValue::Direction(d) => Ok(GKey::Direction(d)),
             GValue::String(s) => Ok(GKey::String(s)),
             GValue::Token(s) => Ok(GKey::String(s.value().clone())),
             GValue::Vertex(s) => Ok(GKey::Vertex(s)),

gremlin-client/src/error.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use crate::structure::GValue;
 
 use thiserror::Error;
@@ -34,7 +36,10 @@ pub enum GremlinError {
 
     #[cfg(feature = "async_gremlin")]
     #[error(transparent)]
-    WebSocketAsync(#[from] async_tungstenite::tungstenite::Error),
+    WebSocketAsync(#[from] Arc<async_tungstenite::tungstenite::Error>),
+    #[cfg(feature = "async_gremlin")]
+    #[error(transparent)]
+    WebSocketPoolAsync(#[from] Arc<mobc::Error<GremlinError>>),
     #[cfg(feature = "async_gremlin")]
     #[error(transparent)]
     ChannelSend(#[from] futures::channel::mpsc::SendError),
@@ -47,10 +52,7 @@ impl From<mobc::Error<GremlinError>> for GremlinError {
     fn from(e: mobc::Error<GremlinError>) -> GremlinError {
         match e {
             mobc::Error::Inner(e) => e,
-            mobc::Error::BadConn => {
-                GremlinError::Generic(String::from("Async pool bad connection"))
-            }
-            mobc::Error::Timeout => GremlinError::Generic(String::from("Async pool timeout")),
+            other => GremlinError::WebSocketPoolAsync(Arc::new(other)),
         }
     }
 }

gremlin-client/src/io/mod.rs

@@ -5,7 +5,7 @@ mod serializer_v3;
 
 use crate::conversion::ToGValue;
 use crate::process::traversal::{Order, Scope};
-use crate::structure::{Cardinality, GValue, T};
+use crate::structure::{Cardinality, Direction, GValue, Merge, T};
 use serde_json::{json, Map, Value};
 use std::string::ToString;
 
@@ -216,7 +216,38 @@ impl GraphSON {
                     "@value" : v
                 }))
             }
-
+            (_, GValue::Merge(merge)) => {
+                let merge_option = match merge {
+                    Merge::OnCreate => "onCreate",
+                    Merge::OnMatch => "onMatch",
+                    Merge::OutV => "outV",
+                    Merge::InV => "inV",
+                };
+                Ok(json!({
+                    "@type" : "g:Merge",
+                    "@value" : merge_option
+                }))
+            }
+            (_, GValue::Direction(direction)) => {
+                let direction = match direction {
+                    Direction::Out | Direction::From => "OUT",
+                    Direction::In | Direction::To => "IN",
+                };
+                Ok(json!({
+                    "@type" : "g:Direction",
+                    "@value" : direction,
+                }))
+            }
+            (_, GValue::Column(column)) => {
+                let column = match column {
+                    crate::structure::Column::Keys => "keys",
+                    crate::structure::Column::Values => "values",
+                };
+                Ok(json!({
+                    "@type" : "g:Column",
+                    "@value" : column,
+                }))
+            }
             (_, _) => panic!("Type {:?} not supported.", value),
         }
     }