Repository: kythe/kythe
Pull request #4985 (2021-07-15 00:43:47)

feat(rust_indexer): Add support for communicating with the analyzer driver
This PR adds support to the Rust indexer to communicate with the analyzer driver. This includes support within the indexer library and a new binary to continuously process compilation units.

shahms: You probably want to setup the git commit hooks (from https://www.kythe.io/contributing/) so you can get the lint warnings earlier.
wcalandro:       
  

 Since this is a fatal error, shouldn't the indexer simply die, which will be detected by the proxy?
  
  

  
  
>   
> On Jul 15, 2021 at 21:42,  <Jay Sachs ***@***.***)>  wrote:
>   
>   
>   
>
>
>
>   
>
> @jaysachs  commented on this pull request.
>
>     
>
> In  kythe/rust/indexer/src/bin/proxy/main.rs (https://github.com/kythe/kythe/pull/4985#discussion_r670960974):
>
>   
>   >  + path::{Path, PathBuf}, +}; +use tempdir::TempDir; + +fn main() ->  Result<()>  { + let mut file_provider = ProxyFileProvider::new(); + let mut kythe_writer = ProxyWriter::new(); + let mut indexer = KytheIndexer::new(&mut kythe_writer); + + // Request and process + loop { + let unit = request_compilation_unit()?; + // Retrieve the save_analysis file + let temp_dir = TempDir::new("rust_indexer").context("Couldn't create temporary directory")?; + let temp_path = PathBuf::new().join(temp_dir.path()); + write_analysis_to_directory(&unit,  &temp_path,  &mut file_provider)?;  
>   
>
> So if this fails  &  returns (or the  TempDir::new) we don't write "done" back to the proxy, and so the conversation will be out of sync.
>
>   
>
> —
>  You are receiving this because you authored the thread.
>  Reply to this email directly,   view it on GitHub (https://github.com/kythe/kythe/pull/4985#pullrequestreview-708001593), or   unsubscribe (https://github.com/notifications/unsubscribe-auth/APPKRXLTKTTIOWY23HSZTEDTX62CVANCNFSM5AMPV4DA).
>
>   
>   
  
  
       

Patch:
diff --git a/.commitlintrc.js b/.commitlintrc.js
index 5d503be2a..5ba6f8994 100644
--- a/.commitlintrc.js
+++ b/.commitlintrc.js
@@ -46,6 +46,9 @@ module.exports = {
         'jvm_common',
         'jvm_extractor',
         'jvm_indexer',
+        'rust_common',
+        'rust_extractor',
+        'rust_indexer',
         'typescript_common',
         'typescript_extractor',
         'typescript_indexer',
diff --git a/kythe/rust/cargo/cargo/BUILD.bazel b/kythe/rust/cargo/cargo/BUILD.bazel
deleted file mode 100644
index d131613c4..000000000
--- a/kythe/rust/cargo/cargo/BUILD.bazel
+++ /dev/null
@@ -1,174 +0,0 @@
-"""
-@generated
-cargo-raze generated Bazel file.
-
-DO NOT EDIT! Replaced on runs of cargo-raze
-"""
-
-package(default_visibility = ["//visibility:public"])
-
-licenses([
-    "notice",  # See individual crates for specific licenses
-])
-
-# Aliased targets
-alias(
-    name = "anyhow",
-    actual = "@raze__anyhow__1_0_40//:anyhow",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "assert_cmd",
-    actual = "@raze__assert_cmd__1_0_4//:assert_cmd",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "clap",
-    actual = "@raze__clap__2_33_3//:clap",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "colored",
-    actual = "@raze__colored__2_0_0//:colored",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "lazy_static",
-    actual = "@raze__lazy_static__1_4_0//:lazy_static",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "predicates",
-    actual = "@raze__predicates__1_0_8//:predicates",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "quick_error",
-    actual = "@raze__quick_error__1_2_3//:quick_error",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "rayon",
-    actual = "@raze__rayon__1_5_1//:rayon",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "regex",
-    actual = "@raze__regex__1_5_4//:regex",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "rls_analysis",
-    actual = "@raze__rls_analysis__0_18_1//:rls_analysis",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "rls_data",
-    actual = "@raze__rls_data__0_19_1//:rls_data",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "rust_crypto",
-    actual = "@raze__rust_crypto__0_2_36//:rust_crypto",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "serde",
-    actual = "@raze__serde__1_0_126//:serde",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "serde_json",
-    actual = "@raze__serde_json__1_0_64//:serde_json",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "serial_test",
-    actual = "@raze__serial_test__0_4_0//:serial_test",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "tempdir",
-    actual = "@raze__tempdir__0_3_7//:tempdir",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-alias(
-    name = "zip",
-    actual = "@raze__zip__0_5_12//:zip",
-    tags = [
-        "cargo-raze",
-        "manual",
-    ],
-)
-
-# Export file for Stardoc support
-exports_files(
-    [
-        "crates.bzl",
-    ],
-    visibility = ["//visibility:public"],
-)
diff --git a/kythe/rust/indexer/BUILD b/kythe/rust/indexer/BUILD
index d4db00d9a..f2e71a760 100644
--- a/kythe/rust/indexer/BUILD
+++ b/kythe/rust/indexer/BUILD
@@ -18,15 +18,17 @@ rust_library(
         "//kythe/rust/cargo:zip",
         "//kythe/rust/cargo:rls_analysis",
         "//kythe/rust/cargo:rls_data",
+        "//kythe/rust/cargo:serde_json",
+        "//kythe/rust/cargo:hex",
     ] + PROTO_COMPILE_DEPS,
 )
 
 rust_binary(
-    name = "indexer",
+    name = "bazel_indexer",
     srcs = glob(
-        include = ["src/bin/**/*.rs"],
+        include = ["src/bin/bazel/*.rs"],
     ),
-    crate_root = ":src/bin/main.rs",
+    crate_root = ":src/bin/bazel/main.rs",
     edition = "2018",
     deps = [
         ":kythe_rust_indexer",
@@ -38,6 +40,25 @@ rust_binary(
     ],
 )
 
+rust_binary(
+    name = "proxy_indexer",
+    srcs = glob(
+        include = ["src/bin/proxy/*.rs"],
+    ),
+    crate_root = ":src/bin/proxy/main.rs",
+    edition = "2018",
+    deps = [
+        ":kythe_rust_indexer",
+        "//kythe/proto:analysis_rust_proto",
+        "//kythe/proto:storage_rust_proto",
+        "//kythe/rust/cargo:anyhow",
+        "//kythe/rust/cargo:hex",
+        "//kythe/rust/cargo:serde_json",
+        "//kythe/rust/cargo:tempdir",
+        "@rules_rust//proto/raze:protobuf",
+    ],
+)
+
 rust_test(
     name = "inline_tests",
     crate = ":kythe_rust_indexer",
@@ -62,8 +83,9 @@ rust_test(
 rust_clippy(
     name = "clippy",
     deps = [
-        ":indexer",
+        ":bazel_indexer",
         ":kythe_rust_indexer",
+        ":proxy_indexer",
     ],
 )
 
diff --git a/kythe/rust/indexer/src/bin/main.rs b/kythe/rust/indexer/src/bin/bazel/main.rs
similarity index 95%
rename from kythe/rust/indexer/src/bin/main.rs
rename to kythe/rust/indexer/src/bin/bazel/main.rs
index ba672dde1..da2e3c1fb 100644
--- a/kythe/rust/indexer/src/bin/main.rs
+++ b/kythe/rust/indexer/src/bin/bazel/main.rs
@@ -56,7 +56,7 @@ fn main() -> Result<()> {
             TempDir::new("rust_indexer").context("Couldn't create temporary directory")?;
         let temp_path = PathBuf::new().join(temp_dir.path());
 
-        // Extract all of the files from the kzip into the temporary directory
+        // Extract the analysis files from the kzip into the temporary directory
         extract_analysis_from_kzip(&unit, &temp_path, &mut kzip_provider)?;
 
         // Index the CompilationUnit
@@ -80,7 +80,7 @@ pub fn extract_analysis_from_kzip(
         if let Some(os_str) = input_path_buf.extension() {
             if let Some("json") = os_str.to_str() {
                 let digest = required_input.get_info().get_digest();
-                let file_contents = provider.contents(digest).with_context(|| {
+                let file_contents = provider.contents(input_path, digest).with_context(|| {
                     format!(
                         "Failed to get contents of file \"{}\" with digest \"{}\"",
                         input_path, digest
diff --git a/kythe/rust/indexer/src/bin/proxy/main.rs b/kythe/rust/indexer/src/bin/proxy/main.rs
new file mode 100644
index 000000000..091992e9b
--- /dev/null
+++ b/kythe/rust/indexer/src/bin/proxy/main.rs
@@ -0,0 +1,143 @@
+// Copyright 2021 The Kythe Authors. All rights reserved.
+//
+// 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.
+extern crate kythe_rust_indexer;
+use kythe_rust_indexer::{
+    error::KytheError,
+    indexer::KytheIndexer,
+    providers::*,
+    writer::ProxyWriter
+};
+
+use analysis_rust_proto::*;
+use anyhow::{anyhow, Context, Result};
+use serde_json::Value;
+use std::{
+    fs::File,
+    io::{self, Write},
+    path::{Path, PathBuf},
+};
+use tempdir::TempDir;
+
+fn main() -> Result<()> {
+    let mut file_provider = ProxyFileProvider::new();
+    let mut kythe_writer = ProxyWriter::new();
+    let mut indexer = KytheIndexer::new(&mut kythe_writer);
+
+    // Request and process
+    loop {
+        let unit = request_compilation_unit()?;
+        // Retrieve the save_analysis file
+        let temp_dir = TempDir::new("rust_indexer").context("Couldn't create temporary directory")?;
+        let temp_path = PathBuf::new().join(temp_dir.path());
+        let write_res = write_analysis_to_directory(&unit, &temp_path, &mut file_provider);
+        if write_res.is_err() {
+            send_done(false, write_res.err().unwrap().to_string())?;
+            continue;
+        }
+
+        // Index the CompilationUnit and let the proxy know we are done
+        let index_res = indexer.index_cu(&unit, &temp_path, &mut file_provider);
+        if index_res.is_ok() {
+            send_done(true, String::new())?;
+        } else {
+            send_done(false, index_res.err().unwrap().to_string())?;
+        }
+    }
+}
+
+/// Takes analysis files present in a CompilationUnit, requests the files
+/// from the proxy, and writes them to `temp_path`
+pub fn write_analysis_to_directory(
+    c_unit: &CompilationUnit,
+    temp_path: &Path,
+    provider: &mut dyn FileProvider,
+) -> std::result::Result<(), KytheError> {
+    for required_input in c_unit.get_required_input() {
+        let input_path = required_input.get_info().get_path();
+        let input_path_buf = PathBuf::new().join(input_path);
+
+        // save_analysis files are JSON files
+        if let Some(os_str) = input_path_buf.extension() {
+            if let Some("json") = os_str.to_str() {
+                let digest = required_input.get_info().get_digest();
+                let file_contents = provider.contents(&input_path, digest).map_err(|err| {
+                    KytheError::IndexerError(
+                        format!(
+                            "Failed to get contents of file \"{}\" with digest \"{}\": {:?}",
+                            input_path, digest, err
+                        )
+                    )
+                })?;
+
+                let output_path = temp_path.join(input_path_buf.file_name().unwrap());
+                let mut output_file = File::create(&output_path)
+                    .map_err(|err| KytheError::IndexerError(format!("Failed to create file: {:?}", err)))?;
+                output_file.write_all(&file_contents).map_err(|err| {
+                    KytheError::IndexerError(
+                        format!(
+                            "Failed to copy contents of \"{}\" with digest \"{}\" to \"{}\": {:?}",
+                            input_path,
+                            digest,
+                            output_path.display(),
+                            err
+                        )
+                    )
+                })?;
+            }
+        }
+    }
+    Ok(())
+}
+
+fn request_compilation_unit() -> Result<CompilationUnit> {
+    println!(r#"{{"req":"analysis_wire"}}"#);
+    io::stdout().flush().context("Failed to flush stdout")?;
+
+    // Read the response from the proxy
+    let mut response_string = String::new();
+    io::stdin()
+        .read_line(&mut response_string)
+        .context("Failed to read response from proxy")?;
+
+    // Convert to json and extract information
+    let response: Value = serde_json::from_str(&response_string)
+        .context("Failed to parse response as JSON")?;
+    if response["rsp"].as_str().unwrap() == "ok" {
+        let args = response["args"].clone();
+        let unit_base64 = args["unit"].as_str().unwrap();
+        let unit_bytes = hex::decode(unit_base64).context("Failed to decode CompilationUnit sent by proxy")?;
+        let unit = protobuf::parse_from_bytes::<CompilationUnit>(&unit_bytes)
+            .context("Failed to parse protobuf")?;
+        Ok(unit)
+    } else {
+        Err(anyhow!("Failed to get a CompilationUnit from the proxy"))
+    }
+}
+
+/// Sends the "done" request to the proxy. The first argument sets the "ok" value, and the second sets the error message if the first argument is false.
+fn send_done(ok: bool, msg: String) -> Result<()> {
+    if ok {
+        println!(r#"{{"req":"done", "args"{{"ok":"true","msg":""}}}}"#);
+    } else {
+        println!(r#"{{"req":"done", "args"{{"ok":"false","msg":"{}"}}}}"#, msg);
+    }
+    io::stdout().flush().context("Failed to flush stdout")?;
+
+    // Grab the response, but we don't care what it is so just throw it away
+    let mut response_string = String::new();
+    io::stdin()
+        .read_line(&mut response_string)
+        .context("Failed to read response from proxy")?;
+    Ok(())
+}
diff --git a/kythe/rust/indexer/src/indexer/analyzers.rs b/kythe/rust/indexer/src/indexer/analyzers.rs
index 01922c92e..c872ed59e 100644
--- a/kythe/rust/indexer/src/indexer/analyzers.rs
+++ b/kythe/rust/indexer/src/indexer/analyzers.rs
@@ -133,7 +133,7 @@ impl<'a> UnitAnalyzer<'a> {
             // Returns a FileReadError if we can't read the file
             let file_contents: String;
             if let Some(file_digest) = self.file_digests.get(&source_file.to_string()) {
-                let file_bytes = self.provider.contents(file_digest)?;
+                let file_bytes = self.provider.contents(&source_file, file_digest)?;
                 file_contents = String::from_utf8(file_bytes)
                     .map_err(|_| KytheError::IndexerError(
                         format!("Failed to read file {} as UTF8 string", source_file.to_string())
diff --git a/kythe/rust/indexer/src/providers.rs b/kythe/rust/indexer/src/providers.rs
index feedf24c8..c03b7371c 100644
--- a/kythe/rust/indexer/src/providers.rs
+++ b/kythe/rust/indexer/src/providers.rs
@@ -14,8 +14,9 @@
 
 use crate::error::KytheError;
 use analysis_rust_proto::*;
+use serde_json::Value;
 use std::fs::File;
-use std::io::{BufReader, Read};
+use std::io::{self, BufReader, Read, Write};
 use std::path::Path;
 use zip::ZipArchive;
 
@@ -23,18 +24,18 @@ use zip::ZipArchive;
 pub trait FileProvider {
     /// Checks whether a file exists and that the contents are available.
     ///
-    /// This function takes a file hash and returns whether
+    /// This function takes a file path and digest, and returns whether
     /// the file exists and that the contents can be requested.
     ///
     /// # Errors
     ///
     /// If the file is not a valid zip archive, a [KytheError::KzipFileError]
     /// will be returned.
-    fn exists(&mut self, file_hash: &str) -> bool;
+    fn exists(&mut self, path: &str, digest: &str) -> Result<bool, KytheError>;
 
     /// Retrieves the byte contents of a file.
     ///
-    /// This function takes a file hash and returns the bytes contents of the
+    /// This function takes a file path and digest, and returns the bytes contents of the
     /// file in a vector.
     ///
     /// # Errors
@@ -43,10 +44,10 @@ pub trait FileProvider {
     /// [FileNotFoundError][KytheError::FileNotFoundError] will be returned.
     /// If the file can't be read, a [FileReadError][KytheError::FileReadError]
     /// will be returned.
-    fn contents(&mut self, file_hash: &str) -> Result<Vec<u8>, KytheError>;
+    fn contents(&mut self, path: &str, digest: &str) -> Result<Vec<u8>, KytheError>;
 }
 
-/// A [FileProvider] that backed by a .kzip file.
+/// A [FileProvider] that is backed by a .kzip file.
 pub struct KzipFileProvider {
     zip_archive: ZipArchive<BufReader<File>>,
     /// The top level folder name inside of the zip archive
@@ -114,21 +115,21 @@ impl KzipFileProvider {
 }
 
 impl FileProvider for KzipFileProvider {
-    /// Given a file hash, returns whether the file exists in the kzip.
-    fn exists(&mut self, file_hash: &str) -> bool {
-        let name = format!("{}/files/{}", self.root_name, file_hash);
-        self.zip_archive.by_name(&name).is_ok()
+    /// Given a file path and digest, returns whether the file exists in the kzip.
+    fn exists(&mut self, _path: &str, digest: &str) -> Result<bool, KytheError> {
+        let name = format!("{}/files/{}", self.root_name, digest);
+        Ok(self.zip_archive.by_name(&name).is_ok())
     }
 
-    /// Given a file hash, returns the vector of bytes of the file from the
+    /// Given a file path and digest, returns the vector of bytes of the file from the
     /// kzip.
     ///
     /// # Errors
     ///
     /// An error will be returned if the file does not exist or cannot be read.
-    fn contents(&mut self, file_hash: &str) -> Result<Vec<u8>, KytheError> {
+    fn contents(&mut self, _path: &str, digest: &str) -> Result<Vec<u8>, KytheError> {
         // Ensure the file exists in the kzip
-        let name = format!("{}/files/{}", self.root_name, file_hash);
+        let name = format!("{}/files/{}", self.root_name, digest);
         let file = self
             .zip_archive
             .by_name(&name)
@@ -139,3 +140,72 @@ impl FileProvider for KzipFileProvider {
         Ok(file_contents)
     }
 }
+
+/// A [FileProvider] that backed by the analysis driver proxy
+pub struct ProxyFileProvider {}
+
+impl ProxyFileProvider {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl Default for ProxyFileProvider {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl FileProvider for ProxyFileProvider {
+    /// Given a file path and digest, returns whether the proxy can find the file.
+    fn exists(&mut self, path: &str, digest: &str) -> Result<bool, KytheError> {
+        // Submit the file request to the proxy
+        println!(r#"{{"req": "file","args":{{"path": "{}","digest": "{}"}}}}"#, path, digest);
+        io::stdout().flush()
+            .map_err(|err| KytheError::IndexerError(format!("Failed to flush stdout: {:?}", err)))?;
+
+        // Read the response from the proxy
+        let mut response_string = String::new();
+        io::stdin()
+            .read_line(&mut response_string)
+            .map_err(|err| KytheError::IndexerError(format!("Failed to read proxy file response from stdin: {:?}", err)))?;
+
+        // Convert to json and extract information
+        let response: Value = serde_json::from_str(&response_string)
+            .map_err(|err| KytheError::IndexerError(format!("Failed to read proxy file response from stdin: {:?}", err)))?;
+        if response["rsp"].as_str().unwrap() == "error" {
+            // The file couldn't be found
+            Ok(false)
+        } else {
+            Ok(true)
+        }
+    }
+
+    fn contents(&mut self, path: &str, digest: &str) -> Result<Vec<u8>, KytheError> {
+        // Submit the file request to the proxy
+        println!(r#"{{"req": "file","args":{{"path": "{}","digest": "{}"}}}}"#, path, digest);
+        io::stdout().flush()
+            .map_err(|err| KytheError::IndexerError(format!("Failed to flush stdout: {:?}", err)))?;
+
+        // Read the response from the proxy
+        let mut response_string = String::new();
+        io::stdin()
+            .read_line(&mut response_string)
+            .map_err(|err| KytheError::IndexerError(format!("Failed to read proxy file response from stdin: {:?}", err)))?;
+
+        // Convert to json and extract information
+        let response: Value = serde_json::from_str(&response_string)
+            .map_err(|err| KytheError::IndexerError(format!("Failed to read proxy file response from stdin: {:?}", err)))?;
+        if response["rsp"].as_str().unwrap() == "error" {
+            // The file couldn't be found
+            Err(KytheError::FileNotFoundError(path.to_string()))
+        } else {
+            let args: Value = response["args"].clone();
+            // Retrieve base64 content and convert to a Vec<u8>
+            let content_base64: &str = args["content"].as_str().unwrap();
+            let content: Vec<u8> = hex::decode(&content_base64)
+                .map_err(|err| KytheError::IndexerError(format!("Failed to convert base64 file contents response to bytes: {:?}", err)))?;
+            Ok(content)
+        }
+    }
+}
diff --git a/kythe/rust/indexer/src/writer.rs b/kythe/rust/indexer/src/writer.rs
index e4a60c0a2..44e887952 100644
--- a/kythe/rust/indexer/src/writer.rs
+++ b/kythe/rust/indexer/src/writer.rs
@@ -14,7 +14,8 @@
 
 use crate::error::KytheError;
 use protobuf::{self, CodedOutputStream, Message};
-use std::io::Write;
+use serde_json::Value;
+use std::io::{self, Write};
 use storage_rust_proto::*;
 
 /// A trait for writing Kythe graph entries to an output.
@@ -79,6 +80,76 @@ impl<'a> KytheWriter for CodedOutputStreamWriter<'a> {
     }
 }
 
+/// A [KytheWriter] that communicates with the analysis driver proxy
+pub struct ProxyWriter {
+    buffer: Vec<String>,
+}
+
+impl ProxyWriter {
+    /// Create a new instance of ProxyWriter
+    ///
+    /// Takes a writer that implements the `Write` trait as the only argument.
+    pub fn new() -> ProxyWriter {
+        Self {
+            buffer: Vec::new(),
+        }
+    }
+}
+
+impl Default for ProxyWriter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl KytheWriter for ProxyWriter {
+    /// Given an [Entry], writes the entry to the buffer.
+    /// If the buffer reaches size 1000, automatically flush to the proxy.
+    ///
+    /// # Errors
+    ///
+    /// An error is returned if the write fails.
+    fn write_entry(&mut self, entry: Entry) -> Result<(), KytheError> {
+        let bytes = entry.write_to_bytes().map_err(KytheError::WriterError)?;
+        // Convert the entry bytes to base64 and surround with quotes to prepare for inserting into a JSON
+        // array
+        let b64 = format!(r#""{}""#, hex::encode(&bytes));
+        self.buffer.push(b64);
+        if self.buffer.len() >= 1000 {
+            self.flush()?;
+        }
+        Ok(())
+    }
+
+    fn flush(&mut self) -> Result<(), KytheError> {
+        // Combine all of the entries
+        let buffer_json_array = self.buffer.join(", ");
+        self.buffer.clear();
+
+        // Write the proxy command to output
+        println!(r#"{{"req":"output_wire","args":[{}]}}"#, buffer_json_array);
+        io::stdout().flush()
+            .map_err(|err| KytheError::IndexerError(format!("Failed to flush stdout: {:?}", err)))?;
+
+        // Read the response from the proxy
+        let mut response_string = String::new();
+        io::stdin()
+            .read_line(&mut response_string)
+            .map_err(|err| KytheError::IndexerError(format!("Failed to read proxy file response from stdin: {:?}", err)))?;
+
+        // Convert to json and extract information
+        let response: Value = serde_json::from_str(&response_string)
+            .map_err(|err| KytheError::IndexerError(format!("Failed to read proxy file response from stdin: {:?}", err)))?;
+        if response["rsp"].as_str().unwrap() == "error" {
+            Err(KytheError::IndexerError(
+                format!("Got error back from proxy after sending entries: {}", response["args"].as_str().unwrap())
+            ))
+        } else {
+            Ok(())
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;


Test patch:
diff --git a/kythe/rust/indexer/tests/providers.rs b/kythe/rust/indexer/tests/providers.rs
index 9ce9c9c91..750601b03 100644
--- a/kythe/rust/indexer/tests/providers.rs
+++ b/kythe/rust/indexer/tests/providers.rs
@@ -41,16 +41,16 @@ fn test_kzip_provider() {
     // Check the `exists` function
     let file_hash = "c9d04c9565fc665c80681fb1d829938026871f66e14f501e08531df66938a789";
     assert!(
-        kzip_provider.exists(file_hash),
+        kzip_provider.exists("/tmp/main.rs", file_hash).unwrap(),
         "File should exist in kzip but doesn't"
     );
     assert!(
-        !kzip_provider.exists("invalid"),
+        !kzip_provider.exists("invalid", "invalid").unwrap(),
         "File shouldn't exist but does"
     );
 
     // Check the `contents` function
-    let contents_result = kzip_provider.contents(file_hash);
+    let contents_result = kzip_provider.contents("/tmp/main.rs", file_hash);
     assert!(!contents_result.is_err());
     let contents_string =
         String::from_utf8(contents_result.unwrap()).expect("File contents was not valid UTF-8");
@@ -59,7 +59,7 @@ fn test_kzip_provider() {
         "File contents did not match expected contents"
     );
 
-    let invalid_contents = kzip_provider.contents("invalid");
+    let invalid_contents = kzip_provider.contents("invalid", "invalid");
     assert!(invalid_contents.is_err(), "Expected Err while reading contents for non-existent file, but received file contents: {:?}", invalid_contents.unwrap());
     let contents_error = invalid_contents.err().unwrap();
     match contents_error {
diff --git a/tools/build_rules/verifier_test/rust_indexer_test.bzl b/tools/build_rules/verifier_test/rust_indexer_test.bzl
index 18b4e31c4..1d06f7a52 100644
--- a/tools/build_rules/verifier_test/rust_indexer_test.bzl
+++ b/tools/build_rules/verifier_test/rust_indexer_test.bzl
@@ -161,7 +161,7 @@ rust_entries = rule(
 
         # The location of the Rust indexer binary.
         "_indexer": attr.label(
-            default = Label("//kythe/rust/indexer"),
+            default = Label("//kythe/rust/indexer:bazel_indexer"),
             executable = True,
             cfg = "host",
         ),