Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: duplicate read proto #1741

Merged
merged 24 commits into from
Apr 18, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions src/config/reader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,24 +10,25 @@ use super::{ConfigModule, Content, Link, LinkType};
use crate::config::{Config, ConfigReaderContext, Source};
use crate::merge_right::MergeRight;
use crate::proto_reader::ProtoReader;
use crate::resource_reader::ResourceReader;
use crate::resource_reader::{Cached, ResourceReader};
use crate::rest::EndpointSet;
use crate::runtime::TargetRuntime;

/// Reads the configuration from a file or from an HTTP URL and resolves all
/// linked extensions to create a ConfigModule.
pub struct ConfigReader {
runtime: TargetRuntime,
resource_reader: ResourceReader,
resource_reader: ResourceReader<Cached>,
proto_reader: ProtoReader,
}

impl ConfigReader {
pub fn init(runtime: TargetRuntime) -> Self {
let resource_reader = ResourceReader::<Cached>::cached(runtime.clone());
Self {
runtime: runtime.clone(),
resource_reader: ResourceReader::init(runtime.clone()),
proto_reader: ProtoReader::init(runtime),
resource_reader: resource_reader.clone(),
proto_reader: ProtoReader::init(resource_reader),
}
}

Expand Down Expand Up @@ -157,12 +158,15 @@ impl ConfigReader {
}

/// Reads a single file and returns the config
pub async fn read<T: ToString>(&self, file: T) -> anyhow::Result<ConfigModule> {
pub async fn read<T: ToString + Send + Sync>(&self, file: T) -> anyhow::Result<ConfigModule> {
self.read_all(&[file]).await
}

/// Reads all the files and returns a merged config
pub async fn read_all<T: ToString>(&self, files: &[T]) -> anyhow::Result<ConfigModule> {
pub async fn read_all<T: ToString + Send + Sync>(
&self,
files: &[T],
) -> anyhow::Result<ConfigModule> {
let files = self.resource_reader.read_files(files).await?;
let mut config_module = ConfigModule::default();

Expand Down Expand Up @@ -249,7 +253,7 @@ mod reader_tests {
cfg = cfg.types([("Test", Type::default())].to_vec());

let server = start_mock_server();
let header_serv = server.mock(|when, then| {
let header_server = server.mock(|when, then| {
when.method(httpmock::Method::GET).path("/bar.graphql");
then.status(200).body(cfg.to_sdl());
});
Expand Down Expand Up @@ -287,7 +291,7 @@ mod reader_tests {
.collect::<Vec<String>>()
);
foo_json_server.assert(); // checks if the request was actually made
header_serv.assert();
header_server.assert();
}

#[tokio::test]
Expand Down
5 changes: 4 additions & 1 deletion src/generator/generator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,17 @@ use crate::generator::from_proto::from_proto;
use crate::generator::source::Source;
use crate::merge_right::MergeRight;
use crate::proto_reader::ProtoReader;
use crate::resource_reader::ResourceReader;
use crate::runtime::TargetRuntime;

pub struct Generator {
proto_reader: ProtoReader,
}
impl Generator {
pub fn init(runtime: TargetRuntime) -> Self {
Self { proto_reader: ProtoReader::init(runtime) }
Self {
proto_reader: ProtoReader::init(ResourceReader::cached(runtime)),
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we changing the API here? I don't think it's required.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ProtoReader need to share ResourceReader to avoid duplicate reading.

}

pub async fn read_all<T: AsRef<str>>(
Expand Down
18 changes: 10 additions & 8 deletions src/proto_reader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,10 @@ use futures_util::future::join_all;
use prost_reflect::prost_types::{FileDescriptorProto, FileDescriptorSet};
use protox::file::{FileResolver, GoogleFileResolver};

use crate::resource_reader::ResourceReader;
use crate::runtime::TargetRuntime;
use crate::resource_reader::{Cached, ResourceReader};

pub struct ProtoReader {
resource_reader: ResourceReader,
resource_reader: ResourceReader<Cached>,
}

pub struct ProtoMetadata {
Expand All @@ -18,8 +17,8 @@ pub struct ProtoMetadata {
}

impl ProtoReader {
pub fn init(runtime: TargetRuntime) -> Self {
Self { resource_reader: ResourceReader::init(runtime) }
pub fn init(resource_reader: ResourceReader<Cached>) -> Self {
Self { resource_reader }
}

pub async fn read_all<T: AsRef<str>>(&self, paths: &[T]) -> anyhow::Result<Vec<ProtoMetadata>> {
Expand Down Expand Up @@ -100,11 +99,14 @@ mod test_proto_config {
use pretty_assertions::assert_eq;

use crate::proto_reader::ProtoReader;
use crate::resource_reader::{Cached, ResourceReader};

#[tokio::test]
async fn test_resolve() {
// Skipping IO tests as they are covered in reader.rs
let reader = ProtoReader::init(crate::runtime::test::init(None));
let reader = ProtoReader::init(ResourceReader::<Cached>::cached(
crate::runtime::test::init(None),
));
reader
.read_proto("google/protobuf/empty.proto")
.await
Expand Down Expand Up @@ -133,7 +135,7 @@ mod test_proto_config {
let runtime = crate::runtime::test::init(None);
let file_rt = runtime.file.clone();

let reader = ProtoReader::init(runtime);
let reader = ProtoReader::init(ResourceReader::<Cached>::cached(runtime));
let helper_map = reader
.resolve_descriptors(reader.read_proto(&test_file).await?)
.await?;
Expand Down Expand Up @@ -177,7 +179,7 @@ mod test_proto_config {
#[tokio::test]
async fn test_proto_no_pkg() -> Result<()> {
let runtime = crate::runtime::test::init(None);
let reader = ProtoReader::init(runtime);
let reader = ProtoReader::init(ResourceReader::<Cached>::cached(runtime));
let mut proto_no_pkg = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
proto_no_pkg.push("src/grpc/tests/proto_no_pkg.graphql");
let config_module = reader.read(proto_no_pkg.to_str().unwrap()).await;
Expand Down
104 changes: 89 additions & 15 deletions src/resource_reader.rs
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use futures_util::future::join_all;
use futures_util::TryFutureExt;
use url::Url;
Expand All @@ -11,16 +14,58 @@ pub struct FileRead {
pub path: String,
}

pub struct ResourceReader {
#[derive(Clone)]
pub struct ResourceReader<A>(A);

impl<A: Reader + Send + Sync> ResourceReader<A> {
/// Reads all the files in parallel
pub async fn read_files<T: ToString + Send + Sync>(
&self,
files: &[T],
) -> anyhow::Result<Vec<FileRead>> {
let files = files.iter().map(|x| {
self.read_file(x.to_string())
.map_err(|e| e.context(x.to_string()))
tusharmath marked this conversation as resolved.
Show resolved Hide resolved
});
let content = join_all(files)
.await
.into_iter()
.collect::<anyhow::Result<Vec<_>>>()?;
Ok(content)
tusharmath marked this conversation as resolved.
Show resolved Hide resolved
}

pub async fn read_file<T: ToString + Send>(&self, file: T) -> anyhow::Result<FileRead> {
self.0.read(file).await
}
}

impl ResourceReader<Cached> {
pub fn cached(runtime: TargetRuntime) -> Self {
ResourceReader(Cached::init(runtime))
}
}

#[async_trait::async_trait]
pub trait Reader {
async fn read<T: ToString + Send>(&self, file: T) -> anyhow::Result<FileRead>;
}

/// Reads the files directly from the filesystem or from an HTTP URL
#[derive(Clone)]
pub struct Direct {
runtime: TargetRuntime,
}

impl ResourceReader {
impl Direct {
pub fn init(runtime: TargetRuntime) -> Self {
Self { runtime }
}
}

#[async_trait::async_trait]
impl Reader for Direct {
/// Reads a file from the filesystem or from an HTTP URL
pub async fn read_file<T: ToString>(&self, file: T) -> anyhow::Result<FileRead> {
async fn read<T: ToString + Send>(&self, file: T) -> anyhow::Result<FileRead> {
// Is an HTTP URL
let content = if let Ok(url) = Url::parse(&file.to_string()) {
if url.scheme().starts_with("http") {
Expand All @@ -41,20 +86,49 @@ impl ResourceReader {

self.runtime.file.read(&file.to_string()).await?
};

Ok(FileRead { content, path: file.to_string() })
}
}

/// Reads all the files in parallel
pub async fn read_files<T: ToString>(&self, files: &[T]) -> anyhow::Result<Vec<FileRead>> {
let files = files.iter().map(|x| {
self.read_file(x.to_string())
.map_err(|e| e.context(x.to_string()))
});
let content = join_all(files)
.await
.into_iter()
.collect::<anyhow::Result<Vec<_>>>()?;
Ok(content)
/// Reads the files from the filesystem or from an HTTP URL with cache
#[derive(Clone)]
pub struct Cached {
direct: Direct,
// Cache file content, path -> content
cache: Arc<Mutex<HashMap<String, String>>>,
}

impl Cached {
pub fn init(runtime: TargetRuntime) -> Self {
Self { direct: Direct::init(runtime), cache: Default::default() }
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Initialize the cache explicitly to ensure it matches the expected type and is thread-safe.

- cache: Default::default(),
+ cache: Arc::new(Mutex::new(HashMap::new())),

Explicitly initializing the cache as Arc<Mutex<HashMap<String, String>>> ensures that the type is correct and that thread safety is maintained.


Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.

Suggested change
Self { direct: Direct::init(runtime), cache: Default::default() }
Self { direct: Direct::init(runtime), cache: Arc::new(Mutex::new(HashMap::new())) }

}
}

#[async_trait::async_trait]
impl Reader for Cached {
/// Reads a file from the filesystem or from an HTTP URL with cache
async fn read<T: ToString + Send>(&self, file: T) -> anyhow::Result<FileRead> {
// check cache
let file_path = file.to_string();
let content = self
.cache
.as_ref()
.lock()
.unwrap()
tusharmath marked this conversation as resolved.
Show resolved Hide resolved
.get(&file_path)
.map(|v| v.to_owned());
let content = if let Some(content) = content {
content.to_owned()
} else {
let file_read = self.direct.read(file.to_string()).await?;
self.cache
.as_ref()
.lock()
.unwrap()
.insert(file_path.to_owned(), file_read.content.clone());
file_read.content
};

Ok(FileRead { content, path: file_path })
}
}
Loading