Remote execution uses tower-grpc to start executions (pantsbuild#7049)

grpcio is still used both for GetOperation requests, and for all fs
operations. It won't be for long, though... :)

src/rust/engine/process_execution/Cargo.toml

@@ -13,18 +13,27 @@ bytes = "0.4.5"
 digest = "0.8"
 fs = { path = "../fs" }
 futures = "^0.1.16"
+# TODO: Switch to a release once https://github.com/alexcrichton/futures-timer/pull/11 and https://github.com/alexcrichton/futures-timer/pull/12 merge
+futures-timer = { git = "https://github.com/pantsbuild/futures-timer", rev = "0b747e565309a58537807ab43c674d8951f9e5a0" }
 grpcio = { git = "https://github.com/pantsbuild/grpc-rs.git", rev = "4dfafe9355dc996d7d0702e7386a6fedcd9734c0", default_features = false, features = ["protobuf-codec"] }
+h2 = "0.1.13"
 hashing = { path = "../hashing" }
+http = "0.1"
 log = "0.4"
+parking_lot = "0.6"
 protobuf = { version = "2.0.4", features = ["with-bytes"] }
 resettable = { path = "../resettable" }
 sha2 = "0.8"
 tempfile = "3"
-# TODO: Switch to a release once https://github.com/alexcrichton/futures-timer/pull/11 and https://github.com/alexcrichton/futures-timer/pull/12 merge
-futures-timer = { git = "https://github.com/pantsbuild/futures-timer", rev = "0b747e565309a58537807ab43c674d8951f9e5a0" }
 time = "0.1.40"
+tokio = "0.1.14"
 tokio-codec = "0.1"
+tokio-connect = { git = "https://github.com/pantsbuild/tokio-connect.git", rev = "f7ad1ca437973d6e24037ac6f7d5ef1013833c0b" }
 tokio-process = "0.2.1"
+tower-grpc = { git = "https://github.com/pantsbuild/tower-grpc.git", rev = "ef19f2e1715f415ecb699e8f17f5845ad2b45daf" }
+tower-h2 = { git = "https://github.com/pantsbuild/tower-h2.git", rev = "44b0efb4983b769283efd5b2a3bc3decbf7c33de" }
+tower-http = { git = "https://github.com/pantsbuild/tower-http.git", rev = "56049ee7f31d4f6c549f5d1d5fbbfd7937df3d00" }
+tower-util = { git = "https://github.com/pantsbuild/tower.git", rev = "7b61c1fc1992c1df684fd3f179644ef0ca9bfa4c" }
 
 [dev-dependencies]
 mock = { path = "../testutil/mock" }

src/rust/engine/process_execution/bazel_protos/src/conversions.rs

@@ -18,19 +18,31 @@ impl<'a> From<&'a hashing::Digest> for crate::build::bazel::remote::execution::v
   }
 }
 
-impl<'a> From<&'a super::remote_execution::Digest> for Result<hashing::Digest, String> {
-  fn from(d: &super::remote_execution::Digest) -> Self {
+impl<'a> From<&'a crate::remote_execution::Digest> for Result<hashing::Digest, String> {
+  fn from(d: &crate::remote_execution::Digest) -> Self {
     hashing::Fingerprint::from_hex_string(d.get_hash())
       .map_err(|err| format!("Bad fingerprint in Digest {:?}: {:?}", d.get_hash(), err))
       .map(|fingerprint| hashing::Digest(fingerprint, d.get_size_bytes() as usize))
   }
 }
 
+impl<'a> From<&'a crate::build::bazel::remote::execution::v2::Digest>
+  for Result<hashing::Digest, String>
+{
+  fn from(d: &crate::build::bazel::remote::execution::v2::Digest) -> Self {
+    hashing::Fingerprint::from_hex_string(&d.hash)
+      .map_err(|err| format!("Bad fingerprint in Digest {:?}: {:?}", d.hash, err))
+      .map(|fingerprint| hashing::Digest(fingerprint, d.size_bytes as usize))
+  }
+}
+
 impl From<crate::google::longrunning::Operation> for crate::operations::Operation {
   fn from(op: crate::google::longrunning::Operation) -> Self {
     let mut dst = Self::new();
     dst.set_name(op.name);
-    dst.set_metadata(prost_any_to_gcprio_any(op.metadata.unwrap()));
+    if let Some(metadata) = op.metadata {
+      dst.set_metadata(prost_any_to_gcprio_any(metadata));
+    }
     dst.set_done(op.done);
     match op.result {
       Some(crate::google::longrunning::operation::Result::Response(response)) => {
@@ -45,6 +57,46 @@ impl From<crate::google::longrunning::Operation> for crate::operations::Operatio
   }
 }
 
+// This should only be used in test contexts. It should be deleted when the mock systems use tower.
+impl From<crate::remote_execution::ExecuteRequest>
+  for crate::build::bazel::remote::execution::v2::ExecuteRequest
+{
+  fn from(req: crate::remote_execution::ExecuteRequest) -> Self {
+    if req.has_execution_policy() || req.has_results_cache_policy() {
+      panic!("Can't convert ExecuteRequest protos with execution policy or results cache policy");
+    }
+    let digest: Result<hashing::Digest, String> = req.get_action_digest().into();
+    Self {
+      action_digest: Some((&digest.expect("Bad digest converting ExecuteRequest proto")).into()),
+      instance_name: req.instance_name,
+      execution_policy: None,
+      results_cache_policy: None,
+      skip_cache_lookup: req.skip_cache_lookup,
+    }
+  }
+}
+
+// This should only be used in test contexts. It should be deleted when the mock systems use tower.
+impl From<crate::build::bazel::remote::execution::v2::ExecuteRequest>
+  for crate::remote_execution::ExecuteRequest
+{
+  fn from(req: crate::build::bazel::remote::execution::v2::ExecuteRequest) -> Self {
+    if req.execution_policy.is_some() || req.results_cache_policy.is_some() {
+      panic!("Can't convert ExecuteRequest protos with execution policy or results cache policy");
+    }
+    let digest: Result<hashing::Digest, String> = (&req
+      .action_digest
+      .expect("Missing digest converting ExecuteRequest proto"))
+      .into();
+
+    let mut ret = Self::new();
+    ret.set_action_digest((&digest.expect("Bad digest converting ExecuteRequest proto")).into());
+    ret.set_instance_name(req.instance_name);
+    ret.set_skip_cache_lookup(req.skip_cache_lookup);
+    ret
+  }
+}
+
 pub fn prost_any_to_gcprio_any(any: prost_types::Any) -> protobuf::well_known_types::Any {
   let prost_types::Any { type_url, value } = any;
   let mut dst = protobuf::well_known_types::Any::new();