This is an automated email from the ASF dual-hosted git repository.
chaokunyang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fory.git
The following commit(s) were added to refs/heads/main by this push:
new 3ab5cbce4 feat(compiler): support Rust gRPC code generation (#3738)
3ab5cbce4 is described below
commit 3ab5cbce4af85207c535815d146599476236a3fc
Author: Peiyang He <[email protected]>
AuthorDate: Fri Jun 12 14:15:35 2026 +0800
feat(compiler): support Rust gRPC code generation (#3738)
## Why?
Support Rust gRPC code generation.
## What does this PR do?
**overview**:
- support Rust unary/streaming gRPC code generation
- add Rust&Java interop tests
Note: in local environment such as Ubuntu, run the test using virtual
environment:
```shell
cd integration_tests/grpc_tests
python3 -m venv .venv
source .venv/bin/activate
./run_tests.sh
```
**details**:
1. Before generating code, first validate that there is no
`thread_safe=false` ref usage in gRPC payload types since tonic requires
all payload types to be thread safe; and move naming collision logic for
service definition from `compiler/fory_compiler/generators/rust.py` to
`compiler/fory_compiler/generators/services/rust.py`.
2. For the given IDL, generate tonic-compatiable client and server stub
code, i.e. `<module_name>_service.rs`, `<module_name>_service_grpc.rs`.
**Note**: considering that codec logic is relatively small, codec code
is also generated as a separate `mod` inside
`<module_name>_service_grpc.rs`, instead of defining a new crate under
`rust/`.
<details>
<summary>Example</summary>
For this simple service definition:
```
package hello;
message HelloRequest {
string name = 1;
}
message HelloResponse {
string greeting = 1;
}
service Greeter {
rpc SayHello (HelloRequest) returns (HelloResponse);
}
```
The generated `service.rs` (ignore the copyright header):
```rust
#[::tonic::async_trait]
pub trait Greeter: ::std::marker::Send + ::std::marker::Sync + 'static {
async fn say_hello(
&self,
request: ::tonic::Request<crate::hello::HelloRequest>,
) -> ::std::result::Result<
::tonic::Response<crate::hello::HelloResponse>,
::tonic::Status,
>;
}
pub const GREETER_SERVICE_NAME: &str = "hello.Greeter";
pub const GREETER_SAY_HELLO_PATH: &str = "/hello.Greeter/SayHello";
```
The generated `service_grpc.rs` (also ignore the copyright header):
```rust
pub mod codec {
pub trait ForyGrpcPayload: Sized + ::std::marker::Send + 'static {
fn encode_fory_payload(
&self,
) -> ::std::result::Result<::std::vec::Vec<u8>, ::fory::Error>;
fn decode_fory_payload(
payload: &[u8],
) -> ::std::result::Result<Self, ::fory::Error>;
}
#[derive(Debug, Clone)]
pub struct ForyCodec<Encode, Decode> {
marker: ::std::marker::PhantomData<(Encode, Decode)>,
}
impl<Encode, Decode> ForyCodec<Encode, Decode> {
pub fn new() -> Self {
Self {
marker: ::std::marker::PhantomData,
}
}
}
impl<Encode, Decode> ::std::default::Default for ForyCodec<Encode,
Decode> {
fn default() -> Self {
Self::new()
}
}
impl<Encode, Decode> ::tonic::codec::Codec for ForyCodec<Encode, Decode>
where
Encode: ForyGrpcPayload,
Decode: ForyGrpcPayload,
{
type Encode = Encode;
type Decode = Decode;
type Encoder = ForyEncoder<Encode>;
type Decoder = ForyDecoder<Decode>;
fn encoder(&mut self) -> Self::Encoder {
ForyEncoder::default()
}
fn decoder(&mut self) -> Self::Decoder {
ForyDecoder::default()
}
}
#[derive(Debug, Clone)]
pub struct ForyEncoder<T> {
marker: ::std::marker::PhantomData<T>,
}
impl<T> ::std::default::Default for ForyEncoder<T> {
fn default() -> Self {
Self {
marker: ::std::marker::PhantomData,
}
}
}
#[derive(Debug, Clone)]
pub struct ForyDecoder<T> {
marker: ::std::marker::PhantomData<T>,
}
impl<T> ::std::default::Default for ForyDecoder<T> {
fn default() -> Self {
Self {
marker: ::std::marker::PhantomData,
}
}
}
fn fory_error_to_tonic_status(error: ::fory::Error) -> ::tonic::Status {
::tonic::Status::internal(error.to_string())
}
impl<T> ::tonic::codec::Encoder for ForyEncoder<T>
where
T: ForyGrpcPayload,
{
type Item = T;
type Error = ::tonic::Status;
fn encode(
&mut self,
item: Self::Item,
dst: &mut ::tonic::codec::EncodeBuf<'_>,
) -> ::std::result::Result<(), Self::Error> {
let bytes =
item.encode_fory_payload().map_err(fory_error_to_tonic_status)?;
::tonic::codec::EncodeBuf::reserve(dst, bytes.len());
::bytes::BufMut::put_slice(dst, &bytes);
Ok(())
}
}
impl<T> ::tonic::codec::Decoder for ForyDecoder<T>
where
T: ForyGrpcPayload,
{
type Item = T;
type Error = ::tonic::Status;
fn decode(
&mut self,
src: &mut ::tonic::codec::DecodeBuf<'_>,
) -> ::std::result::Result<::std::option::Option<Self::Item>,
Self::Error> {
let len = ::bytes::Buf::remaining(src);
if len == 0 {
return Ok(None);
}
let chunk = ::bytes::Buf::chunk(src);
if chunk.len() == len {
let result =
T::decode_fory_payload(chunk).map_err(fory_error_to_tonic_status);
::bytes::Buf::advance(src, len);
return result.map(Some);
}
let payload = ::bytes::Buf::copy_to_bytes(src, len);
T::decode_fory_payload(&payload)
.map(Some)
.map_err(fory_error_to_tonic_status)
}
}
}
impl codec::ForyGrpcPayload for crate::hello::HelloRequest {
fn encode_fory_payload(&self) ->
::std::result::Result<::std::vec::Vec<u8>, ::fory::Error> {
self.to_bytes()
}
fn decode_fory_payload(payload: &[u8]) -> ::std::result::Result<Self,
::fory::Error> {
Self::from_bytes(payload)
}
}
impl codec::ForyGrpcPayload for crate::hello::HelloResponse {
fn encode_fory_payload(&self) ->
::std::result::Result<::std::vec::Vec<u8>, ::fory::Error> {
self.to_bytes()
}
fn decode_fory_payload(payload: &[u8]) -> ::std::result::Result<Self,
::fory::Error> {
Self::from_bytes(payload)
}
}
pub mod greeter_client {
#[derive(Debug, Clone)]
pub struct GreeterClient<T> {
inner: ::tonic::client::Grpc<T>,
}
impl GreeterClient<::tonic::transport::Channel> {
pub async fn connect<D>(
dst: D,
) -> ::std::result::Result<Self, ::tonic::transport::Error>
where
D: ::std::convert::TryInto<::tonic::transport::Endpoint>,
D::Error: Into<::tonic::codegen::StdError>,
{
let conn = ::tonic::transport::Endpoint::new(dst)?.connect().await?;
Ok(Self::new(conn))
}
}
impl<T> GreeterClient<T>
where
T: ::tonic::client::GrpcService<::tonic::body::Body>,
T::Error: Into<::tonic::codegen::StdError>,
T::ResponseBody: ::tonic::codegen::Body<Data = ::tonic::codegen::Bytes>
+ ::std::marker::Send
+ 'static,
<T::ResponseBody as ::tonic::codegen::Body>::Error:
Into<::tonic::codegen::StdError> + ::std::marker::Send,
{
pub fn new(inner: T) -> Self {
let inner = ::tonic::client::Grpc::new(inner);
Self { inner }
}
pub async fn say_hello(
&mut self,
request: impl ::tonic::IntoRequest<crate::hello::HelloRequest>,
) -> ::std::result::Result<
::tonic::Response<crate::hello::HelloResponse>,
::tonic::Status,
> {
self.inner.ready().await.map_err(|e| {
::tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
})?;
let codec = super::codec::ForyCodec::<
crate::hello::HelloRequest,
crate::hello::HelloResponse,
>::default();
let path = ::tonic::codegen::http::uri::PathAndQuery::from_static(
crate::service::GREETER_SAY_HELLO_PATH,
);
let mut req = request.into_request();
req.extensions_mut().insert(::tonic::codegen::GrpcMethod::new(
crate::service::GREETER_SERVICE_NAME,
"SayHello",
));
self.inner.unary(req, path, codec).await
}
}
}
pub mod greeter_server {
#[derive(Debug)]
pub struct GreeterServer<T> {
inner: ::std::sync::Arc<T>,
}
impl<T> GreeterServer<T> {
pub fn new(inner: T) -> Self {
Self::from_arc(::std::sync::Arc::new(inner))
}
pub fn from_arc(inner: ::std::sync::Arc<T>) -> Self {
Self { inner }
}
}
impl<T, B> ::tonic::codegen::Service<::tonic::codegen::http::Request<B>>
for GreeterServer<T>
where
T: crate::service::Greeter,
B: ::tonic::codegen::Body + ::std::marker::Send + 'static,
B::Error: Into<::tonic::codegen::StdError> + ::std::marker::Send +
'static,
{
type Response = ::tonic::codegen::http::Response<::tonic::body::Body>;
type Error = ::std::convert::Infallible;
type Future = ::tonic::codegen::BoxFuture<Self::Response, Self::Error>;
fn poll_ready(
&mut self,
_cx: &mut ::std::task::Context<'_>,
) -> ::std::task::Poll<::std::result::Result<(), Self::Error>> {
::std::task::Poll::Ready(Ok(()))
}
fn call(&mut self, req: ::tonic::codegen::http::Request<B>) ->
Self::Future {
match req.uri().path() {
crate::service::GREETER_SAY_HELLO_PATH => {
struct SayHelloSvc<T: crate::service::Greeter>(pub ::std::sync::Arc<T>);
impl<T: crate::service::Greeter>
::tonic::server::UnaryService<crate::hello::HelloRequest>
for SayHelloSvc<T>
{
type Response = crate::hello::HelloResponse;
type Future = ::tonic::codegen::BoxFuture<
::tonic::Response<Self::Response>,
::tonic::Status,
>;
fn call(
&mut self,
request: ::tonic::Request<crate::hello::HelloRequest>,
) -> Self::Future {
let inner = ::std::sync::Arc::clone(&self.0);
let fut = async move {
<T as crate::service::Greeter>::say_hello(&inner, request).await
};
::std::boxed::Box::pin(fut)
}
}
let inner = self.inner.clone();
let fut = async move {
let method = SayHelloSvc(inner);
let codec = super::codec::ForyCodec::<
crate::hello::HelloResponse,
crate::hello::HelloRequest,
>::default();
let mut grpc = ::tonic::server::Grpc::new(codec);
let res = grpc.unary(method, req).await;
Ok(res)
};
::std::boxed::Box::pin(fut)
}
_ => ::std::boxed::Box::pin(async move {
let mut response = ::tonic::codegen::http::Response::new(
::tonic::body::Body::default(),
);
let headers = response.headers_mut();
headers.insert(
::tonic::Status::GRPC_STATUS,
(::tonic::Code::Unimplemented as i32).into(),
);
headers.insert(
::tonic::codegen::http::header::CONTENT_TYPE,
::tonic::metadata::GRPC_CONTENT_TYPE,
);
Ok(response)
}),
}
}
}
impl<T> ::std::clone::Clone for GreeterServer<T> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
pub const SERVICE_NAME: &str = crate::service::GREETER_SERVICE_NAME;
impl<T> ::tonic::server::NamedService for GreeterServer<T> {
const NAME: &'static str = SERVICE_NAME;
}
}
```
</details>
3. Add interop testing logic for Java and Rust, implement gRPC handlers
and assertions so that Rust&Java interop testing can work, and also
reserve a placeholder crate `generated` to hold the generated gRPC code.
**next:**
- add documentation when above implementation is stable.
- performance (mainly deserializaion) benchmarking and tuning.
## Related issues
https://github.com/apache/fory/issues/3275
## AI Contribution Checklist
- [ ] Substantial AI assistance was used in this PR: `yes` / `no`
- [ ] If `yes`, I included a completed [AI Contribution
Checklist](https://github.com/apache/fory/blob/main/AI_POLICY.md#9-contributor-checklist-for-ai-assisted-prs)
in this PR description and the required `AI Usage Disclosure`.
- [ ] If `yes`, my PR description includes the required `ai_review`
summary and screenshot evidence of the final clean AI review results
from both fresh reviewers on the current PR diff or current HEAD after
the latest code changes.
## Does this PR introduce any user-facing change?
N/A.
## Benchmark
N/A.
---
.github/workflows/ci.yml | 9 +-
.gitignore | 8 +
compiler/fory_compiler/generators/rust.py | 106 +--
compiler/fory_compiler/generators/services/rust.py | 900 ++++++++++++++++++++
.../fory_compiler/tests/test_service_codegen.py | 102 ++-
integration_tests/grpc_tests/generate_grpc.py | 2 +
.../apache/fory/grpc_tests/GrpcInteropTest.java | 118 ++-
integration_tests/grpc_tests/run_tests.sh | 1 +
.../grpc_tests/{run_tests.sh => rust/Cargo.toml} | 33 +-
.../{run_tests.sh => rust/generated/Cargo.toml} | 27 +-
.../grpc_tests/rust/generated/src/lib.rs | 26 +
.../{run_tests.sh => rust/interop/Cargo.toml} | 27 +-
.../grpc_tests/rust/interop/src/main.rs | 909 +++++++++++++++++++++
13 files changed, 2121 insertions(+), 147 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index d9c794f51..6506f5152 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -688,7 +688,7 @@ jobs:
run: python ./ci/run_ci.py java --version integration_tests
grpc_tests:
- name: Java/Python gRPC Tests
+ name: Java/Python/Go/Rust gRPC Tests
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v5
@@ -709,11 +709,16 @@ jobs:
key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
restore-keys: |
${{ runner.os }}-maven-
+ - name: Cache Cargo dependencies
+ uses: Swatinem/rust-cache@v2
+ with:
+ workspaces: integration_tests/grpc_tests/rust
+ cache-bin: "false"
- name: Install Java artifacts for gRPC tests
run: |
cd java
mvn -T16 --no-transfer-progress clean install -DskipTests
-Dmaven.javadoc.skip=true -Dmaven.source.skip=true
- - name: Run Java/Python gRPC Tests
+ - name: Run Java/Python/Go/Rust gRPC Tests
run: ./integration_tests/grpc_tests/run_tests.sh
javascript:
diff --git a/.gitignore b/.gitignore
index 55c36b011..ba7a5f9ff 100644
--- a/.gitignore
+++ b/.gitignore
@@ -38,6 +38,12 @@ bazel-*/
bazel-fory/
bazel-fory/**
**/generated
+!integration_tests/grpc_tests/rust/generated/
+integration_tests/grpc_tests/rust/generated/*
+!integration_tests/grpc_tests/rust/generated/Cargo.toml
+!integration_tests/grpc_tests/rust/generated/src/
+integration_tests/grpc_tests/rust/generated/src/*
+!integration_tests/grpc_tests/rust/generated/src/lib.rs
java/**/generated
javascript/**/dist/
@@ -78,6 +84,7 @@ bazel-genfiles//
# Go
go/**/build/
+integration_tests/grpc_tests/go/grpc-interop
# Rust
rust/target/
@@ -131,6 +138,7 @@ benchmarks/python/proto/
benchmarks/java/dependency-reduced-pom.xml
benchmarks/go/go.test
benchmarks/rust/Cargo.lock
+integration_tests/grpc_tests/rust/Cargo.lock
docs/superpowers
test.md
diff --git a/compiler/fory_compiler/generators/rust.py
b/compiler/fory_compiler/generators/rust.py
index 4569b26e6..54ba2f758 100644
--- a/compiler/fory_compiler/generators/rust.py
+++ b/compiler/fory_compiler/generators/rust.py
@@ -20,6 +20,7 @@
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple
+from fory_compiler.generators.services.rust import RustServiceGeneratorMixin
from fory_compiler.generators.base import BaseGenerator, GeneratedFile
from fory_compiler.frontend.utils import parse_idl_file
from fory_compiler.ir.ast import (
@@ -39,7 +40,7 @@ from fory_compiler.ir.ast import (
from fory_compiler.ir.types import PrimitiveKind
-class RustGenerator(BaseGenerator):
+class RustGenerator(RustServiceGeneratorMixin, BaseGenerator):
"""Generates Rust types with Fory derive macros."""
language_name = "rust"
@@ -461,87 +462,6 @@ class RustGenerator(BaseGenerator):
def generate(self) -> List[GeneratedFile]:
"""Generate Rust files for the schema."""
files = []
- if self.options.grpc:
- # Allocate and validate identifier naming for gRPC service
definition.
- self._ensure_name_caches(self.schema)
- schema_id = id(self.schema)
- if schema_id not in self._named_service_schema_ids:
- schema_source_key =
str(Path(self.schema.source_file).resolve())
- services = [
- service
- for service in self.schema.services
- if str(Path(service.location.file).resolve()) ==
schema_source_key
- ]
- used_traits: Dict[str, str] = {}
- used_modules: Dict[str, str] = {}
- used_constants: Dict[str, str] = {}
- for service in services:
- service_key = self._cache_key(service)
- self._service_trait_identifier_cache[service_key] = (
- self._allocate_scoped_identifier(
- self.to_pascal_case(service.name),
- used_traits,
- "Rust gRPC service traits",
- service.name,
- )
- )
- self._service_client_module_identifier_cache[service_key]
= (
- self._allocate_scoped_identifier(
- f"{self.to_snake_case(service.name)}_client",
- used_modules,
- "Rust gRPC service modules",
- f"{service.name} client module",
- )
- )
- self._service_server_module_identifier_cache[service_key]
= (
- self._allocate_scoped_identifier(
- f"{self.to_snake_case(service.name)}_server",
- used_modules,
- "Rust gRPC service modules",
- f"{service.name} server module",
- )
- )
- self._service_name_constant_identifier_cache[service_key]
= (
- self._allocate_scoped_identifier(
-
f"{self.to_upper_snake_case(service.name)}_SERVICE_NAME",
- used_constants,
- "Rust gRPC service constants",
- service.name,
- )
- )
- used_methods: Dict[str, str] = {}
- used_stream_types: Dict[str, str] = {}
- method_names: Dict[Tuple[object, ...], str] = {}
- stream_types: Dict[Tuple[object, ...], str] = {}
- path_constants: Dict[Tuple[object, ...], str] = {}
- for method in service.methods:
- method_key = self._cache_key(method)
- method_names[method_key] =
self._allocate_scoped_identifier(
- self.to_snake_case(method.name),
- used_methods,
- f"Rust gRPC service {service.name} methods",
- method.name,
- )
- if method.server_streaming:
- stream_types[method_key] =
self._allocate_scoped_identifier(
- f"{self.to_pascal_case(method.name)}Stream",
- used_stream_types,
- f"Rust gRPC service {service.name} stream
types",
- method.name,
- )
- path_constants[method_key] =
self._allocate_scoped_identifier(
- f"{self.to_upper_snake_case(service.name)}_"
- f"{self.to_upper_snake_case(method.name)}_PATH",
- used_constants,
- "Rust gRPC service constants",
- f"{service.name}.{method.name}",
- )
- self._rpc_method_identifier_cache[service_key] =
method_names
- self._rpc_stream_type_identifier_cache[service_key] =
stream_types
- self._rpc_path_constant_identifier_cache[service_key] = (
- path_constants
- )
- self._named_service_schema_ids.add(schema_id)
# Generate a single module file with all types
files.append(self.generate_module())
@@ -1063,6 +983,28 @@ class RustGenerator(BaseGenerator):
break
return lineage
+ def _parent_stack_for_type(self, type_def: object) -> List[Message]:
+ def visit(current: Message, parents: List[Message]) ->
Optional[List[Message]]:
+ if current is type_def:
+ return parents
+ for nested_union in current.nested_unions:
+ if nested_union is type_def:
+ return parents + [current]
+ for nested_enum in current.nested_enums:
+ if nested_enum is type_def:
+ return parents + [current]
+ for nested_message in current.nested_messages:
+ found = visit(nested_message, parents + [current])
+ if found is not None:
+ return found
+ return None
+
+ for top in self.schema.messages:
+ found = visit(top, [])
+ if found is not None:
+ return found
+ return []
+
def union_supports_trait(
self,
union: Union,
diff --git a/compiler/fory_compiler/generators/services/rust.py
b/compiler/fory_compiler/generators/services/rust.py
new file mode 100644
index 000000000..16cf5954e
--- /dev/null
+++ b/compiler/fory_compiler/generators/services/rust.py
@@ -0,0 +1,900 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+"""Rust gRPC service generator helpers."""
+
+from typing import Dict, List, Optional, Set, Tuple
+
+from fory_compiler.generators.base import GeneratedFile
+from fory_compiler.ir.ast import (
+ ArrayType,
+ Field,
+ FieldType,
+ ListType,
+ MapType,
+ Message,
+ NamedType,
+ PrimitiveType,
+ RpcMethod,
+ Service,
+ Union,
+)
+
+
+class RustServiceGeneratorMixin:
+ """Generates Rust gRPC service companions."""
+
+ def generate_services(self) -> List[GeneratedFile]:
+ """Generate Rust grpc service companion module."""
+ local_services = [
+ service
+ for service in self.schema.services
+ if not self.is_imported_type(service)
+ ]
+ if not local_services:
+ return []
+ self.validate_grpc_payload_safe(local_services)
+ self.allocate_grpc_service_identifiers(local_services)
+ return [
+ self.generate_service_api_module(local_services),
+ self.generate_service_grpc_module(local_services),
+ ]
+
+ def allocate_grpc_service_identifiers(self, services: List[Service]) ->
None:
+ """Allocate sanitized identifiers used by generated gRPC code."""
+ self._ensure_name_caches(self.schema)
+ schema_id = id(self.schema)
+ if schema_id in self._named_service_schema_ids:
+ return
+ used_traits: Dict[str, str] = {}
+ used_modules: Dict[str, str] = {}
+ used_constants: Dict[str, str] = {}
+ for service in services:
+ service_key = self._cache_key(service)
+ self._service_trait_identifier_cache[service_key] = (
+ self._allocate_scoped_identifier(
+ self.to_pascal_case(service.name),
+ used_traits,
+ "Rust gRPC service traits",
+ service.name,
+ )
+ )
+ self._service_client_module_identifier_cache[service_key] = (
+ self._allocate_scoped_identifier(
+ f"{self.to_snake_case(service.name)}_client",
+ used_modules,
+ "Rust gRPC service modules",
+ f"{service.name} client module",
+ )
+ )
+ self._service_server_module_identifier_cache[service_key] = (
+ self._allocate_scoped_identifier(
+ f"{self.to_snake_case(service.name)}_server",
+ used_modules,
+ "Rust gRPC service modules",
+ f"{service.name} server module",
+ )
+ )
+ self._service_name_constant_identifier_cache[service_key] = (
+ self._allocate_scoped_identifier(
+ f"{self.to_upper_snake_case(service.name)}_SERVICE_NAME",
+ used_constants,
+ "Rust gRPC service constants",
+ service.name,
+ )
+ )
+ used_methods: Dict[str, str] = {}
+ used_stream_types: Dict[str, str] = {}
+ method_names: Dict[Tuple[object, ...], str] = {}
+ stream_types: Dict[Tuple[object, ...], str] = {}
+ path_constants: Dict[Tuple[object, ...], str] = {}
+ for method in service.methods:
+ method_key = self._cache_key(method)
+ method_name = self.to_snake_case(method.name)
+ # Client stub code uses `new` and `connect` as helper method
names.
+ if method_name in {"connect", "new"}:
+ method_name = f"{method_name}_"
+ method_names[method_key] = self._allocate_scoped_identifier(
+ method_name,
+ used_methods,
+ f"Rust gRPC service {service.name} methods",
+ method.name,
+ )
+ if method.server_streaming:
+ stream_types[method_key] =
self._allocate_scoped_identifier(
+ f"{self.to_pascal_case(method.name)}Stream",
+ used_stream_types,
+ f"Rust gRPC service {service.name} stream types",
+ method.name,
+ )
+ path_constants[method_key] = self._allocate_scoped_identifier(
+ f"{self.to_upper_snake_case(service.name)}_"
+ f"{self.to_upper_snake_case(method.name)}_PATH",
+ used_constants,
+ "Rust gRPC service constants",
+ f"{service.name}.{method.name}",
+ )
+ self._rpc_method_identifier_cache[service_key] = method_names
+ self._rpc_stream_type_identifier_cache[service_key] = stream_types
+ self._rpc_path_constant_identifier_cache[service_key] =
path_constants
+ self._named_service_schema_ids.add(schema_id)
+
+ def grpc_payload_root_names(self, services: List[Service]) -> Set[str]:
+ names: Set[str] = set()
+ for service in services:
+ for method in service.methods:
+ names.add(method.request_type.name)
+ names.add(method.response_type.name)
+ return names
+
+ def validate_grpc_payload_safe(self, services: List[Service]) -> None:
+ # Make sure all gRPC payload types are thread-safe. For now only
reject pointer with `thread_safe=false`.
+ reachable: Set[int] = set()
+
+ def reject_explicit_non_thread_safe_ref(
+ path: List[str], description: str, ref_options: dict
+ ) -> None:
+ if ref_options.get("thread_safe_pointer") is False:
+ raise ValueError(
+ f"Rust gRPC payload type {'.'.join(path)} uses
non-thread-safe {description}; "
+ "remove thread_safe=false or use the default thread-safe
pointer"
+ )
+
+ def visit_named(
+ type_name: str,
+ path: List[str],
+ parent_stack: Optional[List[Message]] = None,
+ ) -> None:
+ resolved = self.resolve_named_type(type_name, parent_stack)
+ if resolved is None:
+ return
+ key = id(resolved)
+ if key in reachable:
+ return
+ reachable.add(key)
+ owner_stack = self._parent_stack_for_type(resolved)
+ if isinstance(resolved, Message):
+ lineage = owner_stack + [resolved]
+ for field in resolved.fields:
+ visit_field(field, path + [field.name], lineage)
+ elif isinstance(resolved, Union):
+ for field in resolved.fields:
+ visit_field(
+ field,
+ path + [self.to_pascal_case(field.name)],
+ owner_stack,
+ )
+
+ def visit_field(
+ field: Field,
+ path: List[str],
+ parent_stack: Optional[List[Message]],
+ ) -> None:
+ if field.ref:
+ reject_explicit_non_thread_safe_ref(path, "ref",
field.ref_options)
+ if isinstance(field.field_type, ListType) and field.element_ref:
+ reject_explicit_non_thread_safe_ref(
+ path, "list element ref", field.element_ref_options
+ )
+ visit_field_type(field.field_type, path, parent_stack)
+
+ def visit_field_type(
+ field_type: FieldType,
+ path: List[str],
+ parent_stack: Optional[List[Message]],
+ ) -> None:
+ if isinstance(field_type, PrimitiveType):
+ return
+ if isinstance(field_type, NamedType):
+ visit_named(field_type.name, path + [field_type.name],
parent_stack)
+ elif isinstance(field_type, ListType):
+ if field_type.element_ref:
+ reject_explicit_non_thread_safe_ref(
+ path, "list element ref",
field_type.element_ref_options
+ )
+ visit_field_type(field_type.element_type, path, parent_stack)
+ elif isinstance(field_type, ArrayType):
+ visit_field_type(field_type.element_type, path, parent_stack)
+ elif isinstance(field_type, MapType):
+ if field_type.value_ref:
+ reject_explicit_non_thread_safe_ref(
+ path, "map value ref", field_type.value_ref_options
+ )
+ visit_field_type(field_type.key_type, path, parent_stack)
+ visit_field_type(field_type.value_type, path, parent_stack)
+
+ for root in self.grpc_payload_root_names(services):
+ visit_named(root, [root])
+
+ def generate_service_api_module(self, services: List[Service]) ->
GeneratedFile:
+ """Generate Rust service API module (service.rs)."""
+ lines: List[str] = []
+ lines.append(self.get_license_header("//"))
+ lines.append("")
+ for i, service in enumerate(services):
+ if i > 0:
+ lines.append("")
+ lines.extend(self.generate_service_trait(service))
+ lines.append("")
+ lines.extend(self.generate_service_constants(service))
+ return GeneratedFile(
+ path=f"{self.get_module_name()}_service.rs",
+ content="\n".join(lines).rstrip() + "\n",
+ )
+
+ def generate_service_trait(self, service: Service) -> List[str]:
+ """Generate tonic service trait."""
+ service_key = self._cache_key(service)
+ trait_name = self._service_trait_identifier_cache[service_key]
+ lines: List[str] = []
+ lines.append("#[::tonic::async_trait]")
+ lines.append(
+ f"pub trait {trait_name}: ::std::marker::Send +
::std::marker::Sync + 'static {{"
+ )
+ for i, method in enumerate(service.methods):
+ if i > 0:
+ lines.append("")
+ lines.extend(
+ self.indent_lines(
+ self.generate_grpc_method_signature(service, method), 1
+ )
+ )
+ lines.append("}")
+ return lines
+
+ def generate_grpc_method_signature(
+ self, service: Service, method: RpcMethod
+ ) -> List[str]:
+ """Generate tonic service trait method signature."""
+ service_key = self._cache_key(service)
+ method_key = self._cache_key(method)
+ method_name =
self._rpc_method_identifier_cache[service_key][method_key]
+ request_type = self.service_type_path(method.request_type)
+ response_type = self.service_type_path(method.response_type)
+ lines: List[str] = []
+ if method.client_streaming:
+ request_arg_type =
f"::tonic::Request<::tonic::Streaming<{request_type}>>"
+ else:
+ request_arg_type = f"::tonic::Request<{request_type}>"
+ if method.server_streaming:
+ stream_type = self._rpc_stream_type_identifier_cache[service_key][
+ method_key
+ ]
+ response_return_type = f"::tonic::Response<Self::{stream_type}>"
+ lines.extend(
+ [
+ f"type {stream_type}:
::tonic::codegen::tokio_stream::Stream<",
+ f" Item = ::std::result::Result<{response_type},
::tonic::Status>,",
+ "> + ::std::marker::Send + 'static;",
+ "",
+ ]
+ )
+ else:
+ response_return_type = f"::tonic::Response<{response_type}>"
+ lines.extend(
+ [
+ f"async fn {method_name}(",
+ " &self,",
+ f" request: {request_arg_type},",
+ ") -> ::std::result::Result<",
+ f" {response_return_type},",
+ " ::tonic::Status,",
+ ">;",
+ ]
+ )
+ return lines
+
+ def service_type_path(self, named_type: NamedType) -> str:
+ """Get Rust path for a gRPC request or response type."""
+ resolved = self.resolve_named_type(named_type.name, parent_stack=None)
+ if resolved is None:
+ raise ValueError(f"Unknown gRPC payload type {named_type.name!r}")
+ type_path = self.get_type_path(resolved,
self._parent_stack_for_type(resolved))
+ if self.is_imported_type(resolved):
+ module = self._module_name_for_type(resolved)
+ return f"crate::{module}::{type_path}"
+ module = self.get_top_level_module_identifier(self.package)
+ return f"crate::{module}::{type_path}"
+
+ def generate_service_constants(self, service: Service) -> List[str]:
+ """Generate service and gRPC path constants."""
+ # e.g. `pub const GREETER_SERVICE_NAME: &str =
"demo.greeter.Greeter";` and
+ # `pub const GREETER_SAY_HELLO_PATH: &str =
"/demo.greeter.Greeter/SayHello";`
+ lines: List[str] = []
+ service_key = self._cache_key(service)
+ service_name_const =
self._service_name_constant_identifier_cache[service_key]
+ service_name = self.get_grpc_service_name(service)
+ lines.append(
+ f"pub const {service_name_const}: &str = "
+ f"{self.rust_string_literal(service_name)};"
+ )
+ for method in service.methods:
+ method_key = self._cache_key(method)
+ rpc_path_const =
self._rpc_path_constant_identifier_cache[service_key][
+ method_key
+ ]
+ lines.append(
+ f"pub const {rpc_path_const}: &str = "
+ f"{self.rust_string_literal(self.get_grpc_method_path(service,
method))};"
+ )
+ return lines
+
+ def generate_service_grpc_module(self, services: List[Service]) ->
GeneratedFile:
+ """Generate Rust service tonic binding module (service_grpc.rs)."""
+ lines: List[str] = []
+ lines.append(self.get_license_header("//"))
+ lines.append("")
+ lines.extend(self.generate_grpc_codec_module())
+ payload_impls = self.generate_grpc_payload_impls(services)
+ if payload_impls:
+ lines.append("")
+ lines.extend(payload_impls)
+ for service in services:
+ lines.append("")
+ lines.extend(self.generate_grpc_client_module(service))
+ lines.append("")
+ lines.extend(self.generate_grpc_server_module(service))
+ return GeneratedFile(
+ path=f"{self.get_module_name()}_service_grpc.rs",
+ content="\n".join(lines).rstrip() + "\n",
+ )
+
+ def generate_grpc_codec_module(self) -> List[str]:
+ """Generate Fory-backed codec used by Rust gRPC stubs, instead of
relying on a separate crate."""
+ return [
+ "pub mod codec {",
+ " pub trait ForyGrpcPayload: Sized + ::std::marker::Send +
'static {",
+ " fn encode_fory_payload(",
+ " &self,",
+ " ) -> ::std::result::Result<::std::vec::Vec<u8>,
::fory::Error>;",
+ "",
+ " fn decode_fory_payload(",
+ " payload: &[u8],",
+ " ) -> ::std::result::Result<Self, ::fory::Error>;",
+ " }",
+ "",
+ " #[derive(Debug, Clone)]",
+ " pub struct ForyCodec<Encode, Decode> {",
+ " marker: ::std::marker::PhantomData<(Encode, Decode)>,",
+ " }",
+ "",
+ " impl<Encode, Decode> ForyCodec<Encode, Decode> {",
+ " pub fn new() -> Self {",
+ " Self {",
+ " marker: ::std::marker::PhantomData,",
+ " }",
+ " }",
+ " }",
+ "",
+ " impl<Encode, Decode> ::std::default::Default for
ForyCodec<Encode, Decode> {",
+ " fn default() -> Self {",
+ " Self::new()",
+ " }",
+ " }",
+ "",
+ " impl<Encode, Decode> ::tonic::codec::Codec for
ForyCodec<Encode, Decode>",
+ " where",
+ " Encode: ForyGrpcPayload,",
+ " Decode: ForyGrpcPayload,",
+ " {",
+ " type Encode = Encode;",
+ " type Decode = Decode;",
+ " type Encoder = ForyEncoder<Encode>;",
+ " type Decoder = ForyDecoder<Decode>;",
+ "",
+ " fn encoder(&mut self) -> Self::Encoder {",
+ " ForyEncoder::default()",
+ " }",
+ "",
+ " fn decoder(&mut self) -> Self::Decoder {",
+ " ForyDecoder::default()",
+ " }",
+ " }",
+ "",
+ " #[derive(Debug, Clone)]",
+ " pub struct ForyEncoder<T> {",
+ " marker: ::std::marker::PhantomData<T>,",
+ " }",
+ "",
+ " impl<T> ::std::default::Default for ForyEncoder<T> {",
+ " fn default() -> Self {",
+ " Self {",
+ " marker: ::std::marker::PhantomData,",
+ " }",
+ " }",
+ " }",
+ "",
+ " #[derive(Debug, Clone)]",
+ " pub struct ForyDecoder<T> {",
+ " marker: ::std::marker::PhantomData<T>,",
+ " }",
+ "",
+ " impl<T> ::std::default::Default for ForyDecoder<T> {",
+ " fn default() -> Self {",
+ " Self {",
+ " marker: ::std::marker::PhantomData,",
+ " }",
+ " }",
+ " }",
+ "",
+ " fn fory_error_to_tonic_status(error: ::fory::Error) ->
::tonic::Status {",
+ " ::tonic::Status::internal(error.to_string())",
+ " }",
+ "",
+ " impl<T> ::tonic::codec::Encoder for ForyEncoder<T>",
+ " where",
+ " T: ForyGrpcPayload,",
+ " {",
+ " type Item = T;",
+ " type Error = ::tonic::Status;",
+ "",
+ " fn encode(",
+ " &mut self,",
+ " item: Self::Item,",
+ " dst: &mut ::tonic::codec::EncodeBuf<'_>,",
+ " ) -> ::std::result::Result<(), Self::Error> {",
+ " let bytes =
item.encode_fory_payload().map_err(fory_error_to_tonic_status)?;",
+ " ::tonic::codec::EncodeBuf::reserve(dst,
bytes.len());",
+ " ::bytes::BufMut::put_slice(dst, &bytes);",
+ " Ok(())",
+ " }",
+ " }",
+ "",
+ " impl<T> ::tonic::codec::Decoder for ForyDecoder<T>",
+ " where",
+ " T: ForyGrpcPayload,",
+ " {",
+ " type Item = T;",
+ " type Error = ::tonic::Status;",
+ "",
+ " fn decode(",
+ " &mut self,",
+ " src: &mut ::tonic::codec::DecodeBuf<'_>,",
+ " ) ->
::std::result::Result<::std::option::Option<Self::Item>, Self::Error> {",
+ " let len = ::bytes::Buf::remaining(src);",
+ " if len == 0 {",
+ " return Ok(None);",
+ " }",
+ "",
+ " let chunk = ::bytes::Buf::chunk(src);",
+ " if chunk.len() == len {",
+ " let result =",
+ "
T::decode_fory_payload(chunk).map_err(fory_error_to_tonic_status);",
+ " ::bytes::Buf::advance(src, len);",
+ " return result.map(Some);",
+ " }",
+ "",
+ " let payload = ::bytes::Buf::copy_to_bytes(src, len);",
+ " T::decode_fory_payload(&payload)",
+ " .map(Some)",
+ " .map_err(fory_error_to_tonic_status)",
+ " }",
+ " }",
+ "}",
+ ]
+
+ def generate_grpc_payload_impls(self, services: List[Service]) ->
List[str]:
+ """Generate `ForyGrpcPayload` trait impls for all gRPC payload
types."""
+ lines: List[str] = []
+ for i, type_path in enumerate(self.grpc_payload_type_paths(services)):
+ if i > 0:
+ lines.append("")
+ lines.extend(self.generate_grpc_payload_impl(type_path))
+ return lines
+
+ def grpc_payload_type_paths(self, services: List[Service]) -> List[str]:
+ """Get unique request and response type paths in service order."""
+ seen: Set[str] = set()
+ type_paths: List[str] = []
+ for service in services:
+ for method in service.methods:
+ for named_type in (method.request_type, method.response_type):
+ type_path = self.service_type_path(named_type)
+ if type_path in seen:
+ continue
+ seen.add(type_path)
+ type_paths.append(type_path)
+ return type_paths
+
+ def generate_grpc_payload_impl(self, type_path: str) -> List[str]:
+ """Generate `ForyGrpcPayload` trait impl for one gRPC request or
response type."""
+ return [
+ f"impl codec::ForyGrpcPayload for {type_path} {{",
+ " fn encode_fory_payload(&self) ->
::std::result::Result<::std::vec::Vec<u8>, ::fory::Error> {",
+ " self.to_bytes()",
+ " }",
+ "",
+ " fn decode_fory_payload(payload: &[u8]) ->
::std::result::Result<Self, ::fory::Error> {",
+ " Self::from_bytes(payload)",
+ " }",
+ "}",
+ ]
+
+ def generate_grpc_client_module(self, service: Service) -> List[str]:
+ """Generate tonic client module for one service."""
+ lines: List[str] = []
+ service_key = self._cache_key(service)
+ client_module =
self._service_client_module_identifier_cache[service_key]
+ lines.append(f"pub mod {client_module} {{")
+
lines.extend(self.indent_lines(self.generate_grpc_client_struct(service), 1))
+ lines.append("")
+ lines.extend(
+ self.indent_lines(self.generate_grpc_client_connect_impl(service),
1)
+ )
+ lines.append("")
+
lines.extend(self.indent_lines(self.generate_grpc_client_impl(service), 1))
+ lines.append("}")
+ return lines
+
+ def generate_grpc_client_struct(self, service: Service) -> List[str]:
+ """Generate tonic client wrapper struct."""
+ service_key = self._cache_key(service)
+ client_name =
f"{self._service_trait_identifier_cache[service_key]}Client"
+ return [
+ "#[derive(Debug, Clone)]",
+ f"pub struct {client_name}<T> {{",
+ " inner: ::tonic::client::Grpc<T>,",
+ "}",
+ ]
+
+ def generate_grpc_client_connect_impl(self, service: Service) -> List[str]:
+ """Generate tonic channel connect constructor."""
+ service_key = self._cache_key(service)
+ client_name =
f"{self._service_trait_identifier_cache[service_key]}Client"
+ return [
+ f"impl {client_name}<::tonic::transport::Channel> {{",
+ " pub async fn connect<D>(",
+ " dst: D,",
+ " ) -> ::std::result::Result<Self, ::tonic::transport::Error>",
+ " where",
+ " D:
::std::convert::TryInto<::tonic::transport::Endpoint>,",
+ " D::Error: Into<::tonic::codegen::StdError>,",
+ " {",
+ " let conn =
::tonic::transport::Endpoint::new(dst)?.connect().await?;",
+ " Ok(Self::new(conn))",
+ " }",
+ "}",
+ ]
+
+ def generate_grpc_client_impl(self, service: Service) -> List[str]:
+ """Generate generic tonic client implementation."""
+ service_key = self._cache_key(service)
+ client_name =
f"{self._service_trait_identifier_cache[service_key]}Client"
+ lines: List[str] = [
+ f"impl<T> {client_name}<T>",
+ "where",
+ " T: ::tonic::client::GrpcService<::tonic::body::Body>,",
+ " T::Error: Into<::tonic::codegen::StdError>,",
+ " T::ResponseBody: ::tonic::codegen::Body<Data =
::tonic::codegen::Bytes>",
+ " + ::std::marker::Send",
+ " + 'static,",
+ " <T::ResponseBody as ::tonic::codegen::Body>::Error:",
+ " Into<::tonic::codegen::StdError> + ::std::marker::Send,",
+ "{",
+ " pub fn new(inner: T) -> Self {",
+ " let inner = ::tonic::client::Grpc::new(inner);",
+ " Self { inner }",
+ " }",
+ ]
+ for method in service.methods:
+ lines.append("")
+ lines.extend(
+ self.indent_lines(self.generate_grpc_client_method(service,
method), 1)
+ )
+ lines.append("}")
+ return lines
+
+ def generate_grpc_client_method(
+ self, service: Service, method: RpcMethod
+ ) -> List[str]:
+ """Generate a tonic client method."""
+ service_key = self._cache_key(service)
+ method_key = self._cache_key(method)
+ method_name =
self._rpc_method_identifier_cache[service_key][method_key]
+ request_type = self.service_type_path(method.request_type)
+ response_type = self.service_type_path(method.response_type)
+ if method.client_streaming:
+ request_arg_type = (
+ f"impl ::tonic::IntoStreamingRequest<Message = {request_type}>"
+ )
+ request_into_method = "into_streaming_request"
+ else:
+ request_arg_type = f"impl ::tonic::IntoRequest<{request_type}>"
+ request_into_method = "into_request"
+ if method.server_streaming:
+ response_return_type = (
+
f"::tonic::Response<::tonic::codec::Streaming<{response_type}>>"
+ )
+ else:
+ response_return_type = f"::tonic::Response<{response_type}>"
+ if method.client_streaming and method.server_streaming:
+ grpc_call = "streaming"
+ elif method.client_streaming:
+ grpc_call = "client_streaming"
+ elif method.server_streaming:
+ grpc_call = "server_streaming"
+ else:
+ grpc_call = "unary"
+ service_module = f"crate::{self.get_module_name()}_service"
+ service_name_const =
f"{service_module}::{self._service_name_constant_identifier_cache[service_key]}"
+ rpc_path_const =
f"{service_module}::{self._rpc_path_constant_identifier_cache[service_key][method_key]}"
+ return [
+ f"pub async fn {method_name}(",
+ " &mut self,",
+ f" request: {request_arg_type},",
+ ") -> ::std::result::Result<",
+ f" {response_return_type},",
+ " ::tonic::Status,",
+ "> {",
+ " self.inner.ready().await.map_err(|e| {",
+ ' ::tonic::Status::unknown(format!("Service was not ready:
{}", e.into()))',
+ " })?;",
+ " let codec = super::codec::ForyCodec::<",
+ f" {request_type},",
+ f" {response_type},",
+ " >::default();",
+ " let path =
::tonic::codegen::http::uri::PathAndQuery::from_static(",
+ f" {rpc_path_const},",
+ " );",
+ f" let mut req = request.{request_into_method}();",
+ "
req.extensions_mut().insert(::tonic::codegen::GrpcMethod::new(",
+ f" {service_name_const},",
+ f" {self.rust_string_literal(method.name)},",
+ " ));",
+ f" self.inner.{grpc_call}(req, path, codec).await",
+ "}",
+ ]
+
+ def generate_grpc_server_module(self, service: Service) -> List[str]:
+ """Generate tonic server module for one service."""
+ lines: List[str] = []
+ service_key = self._cache_key(service)
+ server_module =
self._service_server_module_identifier_cache[service_key]
+ lines.append(f"pub mod {server_module} {{")
+
lines.extend(self.indent_lines(self.generate_grpc_server_struct(service), 1))
+ lines.append("")
+
lines.extend(self.indent_lines(self.generate_grpc_server_impl(service), 1))
+ lines.append("")
+ lines.extend(
+ self.indent_lines(self.generate_grpc_server_service_impl(service),
1)
+ )
+ lines.append("")
+ lines.extend(
+ self.indent_lines(self.generate_grpc_server_clone_impl(service), 1)
+ )
+ lines.append("")
+ lines.extend(
+ self.indent_lines(self.generate_grpc_server_named_impl(service), 1)
+ )
+ lines.append("}")
+ return lines
+
+ def generate_grpc_server_struct(self, service: Service) -> List[str]:
+ """Generate tonic server wrapper struct."""
+ service_key = self._cache_key(service)
+ server_name =
f"{self._service_trait_identifier_cache[service_key]}Server"
+ return [
+ "#[derive(Debug)]",
+ f"pub struct {server_name}<T> {{",
+ " inner: ::std::sync::Arc<T>,",
+ "}",
+ ]
+
+ def generate_grpc_server_impl(self, service: Service) -> List[str]:
+ """Generate constructors for the tonic server wrapper."""
+ service_key = self._cache_key(service)
+ server_name =
f"{self._service_trait_identifier_cache[service_key]}Server"
+ return [
+ f"impl<T> {server_name}<T> {{",
+ " pub fn new(inner: T) -> Self {",
+ " Self::from_arc(::std::sync::Arc::new(inner))",
+ " }",
+ "",
+ " pub fn from_arc(inner: ::std::sync::Arc<T>) -> Self {",
+ " Self { inner }",
+ " }",
+ "}",
+ ]
+
+ def generate_grpc_server_service_impl(self, service: Service) -> List[str]:
+ """Generate tonic Service impl for the server wrapper."""
+ service_key = self._cache_key(service)
+ service_trait_identifier =
self._service_trait_identifier_cache[service_key]
+ server_name = f"{service_trait_identifier}Server"
+ service_trait = (
+
f"crate::{self.get_module_name()}_service::{service_trait_identifier}"
+ )
+ lines: List[str] = [
+ (
+ f"impl<T, B> ::tonic::codegen::Service<"
+ f"::tonic::codegen::http::Request<B>> for {server_name}<T>"
+ ),
+ "where",
+ f" T: {service_trait},",
+ " B: ::tonic::codegen::Body + ::std::marker::Send + 'static,",
+ " B::Error: Into<::tonic::codegen::StdError> +
::std::marker::Send + 'static,",
+ "{",
+ " type Response =
::tonic::codegen::http::Response<::tonic::body::Body>;",
+ " type Error = ::std::convert::Infallible;",
+ " type Future = ::tonic::codegen::BoxFuture<Self::Response,
Self::Error>;",
+ "",
+ " fn poll_ready(",
+ " &mut self,",
+ " _cx: &mut ::std::task::Context<'_>,",
+ " ) -> ::std::task::Poll<::std::result::Result<(),
Self::Error>> {",
+ " ::std::task::Poll::Ready(Ok(()))",
+ " }",
+ "",
+ (
+ " fn call("
+ "&mut self, req: ::tonic::codegen::http::Request<B>"
+ ") -> Self::Future {"
+ ),
+ " match req.uri().path() {",
+ ]
+ for method in service.methods:
+ lines.extend(
+ self.indent_lines(self.generate_grpc_server_route(service,
method), 3)
+ )
+ lines.extend(
+ self.indent_lines(self.generate_grpc_server_unimplemented_route(),
3)
+ )
+ lines.extend(
+ [
+ " }",
+ " }",
+ "}",
+ ]
+ )
+ return lines
+
+ def generate_grpc_server_route(
+ self, service: Service, method: RpcMethod
+ ) -> List[str]:
+ """Generate one tonic server route arm."""
+ service_key = self._cache_key(service)
+ method_key = self._cache_key(method)
+ service_trait_identifier =
self._service_trait_identifier_cache[service_key]
+ method_name =
self._rpc_method_identifier_cache[service_key][method_key]
+ service_trait = (
+
f"crate::{self.get_module_name()}_service::{service_trait_identifier}"
+ )
+ request_type = self.service_type_path(method.request_type)
+ response_type = self.service_type_path(method.response_type)
+ method_service_name_base = (
+ method_name[2:] if method_name.startswith("r#") else method_name
+ )
+ method_service_name =
f"{self.to_pascal_case(method_service_name_base)}Svc"
+ rpc_path_const =
f"crate::{self.get_module_name()}_service::{self._rpc_path_constant_identifier_cache[service_key][method_key]}"
+ if method.client_streaming:
+ request_arg_type =
f"::tonic::Request<::tonic::Streaming<{request_type}>>"
+ else:
+ request_arg_type = f"::tonic::Request<{request_type}>"
+ if method.client_streaming and method.server_streaming:
+ tonic_service_trait = "StreamingService"
+ grpc_call = "streaming"
+ elif method.client_streaming:
+ tonic_service_trait = "ClientStreamingService"
+ grpc_call = "client_streaming"
+ elif method.server_streaming:
+ tonic_service_trait = "ServerStreamingService"
+ grpc_call = "server_streaming"
+ else:
+ tonic_service_trait = "UnaryService"
+ grpc_call = "unary"
+ lines = [
+ f"{rpc_path_const} => {{",
+ (
+ f" struct {method_service_name}<T: {service_trait}>"
+ "(pub ::std::sync::Arc<T>);"
+ ),
+ "",
+ f" impl<T: {service_trait}>
::tonic::server::{tonic_service_trait}<{request_type}>",
+ f" for {method_service_name}<T>",
+ " {",
+ f" type Response = {response_type};",
+ ]
+ if method.server_streaming:
+ stream_type = self._rpc_stream_type_identifier_cache[service_key][
+ method_key
+ ]
+ future_response_type = "Self::ResponseStream"
+ lines.append(f" type ResponseStream = T::{stream_type};")
+ else:
+ future_response_type = "Self::Response"
+ lines.extend(
+ [
+ " type Future = ::tonic::codegen::BoxFuture<",
+ f" ::tonic::Response<{future_response_type}>,",
+ " ::tonic::Status,",
+ " >;",
+ "",
+ " fn call(",
+ " &mut self,",
+ f" request: {request_arg_type},",
+ " ) -> Self::Future {",
+ " let inner = ::std::sync::Arc::clone(&self.0);",
+ " let fut = async move {",
+ f" <T as
{service_trait}>::{method_name}(&inner, request).await",
+ " };",
+ " ::std::boxed::Box::pin(fut)",
+ " }",
+ " }",
+ "",
+ " let inner = self.inner.clone();",
+ " let fut = async move {",
+ f" let method = {method_service_name}(inner);",
+ " let codec = super::codec::ForyCodec::<",
+ f" {response_type},",
+ f" {request_type},",
+ " >::default();",
+ " let mut grpc = ::tonic::server::Grpc::new(codec);",
+ f" let res = grpc.{grpc_call}(method, req).await;",
+ " Ok(res)",
+ " };",
+ " ::std::boxed::Box::pin(fut)",
+ "}",
+ ]
+ )
+ return lines
+
+ def generate_grpc_server_unimplemented_route(self) -> List[str]:
+ """Generate fallback route for unknown gRPC paths."""
+ return [
+ "_ => ::std::boxed::Box::pin(async move {",
+ " let mut response = ::tonic::codegen::http::Response::new(",
+ " ::tonic::body::Body::default(),",
+ " );",
+ " let headers = response.headers_mut();",
+ " headers.insert(",
+ " ::tonic::Status::GRPC_STATUS,",
+ " (::tonic::Code::Unimplemented as i32).into(),",
+ " );",
+ " headers.insert(",
+ " ::tonic::codegen::http::header::CONTENT_TYPE,",
+ " ::tonic::metadata::GRPC_CONTENT_TYPE,",
+ " );",
+ " Ok(response)",
+ "}),",
+ ]
+
+ def generate_grpc_server_clone_impl(self, service: Service) -> List[str]:
+ """Generate Clone impl for the tonic server wrapper."""
+ service_key = self._cache_key(service)
+ server_name =
f"{self._service_trait_identifier_cache[service_key]}Server"
+ return [
+ f"impl<T> ::std::clone::Clone for {server_name}<T> {{",
+ " fn clone(&self) -> Self {",
+ " Self {",
+ " inner: self.inner.clone(),",
+ " }",
+ " }",
+ "}",
+ ]
+
+ def generate_grpc_server_named_impl(self, service: Service) -> List[str]:
+ """Generate NamedService impl for the tonic server wrapper."""
+ service_key = self._cache_key(service)
+ server_name =
f"{self._service_trait_identifier_cache[service_key]}Server"
+ service_name_const =
f"crate::{self.get_module_name()}_service::{self._service_name_constant_identifier_cache[service_key]}"
+ return [
+ f"pub const SERVICE_NAME: &str = {service_name_const};",
+ "",
+ f"impl<T> ::tonic::server::NamedService for {server_name}<T> {{",
+ " const NAME: &'static str = SERVICE_NAME;",
+ "}",
+ ]
diff --git a/compiler/fory_compiler/tests/test_service_codegen.py
b/compiler/fory_compiler/tests/test_service_codegen.py
index 58269d0f7..8170b9988 100644
--- a/compiler/fory_compiler/tests/test_service_codegen.py
+++ b/compiler/fory_compiler/tests/test_service_codegen.py
@@ -21,6 +21,8 @@ from pathlib import Path
from textwrap import dedent
from typing import Dict, Tuple, Type
+import pytest
+
from fory_compiler.cli import compile_file, resolve_imports
from fory_compiler.frontend.fdl.lexer import Lexer
from fory_compiler.frontend.fdl.parser import Parser
@@ -129,7 +131,12 @@ def
test_service_definition_does_not_affect_message_codegen():
def test_generate_services_returns_empty_list_for_unsupported_generators():
schema = parse_fdl(_GREETER_WITH_SERVICE)
for generator_cls in GENERATOR_CLASSES:
- if generator_cls in (JavaGenerator, PythonGenerator, GoGenerator):
+ if generator_cls in (
+ JavaGenerator,
+ PythonGenerator,
+ GoGenerator,
+ RustGenerator,
+ ):
continue
options = GeneratorOptions(output_dir=Path("/tmp"))
generator = generator_cls(schema, options)
@@ -368,6 +375,13 @@ def
test_proto_grpc_services_use_imported_qualified_type_references(tmp_path: Pa
)
assert "marshaller(com.example.common.CommonModels.Shared.class)" in java
+ rust_files = generate_service_files(schema, RustGenerator)
+ rust_service = rust_files["api_service.rs"]
+ assert "request: ::tonic::Request<crate::common::Shared>," in rust_service
+ assert "::tonic::Response<crate::api::Local>" in rust_service
+ assert "crate::common::common::Shared" not in rust_service
+ assert "crate::api::api::Local" not in rust_service
+
def test_proto_grpc_absolute_rpc_type_uses_package_type_not_nested_shadow():
schema = parse_proto(
@@ -396,6 +410,13 @@ def
test_proto_grpc_absolute_rpc_type_uses_package_type_not_nested_shadow():
assert "io.grpc.MethodDescriptor<Request, Response>" in java
assert "io.grpc.MethodDescriptor<demo.Request, Response>" not in java
+ rust_files = generate_service_files(schema, RustGenerator)
+ rust_service = rust_files["demo_service.rs"]
+ assert "request: ::tonic::Request<crate::demo::Request>," in rust_service
+ assert "::tonic::Response<crate::demo::Response>" in rust_service
+ assert "crate::demo::demo::Request" not in rust_service
+ assert "crate::demo::::demo::Request" not in rust_service
+
def test_proto_grpc_absolute_rpc_type_prefers_longest_package_prefix(tmp_path:
Path):
common = tmp_path / "common.proto"
@@ -441,6 +462,12 @@ def
test_proto_grpc_absolute_rpc_type_prefers_longest_package_prefix(tmp_path: P
assert "marshaller(pkg.two.C.class)" in java
assert "io.grpc.MethodDescriptor<beta.C, beta.C>" not in java
+ rust_files = generate_service_files(schema, RustGenerator)
+ rust_service = rust_files["alpha_service.rs"]
+ assert "request: ::tonic::Request<crate::alpha_beta::C>," in rust_service
+ assert "::tonic::Response<crate::alpha_beta::C>" in rust_service
+ assert "crate::alpha::beta::C" not in rust_service
+
def test_java_grpc_service_class_collision_fails():
schema = parse_fdl(
@@ -593,7 +620,7 @@ def test_grpc_method_name_collisions_fail():
schema, GeneratorOptions(output_dir=Path("/tmp"), grpc=True)
)
try:
- rust_generator.generate()
+ rust_generator.generate_services()
except ValueError as e:
assert "Rust name collision" in str(e)
else:
@@ -667,7 +694,7 @@ def test_rust_grpc_service_module_collisions_fail():
schema, GeneratorOptions(output_dir=Path("/tmp"), grpc=True)
)
try:
- generator.generate()
+ generator.generate_services()
except ValueError as e:
assert "Rust name collision" in str(e)
else:
@@ -763,6 +790,8 @@ def test_compile_service_schema_with_grpc_flag(tmp_path:
Path):
assert len(files) >= 1, f"{lang}: expected at least one file with
grpc=True"
assert (lang_dirs["java"] / "demo" / "greeter" /
"GreeterGrpc.java").exists()
assert (lang_dirs["python"] / "demo_greeter_grpc.py").exists()
+ assert (lang_dirs["rust"] / "demo_greeter_service.rs").exists()
+ assert (lang_dirs["rust"] / "demo_greeter_service_grpc.rs").exists()
def test_generated_message_contains_key_signatures():
@@ -778,3 +807,70 @@ def test_generated_message_contains_key_signatures():
all_python = "\n".join(python_files.values())
assert "HelloRequest" in all_python
assert "HelloReply" in all_python
+
+
+def test_rust_grpc_rejects_non_thread_safe_refs():
+ cases = [
+ (
+ "Rust gRPC payload type Request.node uses non-thread-safe ref",
+ """
+ package demo.grpc;
+
+ message Node {}
+
+ message Request {
+ ref(thread_safe=false) Node node = 1;
+ }
+
+ message Response {}
+
+ service Api {
+ rpc Call (Request) returns (Response);
+ }
+ """,
+ ),
+ (
+ "Rust gRPC payload type Request.groups uses non-thread-safe list
element ref",
+ """
+ package demo.grpc;
+
+ message Node {}
+
+ message Request {
+ list<ref(thread_safe=false) Node> groups = 1;
+ }
+
+ message Response {}
+
+ service Api {
+ rpc Call (Request) returns (Response);
+ }
+ """,
+ ),
+ (
+ "Rust gRPC payload type Request.nodes uses non-thread-safe map
value ref",
+ """
+ package demo.grpc;
+
+ message Node {}
+
+ message Request {
+ map<string, ref(thread_safe=false) Node> nodes = 1;
+ }
+
+ message Response {}
+
+ service Api {
+ rpc Call (Request) returns (Response);
+ }
+ """,
+ ),
+ ]
+
+ for message, source in cases:
+ schema = parse_fdl(dedent(source))
+ generator = RustGenerator(
+ schema, GeneratorOptions(output_dir=Path("/tmp"), grpc=True)
+ )
+ with pytest.raises(ValueError, match=message):
+ generator.generate_services()
diff --git a/integration_tests/grpc_tests/generate_grpc.py
b/integration_tests/grpc_tests/generate_grpc.py
index 5d3131640..af3f3f44f 100644
--- a/integration_tests/grpc_tests/generate_grpc.py
+++ b/integration_tests/grpc_tests/generate_grpc.py
@@ -34,6 +34,7 @@ OUTPUTS = {
"java": TEST_DIR / "java/src/main/java/generated",
"python": TEST_DIR / "python/grpc_tests/generated",
"go": TEST_DIR / "go/generated",
+ "rust": TEST_DIR / "rust/generated/src",
}
@@ -74,6 +75,7 @@ def main() -> int:
f"--java_out={OUTPUTS['java']}",
f"--python_out={OUTPUTS['python']}",
f"--go_out={go_pkg_out}",
+ f"--rust_out={OUTPUTS['rust']}",
"--grpc",
],
env=env,
diff --git
a/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/GrpcInteropTest.java
b/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/GrpcInteropTest.java
index c494c6dc2..bbf6a8629 100644
---
a/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/GrpcInteropTest.java
+++
b/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/GrpcInteropTest.java
@@ -71,7 +71,57 @@ public class GrpcInteropTest {
new PeerOutputCollector(process.getInputStream(),
"python-grpc-server");
outputCollector.start();
try {
- int port = waitForPort(process, outputCollector, portFile);
+ int port = waitForPort(process, outputCollector, portFile, "Python");
+ ManagedChannel channel =
+ ManagedChannelBuilder.forAddress("127.0.0.1",
port).usePlaintext().build();
+ try {
+ exerciseFdl(channel);
+ exerciseFbs(channel);
+ exercisePb(channel);
+ } finally {
+ channel.shutdownNow();
+ channel.awaitTermination(10, TimeUnit.SECONDS);
+ }
+ } finally {
+ process.destroy();
+ process.waitFor(10, TimeUnit.SECONDS);
+ if (process.isAlive()) {
+ process.destroyForcibly();
+ process.waitFor(10, TimeUnit.SECONDS);
+ }
+ outputCollector.awaitOutput();
+ Files.deleteIfExists(portFile);
+ }
+ }
+
+ @Test
+ public void testJavaServerRustClient() throws Exception {
+ Server server =
+ ServerBuilder.forPort(0)
+ .addService(new FdlService())
+ .addService(new FbsService())
+ .addService(new PbService())
+ .build()
+ .start();
+ try {
+ runRust("rust-grpc-client", "client", "--target", "127.0.0.1:" +
server.getPort());
+ } finally {
+ server.shutdownNow();
+ server.awaitTermination(10, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testJavaClientRustServer() throws Exception {
+ Path portFile = Files.createTempFile("fory-grpc-rust-", ".port");
+ Files.deleteIfExists(portFile);
+ PeerCommand command = rustCommand("server", "--port-file",
portFile.toString());
+ Process process = startPeer(command);
+ PeerOutputCollector outputCollector =
+ new PeerOutputCollector(process.getInputStream(), "rust-grpc-server");
+ outputCollector.start();
+ try {
+ int port = waitForPort(process, outputCollector, portFile, "Rust");
ManagedChannel channel =
ManagedChannelBuilder.forAddress("127.0.0.1",
port).usePlaintext().build();
try {
@@ -280,11 +330,7 @@ public class GrpcInteropTest {
@Test
public void testJavaServerGoClient() throws Exception {
- Server server =
- ServerBuilder.forPort(0)
- .addService(new FdlService())
- .build()
- .start();
+ Server server = ServerBuilder.forPort(0).addService(new
FdlService()).build().start();
try {
runGo("go-grpc-client", "client", "--target", "127.0.0.1:" +
server.getPort());
} finally {
@@ -303,7 +349,7 @@ public class GrpcInteropTest {
new PeerOutputCollector(process.getInputStream(), "go-grpc-server");
outputCollector.start();
try {
- int port = waitForPort(process, outputCollector, portFile);
+ int port = waitForPort(process, outputCollector, portFile, "Go");
ManagedChannel channel =
ManagedChannelBuilder.forAddress("127.0.0.1",
port).usePlaintext().build();
try {
@@ -397,6 +443,34 @@ public class GrpcInteropTest {
return peerCommand;
}
+ private PeerCommand rustCommand(String... args) {
+ Path repoRoot = repoRoot();
+ Path grpcRoot =
repoRoot.resolve("integration_tests").resolve("grpc_tests");
+ Path rustRoot = grpcRoot.resolve("rust");
+ List<String> command = new ArrayList<>();
+ command.add("cargo");
+ command.add("run");
+ command.add("--quiet");
+ command.add("--manifest-path");
+ command.add(rustRoot.resolve("interop").resolve("Cargo.toml").toString());
+ command.add("--");
+ command.addAll(Arrays.asList(args));
+ PeerCommand peerCommand = new PeerCommand();
+ peerCommand.command = command;
+ peerCommand.workDir = rustRoot;
+ peerCommand.environment.put("CARGO_TERM_COLOR", "never");
+ peerCommand.environment.put("ENABLE_FORY_DEBUG_OUTPUT", "1");
+ peerCommand.environment.put("RUST_BACKTRACE", "1");
+ peerCommand.environment.put("NO_PROXY", "127.0.0.1,localhost");
+ peerCommand.environment.put("no_proxy", "127.0.0.1,localhost");
+ for (String proxyVar :
+ Arrays.asList(
+ "all_proxy", "http_proxy", "https_proxy", "ALL_PROXY",
"HTTP_PROXY", "HTTPS_PROXY")) {
+ peerCommand.environment.put(proxyVar, "");
+ }
+ return peerCommand;
+ }
+
private void runPython(String peer, String... args) throws IOException,
InterruptedException {
Process process = startPeer(pythonCommand(args));
PeerOutputCollector outputCollector = new
PeerOutputCollector(process.getInputStream(), peer);
@@ -419,6 +493,28 @@ public class GrpcInteropTest {
outputCollector.awaitOutput();
}
+ private void runRust(String peer, String... args) throws IOException,
InterruptedException {
+ Process process = startPeer(rustCommand(args));
+ PeerOutputCollector outputCollector = new
PeerOutputCollector(process.getInputStream(), peer);
+ outputCollector.start();
+ boolean finished = process.waitFor(180, TimeUnit.SECONDS);
+ if (!finished) {
+ process.destroyForcibly();
+ process.waitFor(10, TimeUnit.SECONDS);
+ Assert.fail("Peer process timed out for " + peer +
peerOutput(outputCollector));
+ }
+ int exitCode = process.exitValue();
+ if (exitCode != 0) {
+ Assert.fail(
+ "Peer process failed for "
+ + peer
+ + " with exit code "
+ + exitCode
+ + peerOutput(outputCollector));
+ }
+ outputCollector.awaitOutput();
+ }
+
private Process startPeer(PeerCommand command) throws IOException {
ProcessBuilder builder = new ProcessBuilder(command.command);
builder.redirectErrorStream(true);
@@ -427,12 +523,13 @@ public class GrpcInteropTest {
return builder.start();
}
- private int waitForPort(Process process, PeerOutputCollector
outputCollector, Path portFile)
+ private int waitForPort(
+ Process process, PeerOutputCollector outputCollector, Path portFile,
String language)
throws IOException, InterruptedException {
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
while (System.nanoTime() < deadline) {
if (!process.isAlive()) {
- Assert.fail("Python gRPC server exited early" +
peerOutput(outputCollector));
+ Assert.fail(language + " gRPC server exited early" +
peerOutput(outputCollector));
}
if (Files.exists(portFile)) {
String value = new String(Files.readAllBytes(portFile),
StandardCharsets.UTF_8).trim();
@@ -444,7 +541,8 @@ public class GrpcInteropTest {
}
process.destroyForcibly();
process.waitFor(10, TimeUnit.SECONDS);
- Assert.fail("Timed out waiting for Python gRPC server port" +
peerOutput(outputCollector));
+ Assert.fail(
+ "Timed out waiting for " + language + " gRPC server port" +
peerOutput(outputCollector));
return -1;
}
diff --git a/integration_tests/grpc_tests/run_tests.sh
b/integration_tests/grpc_tests/run_tests.sh
index 033d27c56..22b9cd3aa 100755
--- a/integration_tests/grpc_tests/run_tests.sh
+++ b/integration_tests/grpc_tests/run_tests.sh
@@ -29,6 +29,7 @@ python "${SCRIPT_DIR}/generate_grpc.py"
cd "${SCRIPT_DIR}/go"
go build -o grpc-interop .
+cargo build --manifest-path "${SCRIPT_DIR}/rust/Cargo.toml" --workspace --quiet
cd "${ROOT_DIR}/integration_tests/grpc_tests/java"
mvn -T16 --no-transfer-progress \
-Dtest=GrpcInteropTest \
diff --git a/integration_tests/grpc_tests/run_tests.sh
b/integration_tests/grpc_tests/rust/Cargo.toml
old mode 100755
new mode 100644
similarity index 61%
copy from integration_tests/grpc_tests/run_tests.sh
copy to integration_tests/grpc_tests/rust/Cargo.toml
index 033d27c56..de5fe6244
--- a/integration_tests/grpc_tests/run_tests.sh
+++ b/integration_tests/grpc_tests/rust/Cargo.toml
@@ -1,5 +1,3 @@
-#!/bin/bash
-
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
@@ -17,19 +15,22 @@
# specific language governing permissions and limitations
# under the License.
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-ROOT_DIR="$(cd "${SCRIPT_DIR}/../.." && pwd)"
-
-python -m pip install "grpcio>=1.62.2,<1.71"
-python -m pip install -v -e "${ROOT_DIR}/python"
+[workspace]
+members = [
+ "generated",
+ "interop",
+]
+resolver = "2"
-python "${SCRIPT_DIR}/generate_grpc.py"
+[workspace.package]
+version = "1.2.0-alpha.0"
+edition = "2021"
+license = "Apache-2.0"
-cd "${SCRIPT_DIR}/go"
-go build -o grpc-interop .
-cd "${ROOT_DIR}/integration_tests/grpc_tests/java"
-mvn -T16 --no-transfer-progress \
- -Dtest=GrpcInteropTest \
- test
+[workspace.dependencies]
+bytes = "1"
+fory = { path = "../../../rust/fory" }
+fory-core = { path = "../../../rust/fory-core" }
+tokio = { version = "1", features = ["fs", "macros", "net", "rt-multi-thread",
"sync"] }
+tokio-stream = { version = "0.1", features = ["net"] }
+tonic = { version = "0.14", features = ["transport"] }
diff --git a/integration_tests/grpc_tests/run_tests.sh
b/integration_tests/grpc_tests/rust/generated/Cargo.toml
old mode 100755
new mode 100644
similarity index 64%
copy from integration_tests/grpc_tests/run_tests.sh
copy to integration_tests/grpc_tests/rust/generated/Cargo.toml
index 033d27c56..ea65f83f4
--- a/integration_tests/grpc_tests/run_tests.sh
+++ b/integration_tests/grpc_tests/rust/generated/Cargo.toml
@@ -1,5 +1,3 @@
-#!/bin/bash
-
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
@@ -17,19 +15,14 @@
# specific language governing permissions and limitations
# under the License.
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-ROOT_DIR="$(cd "${SCRIPT_DIR}/../.." && pwd)"
-
-python -m pip install "grpcio>=1.62.2,<1.71"
-python -m pip install -v -e "${ROOT_DIR}/python"
-
-python "${SCRIPT_DIR}/generate_grpc.py"
+[package]
+name = "grpc_tests_generated"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
-cd "${SCRIPT_DIR}/go"
-go build -o grpc-interop .
-cd "${ROOT_DIR}/integration_tests/grpc_tests/java"
-mvn -T16 --no-transfer-progress \
- -Dtest=GrpcInteropTest \
- test
+[dependencies]
+bytes.workspace = true
+fory.workspace = true
+fory-core.workspace = true
+tonic.workspace = true
diff --git a/integration_tests/grpc_tests/rust/generated/src/lib.rs
b/integration_tests/grpc_tests/rust/generated/src/lib.rs
new file mode 100644
index 000000000..82ba47079
--- /dev/null
+++ b/integration_tests/grpc_tests/rust/generated/src/lib.rs
@@ -0,0 +1,26 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+pub mod grpc_fbs;
+pub mod grpc_fbs_service;
+pub mod grpc_fbs_service_grpc;
+pub mod grpc_fdl;
+pub mod grpc_fdl_service;
+pub mod grpc_fdl_service_grpc;
+pub mod grpc_pb;
+pub mod grpc_pb_service;
+pub mod grpc_pb_service_grpc;
diff --git a/integration_tests/grpc_tests/run_tests.sh
b/integration_tests/grpc_tests/rust/interop/Cargo.toml
old mode 100755
new mode 100644
similarity index 64%
copy from integration_tests/grpc_tests/run_tests.sh
copy to integration_tests/grpc_tests/rust/interop/Cargo.toml
index 033d27c56..3c06b6349
--- a/integration_tests/grpc_tests/run_tests.sh
+++ b/integration_tests/grpc_tests/rust/interop/Cargo.toml
@@ -1,5 +1,3 @@
-#!/bin/bash
-
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
@@ -17,19 +15,14 @@
# specific language governing permissions and limitations
# under the License.
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-ROOT_DIR="$(cd "${SCRIPT_DIR}/../.." && pwd)"
-
-python -m pip install "grpcio>=1.62.2,<1.71"
-python -m pip install -v -e "${ROOT_DIR}/python"
-
-python "${SCRIPT_DIR}/generate_grpc.py"
+[package]
+name = "grpc_tests_interop"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
-cd "${SCRIPT_DIR}/go"
-go build -o grpc-interop .
-cd "${ROOT_DIR}/integration_tests/grpc_tests/java"
-mvn -T16 --no-transfer-progress \
- -Dtest=GrpcInteropTest \
- test
+[dependencies]
+grpc_tests_generated = { path = "../generated" }
+tokio.workspace = true
+tokio-stream.workspace = true
+tonic.workspace = true
diff --git a/integration_tests/grpc_tests/rust/interop/src/main.rs
b/integration_tests/grpc_tests/rust/interop/src/main.rs
new file mode 100644
index 000000000..30a85ad73
--- /dev/null
+++ b/integration_tests/grpc_tests/rust/interop/src/main.rs
@@ -0,0 +1,909 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+use std::env;
+use std::io::{Error as IoError, ErrorKind};
+use std::path::PathBuf;
+use std::pin::Pin;
+
+use grpc_tests_generated::grpc_fbs::{GrpcFbsRequest, GrpcFbsResponse,
GrpcFbsUnion};
+use grpc_tests_generated::grpc_fbs_service::FbsGrpcService;
+use
grpc_tests_generated::grpc_fbs_service_grpc::fbs_grpc_service_client::FbsGrpcServiceClient;
+use
grpc_tests_generated::grpc_fbs_service_grpc::fbs_grpc_service_server::FbsGrpcServiceServer;
+use grpc_tests_generated::grpc_fdl::{GrpcFdlRequest, GrpcFdlResponse,
GrpcFdlUnion};
+use grpc_tests_generated::grpc_fdl_service::FdlGrpcService;
+use
grpc_tests_generated::grpc_fdl_service_grpc::fdl_grpc_service_client::FdlGrpcServiceClient;
+use
grpc_tests_generated::grpc_fdl_service_grpc::fdl_grpc_service_server::FdlGrpcServiceServer;
+use grpc_tests_generated::grpc_pb::grpc_pb_request;
+use grpc_tests_generated::grpc_pb::grpc_pb_response;
+use grpc_tests_generated::grpc_pb::{GrpcPbRequest, GrpcPbResponse};
+use grpc_tests_generated::grpc_pb_service::PbGrpcService;
+use
grpc_tests_generated::grpc_pb_service_grpc::pb_grpc_service_client::PbGrpcServiceClient;
+use
grpc_tests_generated::grpc_pb_service_grpc::pb_grpc_service_server::PbGrpcServiceServer;
+use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream};
+use tokio_stream::Stream;
+use tonic::transport::{Channel, Server};
+use tonic::{Request, Response, Status};
+
+type AnyError = Box<dyn std::error::Error + Send + Sync>;
+type AnyResult<T> = std::result::Result<T, AnyError>;
+type StatusResult<T> = std::result::Result<T, Status>;
+type GrpcStream<T> = Pin<Box<dyn Stream<Item = StatusResult<T>> + Send +
'static>>;
+
+/// Runs a gRPC client or server according to the CLI arguments.
+#[tokio::main]
+async fn main() -> AnyResult<()> {
+ let args = env::args().skip(1).collect::<Vec<_>>();
+ match args.first().map(String::as_str) {
+ Some("client") => run_client(required_arg(&args[1..],
"--target")?).await,
+ Some("server") => run_server(PathBuf::from(required_arg(&args[1..],
"--port-file")?)).await,
+ _ => Err(invalid_args(
+ "usage: grpc_tests_interop client --target HOST:PORT | server
--port-file PATH",
+ )),
+ }
+}
+
+/// Returns the value following a required CLI flag.
+fn required_arg(args: &[String], name: &str) -> AnyResult<String> {
+ let Some(index) = args.iter().position(|arg| arg == name) else {
+ return Err(invalid_args(format!("missing required argument {name}")));
+ };
+ args.get(index + 1)
+ .cloned()
+ .ok_or_else(|| invalid_args(format!("missing value for {name}")))
+}
+
+fn invalid_args(message: impl Into<String>) -> AnyError {
+ Box::new(IoError::new(ErrorKind::InvalidInput, message.into()))
+}
+
+// Connects to a gRPC server and exercises every schema service as a client.
+async fn run_client(target: String) -> AnyResult<()> {
+ let target = if target.contains("://") {
+ target
+ } else {
+ format!("http://{target}";)
+ };
+ let channel = Channel::from_shared(target)?.connect().await?;
+
+ exercise_fdl(channel.clone()).await?;
+ exercise_fbs(channel.clone()).await?;
+ exercise_pb(channel).await?;
+ Ok(())
+}
+
+// Starts a gRPC server registers services.
+async fn run_server(port_file: PathBuf) -> AnyResult<()> {
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
+ let port = listener.local_addr()?.port();
+ tokio::fs::write(&port_file, port.to_string()).await?;
+ Server::builder()
+ .add_service(FdlGrpcServiceServer::new(FdlServiceImpl))
+ .add_service(FbsGrpcServiceServer::new(FbsServiceImpl))
+ .add_service(PbGrpcServiceServer::new(PbServiceImpl))
+ .serve_with_incoming(TcpListenerStream::new(listener))
+ .await?;
+ Ok(())
+}
+
+// Drains a streaming RPC response into a vector for equality assertions.
+async fn collect_stream<T>(mut stream: tonic::codec::Streaming<T>) ->
StatusResult<Vec<T>> {
+ let mut values = Vec::new();
+ while let Some(value) = stream.message().await? {
+ values.push(value);
+ }
+ Ok(values)
+}
+
+// Builds a FDL request.
+fn fdl_request(id: &str, count: i32, payload: &str) -> GrpcFdlRequest {
+ GrpcFdlRequest {
+ id: id.to_string(),
+ count,
+ payload: payload.to_string(),
+ }
+}
+
+// Builds the expected FDL response for a FDL request.
+fn fdl_response(request: &GrpcFdlRequest, tag: &str, offset: i32) ->
GrpcFdlResponse {
+ GrpcFdlResponse {
+ id: format!("{tag}:{}", request.id),
+ count: request.count + offset,
+ payload: format!("{tag}:{}", request.payload),
+ }
+}
+
+// Builds the aggregate FDL response used by client-streaming RPCs.
+fn fdl_aggregate(requests: &[GrpcFdlRequest]) -> GrpcFdlResponse {
+ GrpcFdlResponse {
+ id: format!(
+ "client:{}",
+ requests
+ .iter()
+ .map(|request| request.id.as_str())
+ .collect::<Vec<_>>()
+ .join("+")
+ ),
+ count: requests.iter().map(|request| request.count).sum(),
+ payload: format!(
+ "client:{}",
+ requests
+ .iter()
+ .map(|request| request.payload.as_str())
+ .collect::<Vec<_>>()
+ .join("+")
+ ),
+ }
+}
+
+// Extracts the request arm from a FDL union.
+fn fdl_request_from_union(union: GrpcFdlUnion) -> StatusResult<GrpcFdlRequest>
{
+ match union {
+ GrpcFdlUnion::Request(request) => Ok(request),
+ _ => Err(Status::invalid_argument("expected GrpcFdlUnion::Request")),
+ }
+}
+
+async fn exercise_fdl(channel: Channel) -> AnyResult<()> {
+ let mut client = FdlGrpcServiceClient::new(channel);
+ let requests = vec![
+ fdl_request("fdl-a", 1, "alpha"),
+ fdl_request("fdl-b", 2, "beta"),
+ ];
+ let first = requests[0].clone();
+ assert_eq!(
+ client.unary_message(first.clone()).await?.into_inner(),
+ fdl_response(&first, "unary", 10)
+ );
+ assert_eq!(
+ collect_stream(
+ client
+ .server_stream_message(first.clone())
+ .await?
+ .into_inner()
+ )
+ .await?,
+ (0..3)
+ .map(|index| fdl_response(&first, &format!("server-{index}"),
index))
+ .collect::<Vec<_>>()
+ );
+ assert_eq!(
+ client
+ .client_stream_message(tokio_stream::iter(requests.clone()))
+ .await?
+ .into_inner(),
+ fdl_aggregate(&requests)
+ );
+ assert_eq!(
+ collect_stream(
+ client
+ .bidi_stream_message(tokio_stream::iter(requests.clone()))
+ .await?
+ .into_inner()
+ )
+ .await?,
+ requests
+ .iter()
+ .enumerate()
+ .map(|(index, request)| fdl_response(request,
&format!("bidi-{index}"), index as i32))
+ .collect::<Vec<_>>()
+ );
+
+ let union_requests = vec![
+ fdl_request("fdl-u-a", 3, "union-alpha"),
+ fdl_request("fdl-u-b", 4, "union-beta"),
+ ];
+ let union_payloads = union_requests
+ .iter()
+ .cloned()
+ .map(GrpcFdlUnion::Request)
+ .collect::<Vec<_>>();
+ let first_union = union_payloads[0].clone();
+ assert_eq!(
+ client.unary_union(first_union.clone()).await?.into_inner(),
+ GrpcFdlUnion::Response(fdl_response(&union_requests[0], "unary", 10))
+ );
+ assert_eq!(
+
collect_stream(client.server_stream_union(first_union).await?.into_inner()).await?,
+ (0..3)
+ .map(|index| {
+ GrpcFdlUnion::Response(fdl_response(
+ &union_requests[0],
+ &format!("server-{index}"),
+ index,
+ ))
+ })
+ .collect::<Vec<_>>()
+ );
+ assert_eq!(
+ client
+ .client_stream_union(tokio_stream::iter(union_payloads.clone()))
+ .await?
+ .into_inner(),
+ GrpcFdlUnion::Response(fdl_aggregate(&union_requests))
+ );
+ assert_eq!(
+ collect_stream(
+ client
+ .bidi_stream_union(tokio_stream::iter(union_payloads))
+ .await?
+ .into_inner()
+ )
+ .await?,
+ union_requests
+ .iter()
+ .enumerate()
+ .map(|(index, request)| {
+ GrpcFdlUnion::Response(fdl_response(
+ request,
+ &format!("bidi-{index}"),
+ index as i32,
+ ))
+ })
+ .collect::<Vec<_>>()
+ );
+ Ok(())
+}
+
+struct FdlServiceImpl;
+
+#[tonic::async_trait]
+impl FdlGrpcService for FdlServiceImpl {
+ type ServerStreamMessageStream = GrpcStream<GrpcFdlResponse>;
+ type BidiStreamMessageStream = GrpcStream<GrpcFdlResponse>;
+ type ServerStreamUnionStream = GrpcStream<GrpcFdlUnion>;
+ type BidiStreamUnionStream = GrpcStream<GrpcFdlUnion>;
+
+ async fn unary_message(
+ &self,
+ request: Request<GrpcFdlRequest>,
+ ) -> StatusResult<Response<GrpcFdlResponse>> {
+ Ok(Response::new(fdl_response(
+ &request.into_inner(),
+ "unary",
+ 10,
+ )))
+ }
+
+ async fn server_stream_message(
+ &self,
+ request: Request<GrpcFdlRequest>,
+ ) -> StatusResult<Response<Self::ServerStreamMessageStream>> {
+ let request = request.into_inner();
+ Ok(Response::new(Box::pin(tokio_stream::iter(
+ (0..3)
+ .map(|index| Ok(fdl_response(&request,
&format!("server-{index}"), index)))
+ .collect::<Vec<_>>(),
+ ))))
+ }
+
+ async fn client_stream_message(
+ &self,
+ request: Request<tonic::Streaming<GrpcFdlRequest>>,
+ ) -> StatusResult<Response<GrpcFdlResponse>> {
+ let mut stream = request.into_inner();
+ let mut requests = Vec::new();
+ while let Some(request) = stream.message().await? {
+ requests.push(request);
+ }
+ Ok(Response::new(fdl_aggregate(&requests)))
+ }
+
+ async fn bidi_stream_message(
+ &self,
+ request: Request<tonic::Streaming<GrpcFdlRequest>>,
+ ) -> StatusResult<Response<Self::BidiStreamMessageStream>> {
+ let mut stream = request.into_inner();
+ let (tx, rx) = tokio::sync::mpsc::channel(4);
+ tokio::spawn(async move {
+ let mut index = 0;
+ loop {
+ match stream.message().await {
+ Ok(Some(request)) => {
+ let response = fdl_response(&request,
&format!("bidi-{index}"), index);
+ if tx.send(Ok(response)).await.is_err() {
+ break;
+ }
+ index += 1;
+ }
+ Ok(None) => break,
+ Err(error) => {
+ let _ = tx.send(Err(error)).await;
+ break;
+ }
+ }
+ }
+ });
+ Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
+ }
+
+ async fn unary_union(
+ &self,
+ request: Request<GrpcFdlUnion>,
+ ) -> StatusResult<Response<GrpcFdlUnion>> {
+ let request = fdl_request_from_union(request.into_inner())?;
+ Ok(Response::new(GrpcFdlUnion::Response(fdl_response(
+ &request, "unary", 10,
+ ))))
+ }
+
+ async fn server_stream_union(
+ &self,
+ request: Request<GrpcFdlUnion>,
+ ) -> StatusResult<Response<Self::ServerStreamUnionStream>> {
+ let request = fdl_request_from_union(request.into_inner())?;
+ Ok(Response::new(Box::pin(tokio_stream::iter(
+ (0..3)
+ .map(|index| {
+ Ok(GrpcFdlUnion::Response(fdl_response(
+ &request,
+ &format!("server-{index}"),
+ index,
+ )))
+ })
+ .collect::<Vec<_>>(),
+ ))))
+ }
+
+ async fn client_stream_union(
+ &self,
+ request: Request<tonic::Streaming<GrpcFdlUnion>>,
+ ) -> StatusResult<Response<GrpcFdlUnion>> {
+ let mut stream = request.into_inner();
+ let mut requests = Vec::new();
+ while let Some(union) = stream.message().await? {
+ requests.push(fdl_request_from_union(union)?);
+ }
+ Ok(Response::new(GrpcFdlUnion::Response(fdl_aggregate(
+ &requests,
+ ))))
+ }
+
+ async fn bidi_stream_union(
+ &self,
+ request: Request<tonic::Streaming<GrpcFdlUnion>>,
+ ) -> StatusResult<Response<Self::BidiStreamUnionStream>> {
+ let mut stream = request.into_inner();
+ let (tx, rx) = tokio::sync::mpsc::channel(4);
+ tokio::spawn(async move {
+ let mut index = 0;
+ loop {
+ match stream.message().await {
+ Ok(Some(union)) => match fdl_request_from_union(union) {
+ Ok(request) => {
+ let response = GrpcFdlUnion::Response(fdl_response(
+ &request,
+ &format!("bidi-{index}"),
+ index,
+ ));
+ if tx.send(Ok(response)).await.is_err() {
+ break;
+ }
+ index += 1;
+ }
+ Err(error) => {
+ let _ = tx.send(Err(error)).await;
+ break;
+ }
+ },
+ Ok(None) => break,
+ Err(error) => {
+ let _ = tx.send(Err(error)).await;
+ break;
+ }
+ }
+ }
+ });
+ Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
+ }
+}
+
+// Builds a FBS request.
+fn fbs_request(id: &str, count: i32, payload: &str) -> GrpcFbsRequest {
+ GrpcFbsRequest {
+ id: id.to_string(),
+ count,
+ payload: payload.to_string(),
+ }
+}
+
+// Builds the expected FBS response for a request.
+fn fbs_response(request: &GrpcFbsRequest, tag: &str, offset: i32) ->
GrpcFbsResponse {
+ GrpcFbsResponse {
+ id: format!("{tag}:{}", request.id),
+ count: request.count + offset,
+ payload: format!("{tag}:{}", request.payload),
+ }
+}
+
+// Builds the aggregate FBS response used by client-streaming RPCs.
+fn fbs_aggregate(requests: &[GrpcFbsRequest]) -> GrpcFbsResponse {
+ GrpcFbsResponse {
+ id: format!(
+ "client:{}",
+ requests
+ .iter()
+ .map(|request| request.id.as_str())
+ .collect::<Vec<_>>()
+ .join("+")
+ ),
+ count: requests.iter().map(|request| request.count).sum(),
+ payload: format!(
+ "client:{}",
+ requests
+ .iter()
+ .map(|request| request.payload.as_str())
+ .collect::<Vec<_>>()
+ .join("+")
+ ),
+ }
+}
+
+// Extracts the request arm from a FBS union.
+fn fbs_request_from_union(union: GrpcFbsUnion) -> StatusResult<GrpcFbsRequest>
{
+ match union {
+ GrpcFbsUnion::GrpcFbsRequest(request) => Ok(request),
+ _ => Err(Status::invalid_argument(
+ "expected GrpcFbsUnion::GrpcFbsRequest",
+ )),
+ }
+}
+
+async fn exercise_fbs(channel: Channel) -> AnyResult<()> {
+ let mut client = FbsGrpcServiceClient::new(channel);
+ let requests = vec![
+ fbs_request("fbs-a", 5, "alpha"),
+ fbs_request("fbs-b", 6, "beta"),
+ ];
+ let first = requests[0].clone();
+ assert_eq!(
+ client.unary_message(first.clone()).await?.into_inner(),
+ fbs_response(&first, "unary", 10)
+ );
+ assert_eq!(
+ collect_stream(
+ client
+ .server_stream_message(first.clone())
+ .await?
+ .into_inner()
+ )
+ .await?,
+ (0..3)
+ .map(|index| fbs_response(&first, &format!("server-{index}"),
index))
+ .collect::<Vec<_>>()
+ );
+ assert_eq!(
+ client
+ .client_stream_message(tokio_stream::iter(requests.clone()))
+ .await?
+ .into_inner(),
+ fbs_aggregate(&requests)
+ );
+ assert_eq!(
+ collect_stream(
+ client
+ .bidi_stream_message(tokio_stream::iter(requests.clone()))
+ .await?
+ .into_inner()
+ )
+ .await?,
+ requests
+ .iter()
+ .enumerate()
+ .map(|(index, request)| fbs_response(request,
&format!("bidi-{index}"), index as i32))
+ .collect::<Vec<_>>()
+ );
+
+ let union_requests = vec![
+ fbs_request("fbs-u-a", 7, "union-alpha"),
+ fbs_request("fbs-u-b", 8, "union-beta"),
+ ];
+ let union_payloads = union_requests
+ .iter()
+ .cloned()
+ .map(GrpcFbsUnion::GrpcFbsRequest)
+ .collect::<Vec<_>>();
+ let first_union = union_payloads[0].clone();
+ assert_eq!(
+ client.unary_union(first_union.clone()).await?.into_inner(),
+ GrpcFbsUnion::GrpcFbsResponse(fbs_response(&union_requests[0],
"unary", 10))
+ );
+ assert_eq!(
+
collect_stream(client.server_stream_union(first_union).await?.into_inner()).await?,
+ (0..3)
+ .map(|index| {
+ GrpcFbsUnion::GrpcFbsResponse(fbs_response(
+ &union_requests[0],
+ &format!("server-{index}"),
+ index,
+ ))
+ })
+ .collect::<Vec<_>>()
+ );
+ assert_eq!(
+ client
+ .client_stream_union(tokio_stream::iter(union_payloads.clone()))
+ .await?
+ .into_inner(),
+ GrpcFbsUnion::GrpcFbsResponse(fbs_aggregate(&union_requests))
+ );
+ assert_eq!(
+ collect_stream(
+ client
+ .bidi_stream_union(tokio_stream::iter(union_payloads))
+ .await?
+ .into_inner()
+ )
+ .await?,
+ union_requests
+ .iter()
+ .enumerate()
+ .map(|(index, request)| {
+ GrpcFbsUnion::GrpcFbsResponse(fbs_response(
+ request,
+ &format!("bidi-{index}"),
+ index as i32,
+ ))
+ })
+ .collect::<Vec<_>>()
+ );
+ Ok(())
+}
+
+struct FbsServiceImpl;
+
+#[tonic::async_trait]
+impl FbsGrpcService for FbsServiceImpl {
+ type ServerStreamMessageStream = GrpcStream<GrpcFbsResponse>;
+ type BidiStreamMessageStream = GrpcStream<GrpcFbsResponse>;
+ type ServerStreamUnionStream = GrpcStream<GrpcFbsUnion>;
+ type BidiStreamUnionStream = GrpcStream<GrpcFbsUnion>;
+
+ async fn unary_message(
+ &self,
+ request: Request<GrpcFbsRequest>,
+ ) -> StatusResult<Response<GrpcFbsResponse>> {
+ Ok(Response::new(fbs_response(
+ &request.into_inner(),
+ "unary",
+ 10,
+ )))
+ }
+
+ async fn server_stream_message(
+ &self,
+ request: Request<GrpcFbsRequest>,
+ ) -> StatusResult<Response<Self::ServerStreamMessageStream>> {
+ let request = request.into_inner();
+ Ok(Response::new(Box::pin(tokio_stream::iter(
+ (0..3)
+ .map(|index| Ok(fbs_response(&request,
&format!("server-{index}"), index)))
+ .collect::<Vec<_>>(),
+ ))))
+ }
+
+ async fn client_stream_message(
+ &self,
+ request: Request<tonic::Streaming<GrpcFbsRequest>>,
+ ) -> StatusResult<Response<GrpcFbsResponse>> {
+ let mut stream = request.into_inner();
+ let mut requests = Vec::new();
+ while let Some(request) = stream.message().await? {
+ requests.push(request);
+ }
+ Ok(Response::new(fbs_aggregate(&requests)))
+ }
+
+ async fn bidi_stream_message(
+ &self,
+ request: Request<tonic::Streaming<GrpcFbsRequest>>,
+ ) -> StatusResult<Response<Self::BidiStreamMessageStream>> {
+ let mut stream = request.into_inner();
+ let (tx, rx) = tokio::sync::mpsc::channel(4);
+ tokio::spawn(async move {
+ let mut index = 0;
+ loop {
+ match stream.message().await {
+ Ok(Some(request)) => {
+ let response = fbs_response(&request,
&format!("bidi-{index}"), index);
+ if tx.send(Ok(response)).await.is_err() {
+ break;
+ }
+ index += 1;
+ }
+ Ok(None) => break,
+ Err(error) => {
+ let _ = tx.send(Err(error)).await;
+ break;
+ }
+ }
+ }
+ });
+ Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
+ }
+
+ async fn unary_union(
+ &self,
+ request: Request<GrpcFbsUnion>,
+ ) -> StatusResult<Response<GrpcFbsUnion>> {
+ let request = fbs_request_from_union(request.into_inner())?;
+ Ok(Response::new(GrpcFbsUnion::GrpcFbsResponse(fbs_response(
+ &request, "unary", 10,
+ ))))
+ }
+
+ async fn server_stream_union(
+ &self,
+ request: Request<GrpcFbsUnion>,
+ ) -> StatusResult<Response<Self::ServerStreamUnionStream>> {
+ let request = fbs_request_from_union(request.into_inner())?;
+ Ok(Response::new(Box::pin(tokio_stream::iter(
+ (0..3)
+ .map(|index| {
+ Ok(GrpcFbsUnion::GrpcFbsResponse(fbs_response(
+ &request,
+ &format!("server-{index}"),
+ index,
+ )))
+ })
+ .collect::<Vec<_>>(),
+ ))))
+ }
+
+ async fn client_stream_union(
+ &self,
+ request: Request<tonic::Streaming<GrpcFbsUnion>>,
+ ) -> StatusResult<Response<GrpcFbsUnion>> {
+ let mut stream = request.into_inner();
+ let mut requests = Vec::new();
+ while let Some(union) = stream.message().await? {
+ requests.push(fbs_request_from_union(union)?);
+ }
+ Ok(Response::new(GrpcFbsUnion::GrpcFbsResponse(fbs_aggregate(
+ &requests,
+ ))))
+ }
+
+ async fn bidi_stream_union(
+ &self,
+ request: Request<tonic::Streaming<GrpcFbsUnion>>,
+ ) -> StatusResult<Response<Self::BidiStreamUnionStream>> {
+ let mut stream = request.into_inner();
+ let (tx, rx) = tokio::sync::mpsc::channel(4);
+ tokio::spawn(async move {
+ let mut index = 0;
+ loop {
+ match stream.message().await {
+ Ok(Some(union)) => match fbs_request_from_union(union) {
+ Ok(request) => {
+ let response =
GrpcFbsUnion::GrpcFbsResponse(fbs_response(
+ &request,
+ &format!("bidi-{index}"),
+ index,
+ ));
+ if tx.send(Ok(response)).await.is_err() {
+ break;
+ }
+ index += 1;
+ }
+ Err(error) => {
+ let _ = tx.send(Err(error)).await;
+ break;
+ }
+ },
+ Ok(None) => break,
+ Err(error) => {
+ let _ = tx.send(Err(error)).await;
+ break;
+ }
+ }
+ }
+ });
+ Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
+ }
+}
+
+// Builds a protobuf request.
+fn pb_request(id: &str, count: u32, payload: grpc_pb_request::Payload) ->
GrpcPbRequest {
+ GrpcPbRequest {
+ id: id.to_string(),
+ count,
+ payload: Some(payload),
+ }
+}
+
+// Maps a protobuf request oneof payload to the expected response payload.
+fn pb_response_payload(
+ payload: &Option<grpc_pb_request::Payload>,
+ tag: &str,
+ offset: u32,
+) -> StatusResult<Option<grpc_pb_response::Payload>> {
+ match payload {
+ None => Ok(None),
+ Some(grpc_pb_request::Payload::Text(value)) =>
Ok(Some(grpc_pb_response::Payload::Text(
+ format!("{tag}:{value}"),
+ ))),
+ Some(grpc_pb_request::Payload::Number(value)) => {
+ Ok(Some(grpc_pb_response::Payload::Number(value + offset)))
+ }
+ Some(grpc_pb_request::Payload::Unknown(_)) =>
Err(Status::invalid_argument(
+ "expected grpc_pb_request::Payload text or number",
+ )),
+ }
+}
+
+// Builds the expected protobuf response for a request.
+fn pb_response(request: &GrpcPbRequest, tag: &str, offset: u32) ->
StatusResult<GrpcPbResponse> {
+ Ok(GrpcPbResponse {
+ id: format!("{tag}:{}", request.id),
+ count: request.count + offset,
+ payload: pb_response_payload(&request.payload, tag, offset)?,
+ })
+}
+
+// Builds the aggregate protobuf response used by client-streaming RPCs.
+fn pb_aggregate(requests: &[GrpcPbRequest]) -> GrpcPbResponse {
+ GrpcPbResponse {
+ id: format!(
+ "client:{}",
+ requests
+ .iter()
+ .map(|request| request.id.as_str())
+ .collect::<Vec<_>>()
+ .join("+")
+ ),
+ count: requests.iter().map(|request| request.count).sum(),
+ payload: Some(grpc_pb_response::Payload::Text(format!(
+ "client:{}",
+ requests
+ .iter()
+ .map(|request| request.id.as_str())
+ .collect::<Vec<_>>()
+ .join("+")
+ ))),
+ }
+}
+
+async fn exercise_pb(channel: Channel) -> AnyResult<()> {
+ let mut client = PbGrpcServiceClient::new(channel);
+ let requests = vec![
+ pb_request(
+ "pb-a",
+ 9,
+ grpc_pb_request::Payload::Text("alpha".to_string()),
+ ),
+ pb_request("pb-b", 10, grpc_pb_request::Payload::Number(42)),
+ ];
+ let first = requests[0].clone();
+ assert_eq!(
+ client.unary_message(first.clone()).await?.into_inner(),
+ pb_response(&first, "unary", 10)?
+ );
+ assert_eq!(
+ collect_stream(
+ client
+ .server_stream_message(first.clone())
+ .await?
+ .into_inner()
+ )
+ .await?,
+ (0..3)
+ .map(|index| pb_response(&first, &format!("server-{index}"),
index))
+ .collect::<StatusResult<Vec<_>>>()?
+ );
+ assert_eq!(
+ client
+ .client_stream_message(tokio_stream::iter(requests.clone()))
+ .await?
+ .into_inner(),
+ pb_aggregate(&requests)
+ );
+ assert_eq!(
+ collect_stream(
+ client
+ .bidi_stream_message(tokio_stream::iter(requests.clone()))
+ .await?
+ .into_inner()
+ )
+ .await?,
+ requests
+ .iter()
+ .enumerate()
+ .map(|(index, request)| pb_response(request,
&format!("bidi-{index}"), index as u32))
+ .collect::<StatusResult<Vec<_>>>()?
+ );
+ Ok(())
+}
+
+struct PbServiceImpl;
+
+#[tonic::async_trait]
+impl PbGrpcService for PbServiceImpl {
+ type ServerStreamMessageStream = GrpcStream<GrpcPbResponse>;
+ type BidiStreamMessageStream = GrpcStream<GrpcPbResponse>;
+
+ async fn unary_message(
+ &self,
+ request: Request<GrpcPbRequest>,
+ ) -> StatusResult<Response<GrpcPbResponse>> {
+ Ok(Response::new(pb_response(
+ &request.into_inner(),
+ "unary",
+ 10,
+ )?))
+ }
+
+ async fn server_stream_message(
+ &self,
+ request: Request<GrpcPbRequest>,
+ ) -> StatusResult<Response<Self::ServerStreamMessageStream>> {
+ let request = request.into_inner();
+ let responses = (0..3)
+ .map(|index| pb_response(&request, &format!("server-{index}"),
index))
+ .collect::<StatusResult<Vec<_>>>()?;
+ Ok(Response::new(Box::pin(tokio_stream::iter(
+ responses.into_iter().map(Ok).collect::<Vec<_>>(),
+ ))))
+ }
+
+ async fn client_stream_message(
+ &self,
+ request: Request<tonic::Streaming<GrpcPbRequest>>,
+ ) -> StatusResult<Response<GrpcPbResponse>> {
+ let mut stream = request.into_inner();
+ let mut requests = Vec::new();
+ while let Some(request) = stream.message().await? {
+ requests.push(request);
+ }
+ Ok(Response::new(pb_aggregate(&requests)))
+ }
+
+ async fn bidi_stream_message(
+ &self,
+ request: Request<tonic::Streaming<GrpcPbRequest>>,
+ ) -> StatusResult<Response<Self::BidiStreamMessageStream>> {
+ let mut stream = request.into_inner();
+ let (tx, rx) = tokio::sync::mpsc::channel(4);
+ tokio::spawn(async move {
+ let mut index = 0;
+ loop {
+ match stream.message().await {
+ Ok(Some(request)) => {
+ match pb_response(&request, &format!("bidi-{index}"),
index) {
+ Ok(response) => {
+ if tx.send(Ok(response)).await.is_err() {
+ break;
+ }
+ index += 1;
+ }
+ Err(error) => {
+ let _ = tx.send(Err(error)).await;
+ break;
+ }
+ }
+ }
+ Ok(None) => break,
+ Err(error) => {
+ let _ = tx.send(Err(error)).await;
+ break;
+ }
+ }
+ }
+ });
+ Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
