Migrate from 'grpc' crate to 'tonic'

## What is the goal of this PR?

We migrated the code from using 'grpc' crate to 'tonic'.

## What are the changes implemented in this PR?

Tonic is better-maintained and more active, so we should prefer it and use it across our Rust ecosystem.

Author: alexjpwalker

.gitignore

@@ -22,3 +22,6 @@
 .DS_Store
 .ijwb
 bazel-*
+target
+Cargo.lock
+Cargo.toml

BUILD

@@ -26,18 +26,20 @@ load("@vaticle_bazel_distribution//crates:rules.bzl", "assemble_crate")
 
 rust_library(
     name = "typedb_client",
-    srcs = glob(["src/**/*.rs"], exclude = glob(["src/typedb_protocol/**/*.rs"])),
+    srcs = glob(["src/**/*.rs"]),
     deps = [
-        "//typedb_protocol",
+        "@vaticle_typedb_protocol//grpc/rust:typedb_protocol",
 
+        "@vaticle_dependencies//library/crates:crossbeam",
         "@vaticle_dependencies//library/crates:futures",
-        "@vaticle_dependencies//library/crates:grpc",
         "@vaticle_dependencies//library/crates:log",
-        "@vaticle_dependencies//library/crates:protobuf",
+        "@vaticle_dependencies//library/crates:prost",
+        "@vaticle_dependencies//library/crates:tokio",
+        "@vaticle_dependencies//library/crates:tonic",
         "@vaticle_dependencies//library/crates:uuid",
     ],
     proc_macro_deps = [
-        "@vaticle_dependencies//library/crates:derivative",
+        "@vaticle_dependencies//library/crates:enum_dispatch",
     ],
     tags = [
         "crate-name=typedb-client",
@@ -52,4 +54,12 @@ assemble_crate(
     homepage = "https://github.com/vaticle/typedb-client-rust",
     license = "apache",
     repository = "https://github.com/vaticle/typedb-client-rust",
-)
+)
+
+# CI targets that are not declared in any BUILD file, but are called externally
+filegroup(
+    name = "ci",
+    data = [
+        "@vaticle_dependencies//ide/rust:sync"
+    ],
+)

WORKSPACE

@@ -33,6 +33,22 @@ load("@io_bazel_rules_kotlin//kotlin:kotlin.bzl", "kotlin_repositories", "kt_reg
 kotlin_repositories()
 kt_register_toolchains()
 
+# TODO: enable this when we load @vaticle_typeql
+## Load //builder/antlr
+#load("@vaticle_dependencies//builder/antlr:deps.bzl", antlr_deps = "deps", "antlr_version")
+#antlr_deps()
+#
+#load("@rules_antlr//antlr:lang.bzl", "JAVA")
+#load("@rules_antlr//antlr:repositories.bzl", "rules_antlr_dependencies")
+#rules_antlr_dependencies(antlr_version, JAVA)
+
+# Load //builder/grpc (required by @vaticle_typedb_protocol)
+load("@vaticle_dependencies//builder/grpc:deps.bzl", grpc_deps = "deps")
+grpc_deps()
+load("@com_github_grpc_grpc//bazel:grpc_deps.bzl",
+com_github_grpc_grpc_deps = "grpc_deps")
+com_github_grpc_grpc_deps()
+
 # Load //builder/rust
 load("@vaticle_dependencies//builder/rust:deps.bzl", rust_deps = "deps")
 rust_deps()
@@ -59,6 +75,13 @@ vaticle_dependencies_ci_pip()
 load("@vaticle_dependencies//distribution:deps.bzl", "vaticle_bazel_distribution")
 vaticle_bazel_distribution()
 
+################################
+# Load @vaticle dependencies #
+################################
+
+load("//dependencies/vaticle:repositories.bzl", "vaticle_typedb_protocol")
+vaticle_typedb_protocol()
+
 ############################
 # Load @maven dependencies #
 ############################

dependencies/ide/sync.sh

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+#
+# Copyright (C) 2022 Vaticle
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+
+bazel run @vaticle_dependencies//ide/rust:sync

dependencies/vaticle/repositories.bzl

@@ -21,5 +21,12 @@ def vaticle_dependencies():
     git_repository(
         name = "vaticle_dependencies",
         remote = "https://github.com/vaticle/dependencies",
-        commit = "e3ba4970a3bb7db5e4725c0daa2a2a154ba9b3b8", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_dependencies
+        commit = "5be6d949ca1e04e4179ea6acb3432be713b9dfb8", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_dependencies
+    )
+    
+def vaticle_typedb_protocol():
+    git_repository(
+        name = "vaticle_typedb_protocol",
+        remote = "https://github.com/vaticle/typedb-protocol",
+        commit = "78e529dc5d40221aed5cc754906246b02f3635c2", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_dependencies
     )

src/answer/mod.rs

@@ -30,7 +30,7 @@ pub struct ConceptMap {
 }
 
 impl ConceptMap {
-    pub(crate) fn from_proto(proto: typedb_protocol::answer::ConceptMap) -> Result<Self> {
+    pub(crate) fn from_proto(proto: typedb_protocol::ConceptMap) -> Result<Self> {
         let mut map = HashMap::with_capacity(proto.map.len());
         for (k, v) in proto.map {
             map.insert(k, Concept::from_proto(v)?);

src/common/error.rs

@@ -19,9 +19,11 @@
  * under the License.
  */
 
-use grpc::{Error as GrpcError, GrpcMessageError, GrpcStatus};
+// use grpc::{Error as GrpcError, GrpcMessageError, GrpcStatus};
 use std::error::Error as StdError;
 use std::fmt::{Debug, Display, Formatter};
+use tokio::sync::mpsc::error::SendError;
+use tonic::Status;
 
 // TODO: try refactoring out the lifetime by storing String instead of &str
 struct MessageTemplate<'a> {
@@ -116,6 +118,7 @@ pub struct ClientMessages<'a> {
     pub cluster_replica_not_primary: Message<'a>,
     pub cluster_token_credential_invalid: Message<'a>,
     pub session_close_failed: Message<'a>,
+    pub session_was_never_closed: Message<'a>,
 }
 
 pub struct ConceptMessages<'a> {
@@ -137,10 +140,11 @@ impl Messages<'_> {
                 unable_to_connect: Message::new(TEMPLATES.client, 5, "Unable to connect to TypeDB server."),
                 db_does_not_exist: Message::new(TEMPLATES.client, 8, "The database '{}' does not exist."),
                 missing_response_field: Message::new(TEMPLATES.client, 9, "Missing field in message received from server: '{}'."),
-                unknown_request_id: Message::new(TEMPLATES.client, 10, "Received a response with unknown request id '{}':\n{}"),
+                unknown_request_id: Message::new(TEMPLATES.client, 10, "Received a response with unknown request id '{}'"),
                 cluster_replica_not_primary: Message::new(TEMPLATES.client, 13, "The replica is not the primary replica."),
                 cluster_token_credential_invalid: Message::new(TEMPLATES.client, 16, "Invalid token credential."),
-                session_close_failed: Message::new(TEMPLATES.client, 17, "Failed to close session. It may still be open on the server, or it may already have been closed previously.")
+                session_close_failed: Message::new(TEMPLATES.client, 17, "Failed to close session. It may still be open on the server, or it may already have been closed previously."),
+                session_was_never_closed: Message::new(TEMPLATES.client, 18, "A session went out of scope without being closed. Sessions should always be closed by awaiting Session::close."),
             },
             concept: ConceptMessages {
                 invalid_concept_casting: Message::new(TEMPLATES.concept, 1, "Invalid concept conversion from '{}' to '{}'"),
@@ -153,14 +157,41 @@ pub const MESSAGES: Messages = Messages::new();
 
 #[derive(Debug)]
 pub enum Error {
-    GrpcError(String, GrpcError),
+    // GrpcError(String, GrpcError),
     Other(String),
 }
 
+impl Error {
+    pub(crate) fn new(msg: String) -> Self {
+        Error::Other(msg)
+    }
+
+    // pub(crate) fn from_grpc(source: GrpcError) -> Self {
+    //     match source {
+    //         GrpcError::Http(_) => Error::GrpcError(String::from(MESSAGES.client.unable_to_connect), source),
+    //         GrpcError::GrpcMessage(ref err) => {
+    //             // TODO: this is awkward because we use gRPC errors to represent some user errors too
+    //             if Error::is_replica_not_primary(err) { Error::new(MESSAGES.client.cluster_replica_not_primary.format(vec![])) }
+    //             else if Error::is_token_credential_invalid(err) { Error::new(MESSAGES.client.cluster_token_credential_invalid.format(vec![])) }
+    //             else { Error::GrpcError(source.to_string().replacen("grpc message error: ", "", 1), source) }
+    //         },
+    //         _ => Error::GrpcError(source.to_string(), source)
+    //     }
+    // }
+
+    // fn is_replica_not_primary(err: &GrpcMessageError) -> bool {
+    //     err.grpc_status == GrpcStatus::Internal as i32 && err.grpc_message.contains("[RPL01]")
+    // }
+    //
+    // fn is_token_credential_invalid(err: &GrpcMessageError) -> bool {
+    //     err.grpc_status == GrpcStatus::Unauthenticated as i32 && err.grpc_message.contains("[CLS08]")
+    // }
+}
+
 impl Display for Error {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         let message = match self {
-            Error::GrpcError(msg, _) => msg,
+            // Error::GrpcError(msg, _) => msg,
             Error::Other(msg) => msg
         };
         write!(f, "{}", message)
@@ -170,35 +201,26 @@ impl Display for Error {
 impl StdError for Error {
     fn source(&self) -> Option<&(dyn StdError + 'static)> {
         match self {
-            Error::GrpcError(_, source) => Some(source),
+            // Error::GrpcError(_, source) => Some(source),
             Error::Other(_) => None
         }
     }
 }
 
-impl Error {
-    pub(crate) fn new(msg: String) -> Self {
-        Error::Other(msg)
-    }
-
-    pub(crate) fn from_grpc(source: GrpcError) -> Self {
-        match source {
-            GrpcError::Http(_) => Error::GrpcError(String::from(MESSAGES.client.unable_to_connect), source),
-            GrpcError::GrpcMessage(ref err) => {
-                // TODO: this is awkward because we use gRPC errors to represent some user errors too
-                if Error::is_replica_not_primary(err) { Error::new(MESSAGES.client.cluster_replica_not_primary.format(vec![])) }
-                else if Error::is_token_credential_invalid(err) { Error::new(MESSAGES.client.cluster_token_credential_invalid.format(vec![])) }
-                else { Error::GrpcError(source.to_string().replacen("grpc message error: ", "", 1), source) }
-            },
-            _ => Error::GrpcError(source.to_string(), source)
-        }
+impl From<Status> for Error {
+    fn from(status: Status) -> Self {
+        Error::Other(status.to_string())
     }
+}
 
-    fn is_replica_not_primary(err: &GrpcMessageError) -> bool {
-        err.grpc_status == GrpcStatus::Internal as i32 && err.grpc_message.contains("[RPL01]")
+impl From<futures::channel::mpsc::SendError> for Error {
+    fn from(err: futures::channel::mpsc::SendError) -> Self {
+        Error::Other(err.to_string())
     }
+}
 
-    fn is_token_credential_invalid(err: &GrpcMessageError) -> bool {
-        err.grpc_status == GrpcStatus::Unauthenticated as i32 && err.grpc_message.contains("[CLS08]")
+impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
+    fn from(err: tokio::sync::mpsc::error::SendError<T>) -> Self {
+        Error::Other(err.to_string())
     }
 }