diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d98bddbc17..dc0cbaa771 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -855,6 +855,50 @@ jobs: cd integration_tests/grpc_tests/java mvn -T16 --no-transfer-progress -Dtest=RustGrpcTest test + grpc_java_cpp_tests: + name: Java/C++ gRPC Tests + needs: changes + if: needs.changes.outputs.grpc_tests == 'true' + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - name: Set up JDK 21 + uses: actions/setup-java@v4 + with: + java-version: 21 + distribution: "temurin" + - name: Set up Python 3.11 + uses: actions/setup-python@v5 + with: + python-version: 3.11 + cache: "pip" + - name: Set up Bazel + uses: bazel-contrib/setup-bazel@c5acdfb288317d0b5c0bbd7a396a3dc868bb0f86 # 0.19.0 + with: + bazelisk-cache: true + bazelisk-version: "1.x" + - name: Cache Maven local repository + uses: actions/cache@v4 + with: + path: ~/.m2/repository + key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }} + restore-keys: | + ${{ runner.os }}-maven- + - 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: Generate gRPC test sources + run: python integration_tests/grpc_tests/generate_grpc.py + - name: Build C++ gRPC peer + run: | + cd cpp + bazel build //integration_tests/grpc_tests/cpp:grpc_interop --config=x86_64 + - name: Run Java/C++ gRPC tests + run: | + cd integration_tests/grpc_tests/java + mvn -T16 --no-transfer-progress -Dtest=CppGrpcTest test + grpc_java_kotlin_tests: name: Java/Kotlin gRPC Tests needs: changes diff --git a/MODULE.bazel b/MODULE.bazel index c96da3f5ea..55f46eea54 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -75,7 +75,12 @@ use_repo( bazel_dep(name = "cython", version = "3.1.3") # Google Test -bazel_dep(name = "googletest", version = "1.15.2") +bazel_dep(name = "googletest", version = "1.17.0") + +# gRPC C++ integration tests +bazel_dep(name = "grpc", version = "1.75.0") +bazel_dep(name = "protobuf", version = "31.1") +single_version_override(module_name = "grpc-java", version = "1.75.0") # Hedron's Compile Commands Extractor for Bazel bazel_dep(name = "hedron_compile_commands", dev_dependency = True) diff --git a/compiler/README.md b/compiler/README.md index 2fb48b3a20..d7eebb98e3 100644 --- a/compiler/README.md +++ b/compiler/README.md @@ -10,7 +10,7 @@ The FDL compiler generates cross-language serialization code from schema definit - **Type ID and namespace support**: Both numeric IDs and name-based type registration - **Field modifiers**: Optional fields, reference tracking, list fields, scalar encoding modifiers - **File imports**: Modular schemas with import support -- **gRPC service generation**: Native gRPC stubs and service bases for Java, Python, Go, Rust, C#, JavaScript, Dart, Kotlin, and Scala +- **gRPC service generation**: Native gRPC stubs and service bases for Java, Python, Go, Rust, C++, C#, JavaScript, Dart, Kotlin, and Scala ## Documentation @@ -335,6 +335,7 @@ fory_compiler/ ├── python.py # Python gRPC companion module (grpcio style) ├── go.py # Go gRPC stub generator (google.golang.org/grpc) ├── rust.py # Rust gRPC service module (tonic style) + ├── cpp.py # C++ synchronous gRPC service companions ├── csharp.py # C# gRPC service companion (Grpc.Core style) ├── javascript.py # JavaScript Node.js and gRPC-Web client generators ├── dart.py # Dart gRPC service companion diff --git a/compiler/fory_compiler/generators/cpp.py b/compiler/fory_compiler/generators/cpp.py index 4d9d6a937b..cc98b2d9dd 100644 --- a/compiler/fory_compiler/generators/cpp.py +++ b/compiler/fory_compiler/generators/cpp.py @@ -17,11 +17,13 @@ """C++ code generator.""" +import re from pathlib import Path from typing import Dict, List, Optional, Set, Tuple import typing from fory_compiler.generators.base import BaseGenerator, GeneratedFile +from fory_compiler.generators.services.cpp import CppServiceGeneratorMixin from fory_compiler.frontend.utils import parse_idl_file from fory_compiler.ir.ast import ( Message, @@ -39,7 +41,7 @@ from fory_compiler.ir.types import PrimitiveKind -class CppGenerator(BaseGenerator): +class CppGenerator(CppServiceGeneratorMixin, BaseGenerator): """Generates C++ classes with FORY_STRUCT macros.""" language_name = "cpp" @@ -741,7 +743,7 @@ def _get_union_selector_base(self, union: Union) -> str: def _ensure_name_caches(self, schema: Schema) -> None: """Construct the naming caches once for a schema file.""" if not hasattr(self, "_named_schema_ids"): - # Init everything. + # We don't initialize cache for gRPC code generation here. self._named_schema_ids: Set[int] = set() self._type_identifier_cache: Dict[Tuple[object, ...], str] = {} self._field_identifier_cache: Dict[ @@ -883,12 +885,19 @@ def generate_bytes_methods(self, class_name: str, indent: str) -> List[str]: lines.append(f"{indent} return {detail}::get_fory().serialize(*this);") lines.append(f"{indent}}}") lines.append("") + # gRPC payload deserialization would require this. lines.append( - f"{indent}static ::fory::Result<{class_name}, ::fory::Error> from_bytes(const ::std::vector<::uint8_t>& data) {{" + f"{indent}static ::fory::Result<{class_name}, ::fory::Error> from_bytes(const ::uint8_t* data, ::std::size_t size) {{" + ) + lines.append( + f"{indent} return {detail}::get_fory().deserialize<{class_name}>(data, size);" ) + lines.append(f"{indent}}}") + lines.append("") lines.append( - f"{indent} return {detail}::get_fory().deserialize<{class_name}>(data);" + f"{indent}static ::fory::Result<{class_name}, ::fory::Error> from_bytes(const ::std::vector<::uint8_t>& data) {{" ) + lines.append(f"{indent} return from_bytes(data.data(), data.size());") lines.append(f"{indent}}}") return lines @@ -911,7 +920,7 @@ def generate_header(self) -> GeneratedFile: includes.add("") includes.add('"fory/serialization/fory.h"') if self.schema_has_unions(): - includes.add("") # todo: what's this?? + includes.add("") includes.add("") includes.add("") includes.add("") @@ -936,14 +945,11 @@ def generate_header(self) -> GeneratedFile: self.collect_union_includes(union, includes) # License header - lines.append("/*") - for line in self.get_license_header(" *").split("\n"): - lines.append(line) - lines.append(" */") + lines.extend(self._cpp_license_lines()) lines.append("") # Header guard - guard_name = f"{self.get_header_name().upper()}_H_" + guard_name = self._cpp_header_guard(f"{self.get_header_name()}.h") lines.append(f"#ifndef {guard_name}") lines.append(f"#define {guard_name}") lines.append("") @@ -1024,6 +1030,24 @@ def generate_header(self) -> GeneratedFile: content="\n".join(lines), ) + def _cpp_license_lines(self) -> List[str]: + """Generate the Apache license block for generated C++ files.""" + lines = ["/*"] + lines.extend(self.get_license_header(" *").split("\n")) + lines.append(" */") + return lines + + def _cpp_header_guard(self, path: str) -> str: + """Generate a header guard name for a path.""" + # Header names can contain '.', '-', '/', or other characters outside + # the conservative ASCII macro identifier set used by generated headers. + return re.sub(r"[^A-Za-z0-9]", "_", path).upper() + "_" + + def indent_lines(self, lines: List[str], level: int) -> List[str]: + """Indent a list of lines by the given level.""" + prefix = " " * level # C++ uses two-spaces style. + return [f"{prefix}{line}" if line else line for line in lines] + def collect_message_includes(self, message: Message, includes: Set[str]): """Collect includes for a message and its nested types recursively.""" for field in message.fields: diff --git a/compiler/fory_compiler/generators/rust.py b/compiler/fory_compiler/generators/rust.py index 54ba2f7588..e705603ba4 100644 --- a/compiler/fory_compiler/generators/rust.py +++ b/compiler/fory_compiler/generators/rust.py @@ -407,7 +407,7 @@ def _allocate_scoped_message_identifiers(self, message: Message) -> None: def _ensure_name_caches(self, schema: Schema) -> None: """Construct the naming caches once for a schema file.""" if not hasattr(self, "_named_schema_ids"): - # Init everything. + # We don't initialize cache for gRPC code generation here. self._named_schema_ids: Set[int] = set() self._type_identifier_cache: Dict[Tuple[object, ...], str] = {} self._module_identifier_cache: Dict[Tuple[object, ...], str] = {} @@ -420,26 +420,6 @@ def _ensure_name_caches(self, schema: Schema) -> None: self._union_case_identifier_cache: Dict[ Tuple[object, ...], Dict[Tuple[object, ...], str] ] = {} - self._named_service_schema_ids: Set[int] = set() - self._service_trait_identifier_cache: Dict[Tuple[object, ...], str] = {} - self._service_client_module_identifier_cache: Dict[ - Tuple[object, ...], str - ] = {} - self._service_server_module_identifier_cache: Dict[ - Tuple[object, ...], str - ] = {} - self._service_name_constant_identifier_cache: Dict[ - Tuple[object, ...], str - ] = {} - self._rpc_method_identifier_cache: Dict[ - Tuple[object, ...], Dict[Tuple[object, ...], str] - ] = {} - self._rpc_stream_type_identifier_cache: Dict[ - Tuple[object, ...], Dict[Tuple[object, ...], str] - ] = {} - self._rpc_path_constant_identifier_cache: Dict[ - Tuple[object, ...], Dict[Tuple[object, ...], str] - ] = {} schema_id = id(schema) if schema_id in self._named_schema_ids: # Cache exists. diff --git a/compiler/fory_compiler/generators/services/cpp.py b/compiler/fory_compiler/generators/services/cpp.py new file mode 100644 index 0000000000..b236eb937d --- /dev/null +++ b/compiler/fory_compiler/generators/services/cpp.py @@ -0,0 +1,830 @@ +# 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. + +"""C++ gRPC service generator helpers.""" + +import hashlib +import re +from typing import Dict, List, Set, Tuple + +from fory_compiler.generators.base import GeneratedFile +from fory_compiler.generators.services.base import StreamingMode, streaming_mode +from fory_compiler.ir.ast import NamedType, RpcMethod, Service + + +class CppServiceGeneratorMixin: + """Generates C++ gRPC service companions.""" + + def generate_services(self) -> List[GeneratedFile]: + """Generate C++ 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.allocate_grpc_service_identifiers(local_services) + return [ + self.generate_service_api_header(local_services), + self.generate_service_grpc_header(local_services), + self.generate_service_grpc_source(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) + if not hasattr(self, "_named_service_schema_ids"): + self._named_service_schema_ids: Set[int] = set() + self._service_interface_identifier_cache: Dict[Tuple[object, ...], str] = {} + self._service_stub_identifier_cache: Dict[Tuple[object, ...], str] = {} + self._service_grpc_identifier_cache: Dict[Tuple[object, ...], str] = {} + self._service_name_constant_identifier_cache: Dict[ + Tuple[object, ...], str + ] = {} + self._rpc_method_identifier_cache: Dict[ + Tuple[object, ...], Dict[Tuple[object, ...], str] + ] = {} + self._rpc_path_constant_identifier_cache: Dict[ + Tuple[object, ...], Dict[Tuple[object, ...], str] + ] = {} + self._rpc_method_member_identifier_cache: Dict[ + Tuple[object, ...], Dict[Tuple[object, ...], str] + ] = {} + + schema_id = id(self.schema) + if schema_id in self._named_service_schema_ids: + return + + # Reserve the fixed service companion namespace in the schema scope. + # e.g.: + # + # namespace demo; + # + # message service { + # string name = 1; + # } + # + # service Greeter { + # rpc SayHello (service) returns (service); + # } + # + # This IDL itself is valid. But in our design, both `message service` + # and `service Greeter` would take up the same `::demo::service` namespace. + # We should detect this kind of collision and throw errors to the user. + used_schema_names: Dict[str, str] = {} + for type_def in self.schema.enums + self.schema.unions + self.schema.messages: + if not self.is_imported_type(type_def): + identifier = self.get_type_identifier(type_def) + used_schema_names[identifier] = f"model type {type_def.name!r}" + self._allocate_scoped_identifier( + "service", + used_schema_names, + "C++ gRPC schema namespace", + "generated service namespace", + ) + # For package name `demo`, + # `used_service_names` checks collisions under `demo::service`, + # `used_grpc_names` checks collisions under `demo::service::grpc`. + used_service_names: Dict[str, str] = {} + self._allocate_scoped_identifier( + "grpc", + used_service_names, + "C++ gRPC service namespace", + "generated gRPC transport namespace", + ) + used_grpc_names: Dict[str, str] = {} + for service in services: + service_key = self._cache_key(service) + self._service_interface_identifier_cache[service_key] = ( + self._allocate_scoped_identifier( + service.name, + used_service_names, + "C++ gRPC service namespace", + f"service interface {service.name}", + ) + ) + self._service_stub_identifier_cache[service_key] = ( + self._allocate_scoped_identifier( + f"{service.name}Stub", + used_grpc_names, + "C++ gRPC transport namespace", + f"service stub {service.name}", + ) + ) + self._service_grpc_identifier_cache[service_key] = ( + self._allocate_scoped_identifier( + f"{service.name}ServiceGrpc", + used_grpc_names, + "C++ gRPC transport namespace", + f"gRPC service wrapper {service.name}", + ) + ) + self._service_name_constant_identifier_cache[service_key] = ( + self._allocate_scoped_identifier( + f"{self.to_pascal_case(service.name)}ServiceName", + used_service_names, + "C++ gRPC service namespace", + f"service name constant for {service.name}", + ) + ) + stub_name = self._service_stub_identifier_cache[service_key] + + # Even though C++ allows function overloading, at IDL level, method names in the same service cannot be identical, + # since gRPC relies on their names to locate the handler. + used_methods: Dict[str, str] = {} + used_members: Dict[str, str] = {} + method_names: Dict[Tuple[object, ...], str] = {} + path_constants: Dict[Tuple[object, ...], str] = {} + rpc_members: Dict[Tuple[object, ...], str] = {} + for method in service.methods: + method_key = self._cache_key(method) + method_names[method_key] = self._allocate_scoped_identifier( + method.name, + used_methods, + f"C++ gRPC service {service.name} methods", + method.name, + ) + method_name = method_names[method_key] + # Client stub code uses `channel_` as a data member name. + if method_name == "channel_" or method_name in used_members: + raise ValueError( + f"C++ name collision in gRPC stub {stub_name}: method " + f"{method.name!r} conflicts with member {method_name!r}" + ) + used_members[method_name] = method.name + rpc_members[method_key] = self._allocate_scoped_identifier( + f"rpcmethod_{method_name}_", + used_members, + f"C++ gRPC stub {stub_name} members", + f"RPC method member for {service.name}.{method.name}", + ) + path_constants[method_key] = self._allocate_scoped_identifier( + f"{self.to_pascal_case(service.name)}" + f"{self.to_pascal_case(method.name)}Path", + used_service_names, + "C++ gRPC service namespace", + f"RPC path constant for {service.name}.{method.name}", + ) + self._rpc_method_identifier_cache[service_key] = method_names + self._rpc_method_member_identifier_cache[service_key] = rpc_members + self._rpc_path_constant_identifier_cache[service_key] = path_constants + + self._named_service_schema_ids.add(schema_id) + + def generate_service_api_header(self, services: List[Service]) -> GeneratedFile: + """Generate C++ service API header file (service.h).""" + path = f"{self.get_header_name()}_service.h" + guard = self._cpp_header_guard(path) + lines = self._cpp_license_lines() + lines.extend( + [ + "", + f"#ifndef {guard}", + f"#define {guard}", + "", + f'#include "{self.get_header_name()}.h"', + "", + "#include ", + "#include ", + "#include ", + "", + ] + ) + service_namespace = self.service_namespace() + lines.extend([f"namespace {service_namespace} {{", ""]) + for i, service in enumerate(services): + if i > 0: + lines.append("") + lines.extend(self.generate_service_interface(service)) + lines.append("") + lines.extend(self.generate_service_constants(service)) + lines.extend(["", f"}} // namespace {service_namespace}"]) + lines.extend(["", f"#endif // {guard}", ""]) # End header guard. + return GeneratedFile(path=path, content="\n".join(lines)) + + def generate_service_interface(self, service: Service) -> List[str]: + """Generate gRPC service interface.""" + service_key = self._cache_key(service) + interface_name = self._service_interface_identifier_cache[service_key] + lines = [ + f"class {interface_name} {{", + " public:", + ] + lines.extend(self.indent_lines([f"virtual ~{interface_name}() = default;"], 1)) + 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 gRPC 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) + mode = streaming_mode(method) + lines = [f"virtual ::grpc::Status {method_name}("] + if mode == StreamingMode.UNARY: + lines.extend( + [ + " ::grpc::ServerContext* context,", + f" const {request_type}* request,", + f" {response_type}* response) = 0;", + ] + ) + elif mode == StreamingMode.CLIENT_STREAMING: + lines.extend( + [ + " ::grpc::ServerContext* context,", + f" ::grpc::ServerReader<{request_type}>* reader,", + f" {response_type}* response) = 0;", + ] + ) + elif mode == StreamingMode.SERVER_STREAMING: + lines.extend( + [ + " ::grpc::ServerContext* context,", + f" const {request_type}* request,", + f" ::grpc::ServerWriter<{response_type}>* writer) = 0;", + ] + ) + else: + lines.extend( + [ + " ::grpc::ServerContext* context,", + " ::grpc::ServerReaderWriter<", + f" {response_type}, {request_type}>* stream) = 0;", + ] + ) + return lines + + def service_type_path(self, named_type: NamedType) -> str: + """Get C++ path for a gRPC request or response type.""" + resolved = self.schema.get_type(named_type.name) + if resolved is None: + raise ValueError(f"Unknown gRPC message type {named_type.name!r}") + return self.generate_namespaced_type( + named_type, parent_stack=None, global_qualify=True + ) + + def generate_service_constants(self, service: Service) -> List[str]: + """Generate service and gRPC path constants.""" + 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'inline constexpr char {service_name_const}[] = "{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"inline constexpr char {rpc_path_const}[] = " + f'"{self.get_grpc_method_path(service, method)}";' + ) + return lines + + def generate_service_grpc_header(self, services: List[Service]) -> GeneratedFile: + """Generate C++ service transport binding header file (service_grpc.h).""" + path = f"{self.get_header_name()}_service_grpc.h" + guard = self._cpp_header_guard(path) + lines = self._cpp_license_lines() + lines.extend( + [ + "", + f"#ifndef {guard}", + f"#define {guard}", + "", + f'#include "{self.get_header_name()}_service.h"', + "", + "#include ", + "#include ", + "#include ", + "#include ", + "#include ", + "", + "#include ", + "#include ", + "#include ", + "#include ", + "#include ", + "#include ", + "#include ", + "#include ", + "#include ", + "#include ", + "#include ", + "", + ] + ) + lines.extend(self.generate_grpc_codec()) + payload_impls = self.generate_grpc_payload_impls(services) + if payload_impls: + lines.append("") + lines.extend(payload_impls) + grpc_namespace = f"{self.service_namespace()}::grpc" + lines.extend(["", f"namespace {grpc_namespace} {{"]) + for service in services: + lines.append("") + lines.extend(self.generate_grpc_client_declaration(service)) + lines.append("") + lines.extend(self.generate_grpc_server_declaration(service)) + lines.extend(["", f"}} // namespace {grpc_namespace}"]) + lines.extend(["", f"#endif // {guard}", ""]) # End header guard. + return GeneratedFile(path=path, content="\n".join(lines)) + + def generate_grpc_codec(self) -> List[str]: + """Generate Fory-backed codec used by C++ gRPC stubs, instead of relying on a separate file.""" + return [ + "#ifndef FORY_GENERATED_GRPC_SERIALIZATION_TRAITS_", + "#define FORY_GENERATED_GRPC_SERIALIZATION_TRAITS_", + "", + "namespace fory::grpc::detail {", + "", + "template ", + "class ForyGrpcSerializationTraits {", + " public:", + " static ::grpc::Status Serialize(", + " ::grpc_event_engine::experimental::MemoryAllocator* allocator,", + " const Message& message, ::grpc::ByteBuffer* buffer,", + " bool* own_buffer) {", + " (void)allocator;", + " return Serialize(message, buffer, own_buffer);", + " }", + "", + " static ::grpc::Status Serialize(const Message& message,", + " ::grpc::ByteBuffer* buffer,", + " bool* own_buffer) {", + " if (buffer == nullptr || own_buffer == nullptr) {", + " return ::grpc::Status(::grpc::StatusCode::INTERNAL,", + ' "Missing gRPC serialization output");', + " }", + " auto result = message.to_bytes();", + " if (!result.ok()) {", + " return ::grpc::Status(::grpc::StatusCode::INTERNAL,", + " result.error().to_string());", + " }", + " auto* bytes = new ::std::vector<::uint8_t>(", + " ::std::move(result).value());", + " ::grpc::Slice slice;", + " if (bytes->empty()) {", + " delete bytes;", + " slice = ::grpc::Slice(0);", + " } else {", + " slice = ::grpc::Slice(bytes->data(), bytes->size(),", + " &ForyGrpcSerializationTraits::DestroyBytes,", + " bytes);", + " }", + " ::grpc::ByteBuffer encoded(&slice, 1);", + " buffer->Swap(&encoded);", + " *own_buffer = true;", + " return ::grpc::Status::OK;", + " }", + "", + " static ::grpc::Status Deserialize(::grpc::ByteBuffer* buffer,", + " Message* message) {", + " if (buffer == nullptr || message == nullptr || !buffer->Valid()) {", + " return ::grpc::Status(::grpc::StatusCode::INTERNAL,", + ' "Missing gRPC message bytes");', + " }", + " ::grpc::Slice slice;", + " auto status = buffer->TrySingleSlice(&slice);", + " if (!status.ok()) {", + " status = buffer->DumpToSingleSlice(&slice);", + " if (!status.ok()) {", + " return status;", + " }", + " }", + " auto result = Message::from_bytes(slice.begin(), slice.size());", + " buffer->Clear();", + " if (!result.ok()) {", + " return ::grpc::Status(::grpc::StatusCode::INTERNAL,", + " result.error().to_string());", + " }", + " *message = ::std::move(result).value();", + " return ::grpc::Status::OK;", + " }", + "", + " private:", + " static void DestroyBytes(void* data) {", + " delete static_cast<::std::vector<::uint8_t>*>(data);", + " }", + "};", + "", + "} // namespace fory::grpc::detail", + "", + "#endif // FORY_GENERATED_GRPC_SERIALIZATION_TRAITS_", + ] + + def generate_grpc_payload_impls(self, services: List[Service]) -> List[str]: + """Generate `SerializationTraits` specializations 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 `SerializationTraits` specializations for one gRPC request or response type.""" + # Each type-specific `grpc::SerializationTraits` specialization needs a + # stable include guard because the same payload type can be referenced by + # multiple generated service headers. + # See the test_cpp_grpc_shared_message_specialization_guard case in test_service_codege.py as an example. + # C++ type paths such as `::demo::greeter::Hello` are not valid macro identifiers, + # so the guard uses a readable sanitized prefix like `DEMO_GREETER_HELLO` plus a digest + # of the original type path to avoid collisions. + readable = re.sub(r"[^A-Za-z0-9]+", "_", type_path).strip("_").upper() + # We use sha256 to compute the digest since its stability. + digest = hashlib.sha256(type_path.encode("utf-8")).hexdigest()[:12].upper() + guard = f"FORY_GENERATED_GRPC_TRAITS_{readable}_{digest}_" + return [ + f"#ifndef {guard}", + f"#define {guard}", + "namespace grpc {", + "template <>", + f"class SerializationTraits<{type_path}, void>", + " : public ::fory::grpc::detail::ForyGrpcSerializationTraits<", + f" {type_path}> {{}};", + "} // namespace grpc", + f"#endif // {guard}", + ] + + def generate_service_grpc_source(self, services: List[Service]) -> GeneratedFile: + """Generate gRPC client and server transport definitions (service_grpc.cc).""" + path = f"{self.get_header_name()}_service_grpc.cc" + lines = self._cpp_license_lines() + lines.extend( + [ + "", + f'#include "{self.get_header_name()}_service_grpc.h"', + "", + "#include ", + "#include ", + "#include ", + "", + ] + ) + grpc_namespace = f"{self.service_namespace()}::grpc" + lines.extend([f"namespace {grpc_namespace} {{", ""]) + for i, service in enumerate(services): + if i > 0: + lines.append("") + lines.extend(self.generate_grpc_client_definition(service)) + lines.append("") + lines.extend(self.generate_grpc_server_definition(service)) + lines.extend(["", f"}} // namespace {grpc_namespace}"]) + lines.append("") + return GeneratedFile(path=path, content="\n".join(lines)) + + def generate_grpc_client_declaration(self, service: Service) -> List[str]: + """Generate gRPC client declaration in service_grpc.h.""" + service_key = self._cache_key(service) + stub_name = self._service_stub_identifier_cache[service_key] + lines = [ + f"class {stub_name} final {{", + " public:", + ] + lines.extend( + self.indent_lines( + [ + f"static ::std::unique_ptr<{stub_name}> NewStub(", + " const ::std::shared_ptr<::grpc::ChannelInterface>& channel,", + " const ::grpc::StubOptions& options = ::grpc::StubOptions());", + ], + 1, + ) + ) + for method in service.methods: + lines.append("") + lines.extend( + self.indent_lines( + self.generate_grpc_client_method_declaration(service, method), 1 + ) + ) + lines.extend( + [ + "", + " private:", + ] + ) + lines.extend( + self.indent_lines( + [ + f"{stub_name}(", + " const ::std::shared_ptr<::grpc::ChannelInterface>& channel,", + " const ::grpc::StubOptions& options);", + ], + 1, + ) + ) + lines.append("") + lines.extend( + self.indent_lines( + ["::std::shared_ptr<::grpc::ChannelInterface> channel_;"], 1 + ) + ) + for method in service.methods: + member = self._rpc_method_member_identifier_cache[service_key][ + self._cache_key(method) + ] + lines.extend( + self.indent_lines([f"::grpc::internal::RpcMethod {member};"], 1) + ) + lines.append("};") + return lines + + def generate_grpc_client_definition(self, service: Service) -> List[str]: + """Generate gRPC client implementation in service_grpc.cc.""" + service_key = self._cache_key(service) + stub_name = self._service_stub_identifier_cache[service_key] + lines = [ + f"::std::unique_ptr<{stub_name}> {stub_name}::NewStub(", + " const ::std::shared_ptr<::grpc::ChannelInterface>& channel,", + " const ::grpc::StubOptions& options) {", + f" return ::std::unique_ptr<{stub_name}>(new {stub_name}(channel, options));", + "}", + "", + f"{stub_name}::{stub_name}(", + " const ::std::shared_ptr<::grpc::ChannelInterface>& channel,", + " const ::grpc::StubOptions& options)", + f" : channel_(channel){',' if service.methods else ''}", + ] + for i, method in enumerate(service.methods): + method_key = self._cache_key(method) + member = self._rpc_method_member_identifier_cache[service_key][method_key] + path_constant = self.service_path_constant_type_path(service, method) + method_type = self._grpc_method_type(method) + suffix = "," if i + 1 < len(service.methods) else "" + lines.extend( + [ + f" {member}({path_constant}, options.suffix_for_stats(),", + f" ::grpc::internal::RpcMethod::{method_type}, channel){suffix}", + ] + ) + lines.append("{}") + for method in service.methods: + lines.append("") + lines.extend(self.generate_grpc_client_method_definition(service, method)) + return lines + + def generate_grpc_client_method_declaration( + self, service: Service, method: RpcMethod + ) -> List[str]: + """Generate gRPC client method declaration in service_grpc.h.""" + service_key = self._cache_key(service) + method_name = self._rpc_method_identifier_cache[service_key][ + self._cache_key(method) + ] + request_type = self.service_type_path(method.request_type) + response_type = self.service_type_path(method.response_type) + mode = streaming_mode(method) + if mode == StreamingMode.UNARY: + return [ + f"::grpc::Status {method_name}(", + " ::grpc::ClientContext* context,", + f" const {request_type}& request, {response_type}* response);", + ] + if mode == StreamingMode.CLIENT_STREAMING: + return [ + f"::std::unique_ptr<::grpc::ClientWriter<{request_type}>> {method_name}(", + " ::grpc::ClientContext* context,", + f" {response_type}* response);", + ] + if mode == StreamingMode.SERVER_STREAMING: + return [ + f"::std::unique_ptr<::grpc::ClientReader<{response_type}>> {method_name}(", + " ::grpc::ClientContext* context,", + f" const {request_type}& request);", + ] + return [ + "::std::unique_ptr<::grpc::ClientReaderWriter<", + f" {request_type}, {response_type}>> {method_name}(", + " ::grpc::ClientContext* context);", + ] + + def generate_grpc_client_method_definition( + self, service: Service, method: RpcMethod + ) -> List[str]: + """Generate gRPC client method implementation in service_grpc.cc.""" + service_key = self._cache_key(service) + method_key = self._cache_key(method) + stub_name = self._service_stub_identifier_cache[service_key] + method_name = self._rpc_method_identifier_cache[service_key][method_key] + member = self._rpc_method_member_identifier_cache[service_key][method_key] + request_type = self.service_type_path(method.request_type) + response_type = self.service_type_path(method.response_type) + mode = streaming_mode(method) + if mode == StreamingMode.UNARY: + return [ + f"::grpc::Status {stub_name}::{method_name}(", + " ::grpc::ClientContext* context,", + f" const {request_type}& request, {response_type}* response) {{", + " return ::grpc::internal::BlockingUnaryCall<", + f" {request_type}, {response_type}>(", + f" channel_.get(), {member}, context, request, response);", + "}", + ] + if mode == StreamingMode.CLIENT_STREAMING: + return [ + f"::std::unique_ptr<::grpc::ClientWriter<{request_type}>>", + f"{stub_name}::{method_name}(::grpc::ClientContext* context,", + f" {response_type}* response) {{", + f" return ::std::unique_ptr<::grpc::ClientWriter<{request_type}>>(", + f" ::grpc::internal::ClientWriterFactory<{request_type}>::Create(", + f" channel_.get(), {member}, context, response));", + "}", + ] + if mode == StreamingMode.SERVER_STREAMING: + return [ + f"::std::unique_ptr<::grpc::ClientReader<{response_type}>>", + f"{stub_name}::{method_name}(::grpc::ClientContext* context,", + f" const {request_type}& request) {{", + f" return ::std::unique_ptr<::grpc::ClientReader<{response_type}>>(", + f" ::grpc::internal::ClientReaderFactory<{response_type}>::Create(", + f" channel_.get(), {member}, context, request));", + "}", + ] + return [ + "::std::unique_ptr<::grpc::ClientReaderWriter<", + f" {request_type}, {response_type}>>", + f"{stub_name}::{method_name}(::grpc::ClientContext* context) {{", + " return ::std::unique_ptr<::grpc::ClientReaderWriter<", + f" {request_type}, {response_type}>>(", + " ::grpc::internal::ClientReaderWriterFactory<", + f" {request_type}, {response_type}>::Create(", + f" channel_.get(), {member}, context));", + "}", + ] + + def generate_grpc_server_declaration(self, service: Service) -> List[str]: + """Generate gRPC server declaration in service_grpc.h.""" + service_key = self._cache_key(service) + interface_type = self.service_interface_type_path(service) + grpc_name = self._service_grpc_identifier_cache[service_key] + lines = [ + f"class {grpc_name} final : public ::grpc::Service {{", + " public:", + ] + lines.extend( + self.indent_lines( + [ + "// The caller owns impl; it must outlive this adapter and server.", + f"explicit {grpc_name}({interface_type}* impl);", + ], + 1, + ) + ) + lines.extend(["", " private:"]) + lines.extend(self.indent_lines([f"{interface_type}* impl_;"], 1)) + lines.append("};") + return lines + + def generate_grpc_server_definition(self, service: Service) -> List[str]: + """Generate gRPC server implementation in service_grpc.cc.""" + service_key = self._cache_key(service) + interface_type = self.service_interface_type_path(service) + grpc_name = self._service_grpc_identifier_cache[service_key] + lines = [f"{grpc_name}::{grpc_name}({interface_type}* impl) : impl_(impl) {{"] + for method in service.methods: + lines.extend( + self.indent_lines(self.generate_grpc_server_route(service, method), 1) + ) + lines.append("}") + return lines + + def generate_grpc_server_route( + self, service: Service, method: RpcMethod + ) -> List[str]: + """Generate one RpcServiceMethod registration.""" + service_key = self._cache_key(service) + method_key = self._cache_key(method) + interface_type = self.service_interface_type_path(service) + method_name = self._rpc_method_identifier_cache[service_key][method_key] + path_constant = self.service_path_constant_type_path(service, method) + request_type = self.service_type_path(method.request_type) + response_type = self.service_type_path(method.response_type) + method_type = self._grpc_method_type(method) + mode = streaming_mode(method) + lines = [ + "AddMethod(new ::grpc::internal::RpcServiceMethod(", + f" {path_constant}, ::grpc::internal::RpcMethod::{method_type},", + ] + if mode == StreamingMode.UNARY: + lines.extend( + [ + " new ::grpc::internal::RpcMethodHandler<", + f" {interface_type}, {request_type}, {response_type}>(", + f" []({interface_type}* service, ::grpc::ServerContext* context,", + f" const {request_type}* request, {response_type}* response) {{", + f" return service->{method_name}(context, request, response);", + " },", + ] + ) + elif mode == StreamingMode.CLIENT_STREAMING: + lines.extend( + [ + " new ::grpc::internal::ClientStreamingHandler<", + f" {interface_type}, {request_type}, {response_type}>(", + f" []({interface_type}* service, ::grpc::ServerContext* context,", + f" ::grpc::ServerReader<{request_type}>* reader,", + f" {response_type}* response) {{", + f" return service->{method_name}(context, reader, response);", + " },", + ] + ) + elif mode == StreamingMode.SERVER_STREAMING: + lines.extend( + [ + " new ::grpc::internal::ServerStreamingHandler<", + f" {interface_type}, {request_type}, {response_type}>(", + f" []({interface_type}* service, ::grpc::ServerContext* context,", + f" const {request_type}* request,", + f" ::grpc::ServerWriter<{response_type}>* writer) {{", + f" return service->{method_name}(context, request, writer);", + " },", + ] + ) + else: + lines.extend( + [ + " new ::grpc::internal::BidiStreamingHandler<", + f" {interface_type}, {request_type}, {response_type}>(", + f" []({interface_type}* service, ::grpc::ServerContext* context,", + " ::grpc::ServerReaderWriter<", + f" {response_type}, {request_type}>* stream) {{", + f" return service->{method_name}(context, stream);", + " },", + ] + ) + lines.extend([" impl_)));", ""]) + return lines + + def _grpc_method_type(self, method: RpcMethod) -> str: + mode = streaming_mode(method) + return { + StreamingMode.UNARY: "NORMAL_RPC", + StreamingMode.CLIENT_STREAMING: "CLIENT_STREAMING", + StreamingMode.SERVER_STREAMING: "SERVER_STREAMING", + StreamingMode.BIDIRECTIONAL: "BIDI_STREAMING", + }[mode] + + def service_namespace(self) -> str: + schema_namespace = self.get_namespace() + if schema_namespace: + return f"{schema_namespace}::service" + return "service" + + def service_interface_type_path(self, service: Service) -> str: + service_key = self._cache_key(service) + interface_name = self._service_interface_identifier_cache[service_key] + return f"::{self.service_namespace()}::{interface_name}" + + def service_path_constant_type_path( + self, service: Service, method: RpcMethod + ) -> str: + service_key = self._cache_key(service) + method_key = self._cache_key(method) + constant = self._rpc_path_constant_identifier_cache[service_key][method_key] + return f"::{self.service_namespace()}::{constant}" diff --git a/compiler/fory_compiler/generators/services/rust.py b/compiler/fory_compiler/generators/services/rust.py index 16cf5954e1..0edba4a308 100644 --- a/compiler/fory_compiler/generators/services/rust.py +++ b/compiler/fory_compiler/generators/services/rust.py @@ -20,6 +20,7 @@ from typing import Dict, List, Optional, Set, Tuple from fory_compiler.generators.base import GeneratedFile +from fory_compiler.generators.services.base import StreamingMode, streaming_mode from fory_compiler.ir.ast import ( ArrayType, Field, @@ -57,6 +58,28 @@ def generate_services(self) -> List[GeneratedFile]: def allocate_grpc_service_identifiers(self, services: List[Service]) -> None: """Allocate sanitized identifiers used by generated gRPC code.""" self._ensure_name_caches(self.schema) + if not hasattr(self, "_named_service_schema_ids"): + self._named_service_schema_ids: Set[int] = set() + self._service_trait_identifier_cache: Dict[Tuple[object, ...], str] = {} + self._service_client_module_identifier_cache: Dict[ + Tuple[object, ...], str + ] = {} + self._service_server_module_identifier_cache: Dict[ + Tuple[object, ...], str + ] = {} + self._service_name_constant_identifier_cache: Dict[ + Tuple[object, ...], str + ] = {} + self._rpc_method_identifier_cache: Dict[ + Tuple[object, ...], Dict[Tuple[object, ...], str] + ] = {} + self._rpc_stream_type_identifier_cache: Dict[ + Tuple[object, ...], Dict[Tuple[object, ...], str] + ] = {} + self._rpc_path_constant_identifier_cache: Dict[ + Tuple[object, ...], Dict[Tuple[object, ...], str] + ] = {} + schema_id = id(self.schema) if schema_id in self._named_service_schema_ids: return @@ -265,12 +288,13 @@ def generate_grpc_method_signature( 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) + mode = streaming_mode(method) lines: List[str] = [] - if method.client_streaming: + if mode in (StreamingMode.CLIENT_STREAMING, StreamingMode.BIDIRECTIONAL): request_arg_type = f"::tonic::Request<::tonic::Streaming<{request_type}>>" else: request_arg_type = f"::tonic::Request<{request_type}>" - if method.server_streaming: + if mode in (StreamingMode.SERVER_STREAMING, StreamingMode.BIDIRECTIONAL): stream_type = self._rpc_stream_type_identifier_cache[service_key][ method_key ] diff --git a/compiler/fory_compiler/tests/test_service_codegen.py b/compiler/fory_compiler/tests/test_service_codegen.py index 2d63b2ecee..a430df80e9 100644 --- a/compiler/fory_compiler/tests/test_service_codegen.py +++ b/compiler/fory_compiler/tests/test_service_codegen.py @@ -18,6 +18,7 @@ """Codegen smoke tests for schemas that contain service definitions.""" from pathlib import Path +import re import shutil import subprocess from textwrap import dedent @@ -158,6 +159,7 @@ def test_unsupported_generators_no_services(): KotlinGenerator, JavaScriptGenerator, DartGenerator, + CppGenerator, ): continue options = GeneratorOptions(output_dir=Path("/tmp")) @@ -1983,6 +1985,9 @@ def test_grpc_flag_compiles_services(tmp_path: Path, capsys): assert output.count("demo_greeter_grpc.go") == 1 assert (lang_dirs["rust"] / "demo_greeter_service.rs").exists() assert (lang_dirs["rust"] / "demo_greeter_service_grpc.rs").exists() + assert (lang_dirs["cpp"] / "demo_greeter_service.h").exists() + assert (lang_dirs["cpp"] / "demo_greeter_service_grpc.h").exists() + assert (lang_dirs["cpp"] / "demo_greeter_service_grpc.cc").exists() assert (lang_dirs["csharp"] / "demo" / "greeter" / "Service.cs").exists() assert (lang_dirs["csharp"] / "demo" / "greeter" / "GreeterGrpc.cs").exists() assert (lang_dirs["scala"] / "demo" / "greeter" / "GreeterGrpc.scala").exists() @@ -3602,3 +3607,285 @@ def test_dart_grpc_reserved_methods(): msg = str(excinfo.value) assert "inherited Dart member" in msg assert f"Svc.{rpc_name} -> {emitted}" in msg + + +def test_cpp_grpc_companions_and_codec(): + schema = parse_fdl(_GREETER_WITH_SERVICE) + files = generate_service_files(schema, CppGenerator) + assert set(files) == { + "demo_greeter_service.h", + "demo_greeter_service_grpc.h", + "demo_greeter_service_grpc.cc", + } + + api = files["demo_greeter_service.h"] + assert "namespace demo::greeter::service {" in api + assert "class Greeter {" in api + assert "virtual ::grpc::Status SayHello(" in api + assert 'GreeterServiceName[] = "demo.greeter.Greeter"' in api + assert 'GreeterSayHelloPath[] = "/demo.greeter.Greeter/SayHello"' in api + + header = files["demo_greeter_service_grpc.h"] + assert "FORY_GENERATED_GRPC_SERIALIZATION_TRAITS_" in header + assert "class ForyGrpcSerializationTraits" in header + assert "SerializationTraits<::demo::greeter::HelloRequest, void>" in header + assert "SerializationTraits<::demo::greeter::HelloReply, void>" in header + assert "message.to_bytes()" in header + assert "Message::from_bytes(slice.begin(), slice.size())" in header + assert "namespace demo::greeter::service::grpc {" in header + assert "class GreeterStub final" in header + assert "class GreeterServiceGrpc final : public ::grpc::Service" in header + assert ( + "explicit GreeterServiceGrpc(::demo::greeter::service::Greeter* impl);" + in header + ) + assert "The caller owns impl; it must outlive this adapter and server." in header + + source = files["demo_greeter_service_grpc.cc"] + assert "namespace demo::greeter::service::grpc {" in source + assert "::demo::greeter::service::GreeterSayHelloPath" in source + assert "BlockingUnaryCall<" in source + assert "RpcMethod::NORMAL_RPC" in source + assert "RpcMethodHandler<" in source + + +def test_cpp_grpc_streaming_modes(): + schema = parse_fdl( + dedent( + """ + package demo.streaming; + + message Req {} + message Res {} + + service Streamer { + rpc Unary (Req) returns (Res); + rpc Upload (stream Req) returns (Res); + rpc Download (Req) returns (stream Res); + rpc Chat (stream Req) returns (stream Res); + } + """ + ) + ) + files = generate_service_files(schema, CppGenerator) + api = files["demo_streaming_service.h"] + header = files["demo_streaming_service_grpc.h"] + source = files["demo_streaming_service_grpc.cc"] + + assert "::grpc::ServerReader<::demo::streaming::Req>* reader" in api + assert "::grpc::ServerWriter<::demo::streaming::Res>* writer" in api + assert "::grpc::ServerReaderWriter<" in api + assert "::demo::streaming::Res, ::demo::streaming::Req>* stream" in api + + assert "::grpc::ClientWriter<::demo::streaming::Req>" in header + assert "::grpc::ClientReader<::demo::streaming::Res>" in header + assert "::grpc::ClientReaderWriter<" in header + assert "::demo::streaming::Req, ::demo::streaming::Res" in header + + assert "RpcMethod::NORMAL_RPC" in source + assert "RpcMethod::CLIENT_STREAMING" in source + assert "RpcMethod::SERVER_STREAMING" in source + assert "RpcMethod::BIDI_STREAMING" in source + assert "ClientWriterFactory<::demo::streaming::Req>::Create" in source + assert "ClientReaderFactory<::demo::streaming::Res>::Create" in source + assert "ClientReaderWriterFactory<" in source + assert "ClientStreamingHandler<" in source + assert "ServerStreamingHandler<" in source + assert "BidiStreamingHandler<" in source + + +def test_cpp_grpc_nested_union_and_default_package(): + schema = parse_fdl( + dedent( + """ + message Envelope { + message Request {} + union Reply { + string text = 1; + } + } + + service Nested { + rpc Call (Envelope.Request) returns (Envelope.Reply); + } + """ + ) + ) + files = generate_service_files(schema, CppGenerator) + assert set(files) == { + "generated_service.h", + "generated_service_grpc.h", + "generated_service_grpc.cc", + } + api = files["generated_service.h"] + header = files["generated_service_grpc.h"] + assert "namespace service {" in api + assert "namespace service::grpc {" in header + assert 'NestedServiceName[] = "Nested"' in api + assert 'NestedCallPath[] = "/Nested/Call"' in api + assert "const ::Envelope::Request* request" in api + assert "::Envelope::Reply* response" in api + assert "SerializationTraits<::Envelope::Request, void>" in header + assert "SerializationTraits<::Envelope::Reply, void>" in header + + +def test_cpp_grpc_imported_message_paths(tmp_path: Path): + common = tmp_path / "common.fdl" + common.write_text( + dedent( + """ + package demo.common; + + message Envelope { + message Request {} + union Reply { + string text = 1; + } + } + """ + ) + ) + service = tmp_path / "service.fdl" + service.write_text( + dedent( + """ + package demo.api; + import "common.fdl"; + + service Api { + rpc Call (demo.common.Envelope.Request) + returns (demo.common.Envelope.Reply); + } + """ + ) + ) + schema = resolve_imports(service) + files = generate_service_files(schema, CppGenerator) + assert set(files) == { + "demo_api_service.h", + "demo_api_service_grpc.h", + "demo_api_service_grpc.cc", + } + api = files["demo_api_service.h"] + header = files["demo_api_service_grpc.h"] + assert "const ::demo::common::Envelope::Request* request" in api + assert "::demo::common::Envelope::Reply* response" in api + assert "SerializationTraits<::demo::common::Envelope::Request, void>" in header + assert "SerializationTraits<::demo::common::Envelope::Reply, void>" in header + + +def test_cpp_grpc_shared_message_specialization_guard(tmp_path: Path): + common = tmp_path / "common.fdl" + common.write_text( + dedent( + """ + package demo.common; + message Shared {} + """ + ) + ) + headers = [] + for name in ("first", "second"): + service = tmp_path / f"{name}.fdl" + service.write_text( + dedent( + f""" + package demo.{name}; + import "common.fdl"; + service Api {{ + rpc Call (demo.common.Shared) returns (demo.common.Shared); + }} + """ + ) + ) + schema = resolve_imports(service) + files = generate_service_files(schema, CppGenerator) + headers.append(files[f"demo_{name}_service_grpc.h"]) + + guard_pattern = re.compile( + r"FORY_GENERATED_GRPC_TRAITS_DEMO_COMMON_SHARED_[0-9A-F]{12}_" + ) + first_guard = guard_pattern.search(headers[0]) + second_guard = guard_pattern.search(headers[1]) + assert first_guard is not None + assert second_guard is not None + assert first_guard.group(0) == second_guard.group(0) + assert headers[0].count("SerializationTraits<::demo::common::Shared, void>") == 1 + assert headers[1].count("SerializationTraits<::demo::common::Shared, void>") == 1 + + +def test_cpp_grpc_name_collisions(): + schema = parse_fdl( + dedent( + """ + package demo.collision; + message Greeter {} + message Req {} + message Res {} + service Greeter { + rpc Call (Req) returns (Res); + } + """ + ) + ) + files = generate_service_files(schema, CppGenerator) + assert "namespace demo::collision::service {" in files["demo_collision_service.h"] + assert "class Greeter {" in files["demo_collision_service.h"] + + service_schema = parse_fdl( + dedent( + """ + package demo.collision; + message Req {} + message Res {} + service Greeter { + rpc Call (Req) returns (Res); + } + service GreeterServiceName { + rpc Call (Req) returns (Res); + } + """ + ) + ) + with pytest.raises(ValueError, match="C\\+\\+ name collision"): + generate_service_files(service_schema, CppGenerator) + + grpc_namespace_schema = parse_fdl( + dedent( + """ + package demo.collision; + message Req {} + message Res {} + service grpc { + rpc Call (Req) returns (Res); + } + """ + ) + ) + with pytest.raises(ValueError, match="generated gRPC transport namespace"): + generate_service_files(grpc_namespace_schema, CppGenerator) + + method_schema = parse_fdl( + dedent( + """ + package demo.collision; + message Req {} + message Res {} + service Greeter { + rpc NewStub (Req) returns (Res); + } + """ + ) + ) + method_files = generate_service_files(method_schema, CppGenerator) + method_header = method_files["demo_collision_service_grpc.h"] + method_source = method_files["demo_collision_service_grpc.cc"] + assert " static ::std::unique_ptr NewStub(" in method_header + assert ( + " ::grpc::Status NewStub(\n ::grpc::ClientContext* context," + in method_header + ) + assert ( + "::grpc::Status GreeterStub::NewStub(\n ::grpc::ClientContext* context," + in method_source + ) diff --git a/docs/compiler/compiler-guide.md b/docs/compiler/compiler-guide.md index 184b48cc4c..4fb9ab1aeb 100644 --- a/docs/compiler/compiler-guide.md +++ b/docs/compiler/compiler-guide.md @@ -149,10 +149,10 @@ foryc user.fdl order.fdl product.fdl --output ./generated foryc compiler/examples/service.fdl --java_out=./generated/java --python_out=./generated/python --go_out=./generated/go --rust_out=./generated/rust --csharp_out=./generated/csharp --dart_out=./generated/dart --scala_out=./generated/scala --kotlin_out=./generated/kotlin --javascript_out=./generated/javascript ``` -**Generate Java, Python, Go, Rust, C#, Dart, Scala, Kotlin, and Node.js JavaScript gRPC service companions:** +**Generate Java, Python, Go, Rust, C++, C#, Dart, Scala, Kotlin, and Node.js JavaScript gRPC service companions:** ```bash -foryc compiler/examples/service.fdl --java_out=./generated/java --python_out=./generated/python --go_out=./generated/go --rust_out=./generated/rust --csharp_out=./generated/csharp --dart_out=./generated/dart --scala_out=./generated/scala --kotlin_out=./generated/kotlin --javascript_out=./generated/javascript --grpc +foryc compiler/examples/service.fdl --java_out=./generated/java --python_out=./generated/python --go_out=./generated/go --rust_out=./generated/rust --cpp_out=./generated/cpp --csharp_out=./generated/csharp --dart_out=./generated/dart --scala_out=./generated/scala --kotlin_out=./generated/kotlin --javascript_out=./generated/javascript --grpc ``` The generated gRPC service code uses Fory to serialize request and response diff --git a/docs/compiler/generated-code.md b/docs/compiler/generated-code.md index 68fddf234c..554291ad98 100644 --- a/docs/compiler/generated-code.md +++ b/docs/compiler/generated-code.md @@ -603,6 +603,8 @@ class Person final { void set_number(Arg&& arg, Args&&... args); fory::Result, fory::Error> to_bytes() const; + static fory::Result from_bytes( + const uint8_t* data, std::size_t size); static fory::Result from_bytes(const std::vector& data); }; @@ -714,6 +716,53 @@ auto bytes = person.to_bytes(); auto restored = addressbook::Person::from_bytes(bytes.value()); ``` +### gRPC Service Companions + +When a schema contains services and the compiler is run with `--grpc`, C++ +generation emits a service API header file and gRPC binding files. For a schema +header named `addressbook.h`, those files are `addressbook_service.h`, +`addressbook_service_grpc.h`, and `addressbook_service_grpc.cc`. + +The service API header file contains synchronous business interfaces and gRPC path +constants under `::::service`: + +```cpp +namespace addressbook::service { + +class AddressBookService { + public: + virtual ~AddressBookService() = default; + + virtual ::grpc::Status Lookup( + ::grpc::ServerContext* context, + const ::addressbook::Person* request, + ::addressbook::AddressBook* response) = 0; +}; + +inline constexpr char AddressBookServiceServiceName[] = + "addressbook.AddressBookService"; +inline constexpr char AddressBookServiceLookupPath[] = + "/addressbook.AddressBookService/Lookup"; + +} // namespace addressbook::service +``` + +The gRPC binding header file and source file contain Fory-backed codecs, generated +`grpc::SerializationTraits` specializations, and synchronous client/server +wrappers under `::::service::grpc`. They serialize each +request or response with the generated model type's `to_bytes` and `from_bytes` +helpers: + +```cpp +template <> +class SerializationTraits<::addressbook::Person, void> + : public ::fory::grpc::detail::ForyGrpcSerializationTraits< + ::addressbook::Person> {}; +``` + +Applications compiling the generated C++ service files must provide gRPC C++ +and link the generated `_service_grpc.cc` file with Fory serialization. + ## Go ### Output Layout diff --git a/docs/guide/cpp/grpc-support.md b/docs/guide/cpp/grpc-support.md new file mode 100644 index 0000000000..4df1963b09 --- /dev/null +++ b/docs/guide/cpp/grpc-support.md @@ -0,0 +1,277 @@ +--- +title: gRPC Support +sidebar_position: 8 +id: grpc_support +license: | + 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. +--- + +Fory can generate C++ gRPC service companions for schemas that define services. +The generated code uses gRPC C++ for transport and Fory for request and +response payload serialization. + +Use this mode when every RPC peer is generated from the same Fory IDL, protobuf +IDL, or FlatBuffers IDL and you want gRPC transport semantics with Fory payload +encoding. Use standard protobuf gRPC code generation when clients or tools must +consume protobuf message bytes directly. + +## Add Dependencies + +With Bazel, use the Fory label that is visible from your workspace. +The example below is for a project that consumes Fory as an external module named `fory`: + +```bazel +load("@rules_cc//cc:defs.bzl", "cc_library") + +cc_library( + name = "greeter_generated", + srcs = ["generated/demo_greeter_service_grpc.cc"], + hdrs = glob(["generated/*.h"]), + includes = ["generated"], + deps = [ + "@fory//cpp/fory/serialization:fory_serialization", + "@grpc//:grpc++", + ], +) +``` + +Inside the Fory repository, use `//cpp/fory/serialization:fory_serialization` +instead of `@fory//cpp/fory/serialization:fory_serialization`. + +With CMake, make the Fory C++ target visible first. For an installed Fory +package, use `find_package(Fory CONFIG REQUIRED)`, then add the generated source +and link both libraries explicitly: + +```cmake +find_package(Fory CONFIG REQUIRED) +find_package(gRPC CONFIG REQUIRED) + +add_library(greeter_generated + generated/demo_greeter_service_grpc.cc +) +target_compile_features(greeter_generated PUBLIC cxx_std_17) +target_include_directories(greeter_generated PUBLIC generated) +target_link_libraries(greeter_generated PUBLIC + fory::serialization + gRPC::grpc++ +) +``` + +If your project brings Fory in with `FetchContent` or `add_subdirectory`, call +that before linking `fory::serialization`. + +Use dependency versions that are compatible with the rest of your service +stack. + +## Define a Service + +Service definitions can come from Fory IDL, protobuf IDL, or FlatBuffers +`rpc_service` definitions. A Fory IDL service looks like this: + +```protobuf +package demo.greeter; + +message HelloRequest { + string name = 1; +} + +message HelloReply { + string reply = 1; +} + +service Greeter { + rpc SayHello (HelloRequest) returns (HelloReply); +} +``` + +Generate C++ model and gRPC companion code with `--grpc`: + +```bash +foryc service.fdl --cpp_out=./generated/cpp --grpc +``` + +For this schema, the C++ generator emits: + +| File | Purpose | +| ------------------------------ | -------------------------------------------------- | +| `demo_greeter.h` | Fory model types and registration helpers | +| `demo_greeter_service.h` | Synchronous service interface and path constants | +| `demo_greeter_service_grpc.h` | Synchronous client, server adapter, and Fory codec | +| `demo_greeter_service_grpc.cc` | Stub calls and server route implementations | + +Include the generated gRPC header from application code and compile +`demo_greeter_service_grpc.cc` once in your build target. The codec is generated +directly in the gRPC header; there is no separate Fory gRPC runtime source file. + +## Implement a Server + +Implement the generated synchronous interface and register the generated server +adapter with a normal gRPC C++ server. + +```cpp +#include "demo_greeter_service_grpc.h" + +#include +#include +#include + +class MyGreeter final : public demo::greeter::service::Greeter { + public: + ::grpc::Status SayHello(::grpc::ServerContext* context, + const ::demo::greeter::HelloRequest* request, + ::demo::greeter::HelloReply* response) override { + (void)context; + response->set_reply("Hello, " + request->name()); + return ::grpc::Status::OK; + } +}; + +MyGreeter implementation; +demo::greeter::service::grpc::GreeterServiceGrpc service(&implementation); +::grpc::ServerBuilder builder; +builder.AddListeningPort("0.0.0.0:50051", ::grpc::InsecureServerCredentials()); +builder.RegisterService(&service); +std::unique_ptr<::grpc::Server> server = builder.BuildAndStart(); +server->Wait(); +``` + +Generated request and response types are serialized by the generated service +code, so service implementations do not perform manual Fory registration. + +## Create a Client + +Use the generated synchronous client stub: + +```cpp +#include "demo_greeter_service_grpc.h" + +#include +#include +#include + +auto channel = + ::grpc::CreateChannel("localhost:50051", ::grpc::InsecureChannelCredentials()); +auto stub = demo::greeter::service::grpc::GreeterStub::NewStub(channel); + +demo::greeter::HelloRequest request; +request.set_name("Fory"); +demo::greeter::HelloReply response; +::grpc::ClientContext context; +::grpc::Status status = stub->SayHello(&context, request, &response); +if (status.ok()) { + std::cout << response.reply() << std::endl; +} +``` + +gRPC C++ still owns channel configuration, credentials, deadlines, metadata, +cancellation, retry policy, and transport lifecycle. + +## Streaming RPCs + +Fory service definitions can use unary, server-streaming, client-streaming, and +bidirectional streaming RPC shapes: + +```protobuf +service Greeter { + rpc SayHello (HelloRequest) returns (HelloReply); + rpc LotsOfReplies (HelloRequest) returns (stream HelloReply); + rpc LotsOfGreetings (stream HelloRequest) returns (HelloReply); + rpc Chat (stream HelloRequest) returns (stream HelloReply); +} +``` + +Generated C++ code follows synchronous gRPC C++ conventions: + +- Unary methods return `grpc::Status` and use request and response pointers. +- Server-streaming methods return `std::unique_ptr>` on + clients and receive `grpc::ServerWriter*` on servers. +- Client-streaming methods return `std::unique_ptr>` on + clients and receive `grpc::ServerReader*` on servers. +- Bidirectional methods return + `std::unique_ptr>` on clients and receive + `grpc::ServerReaderWriter*` on servers. +- The generated codec is used for every message frame, including streaming + frames. + +Use the generated method signatures as the source of truth for concrete request +and response types in your service implementation: + +```cpp +::grpc::Status LotsOfReplies( + ::grpc::ServerContext* context, + const ::demo::greeter::HelloRequest* request, + ::grpc::ServerWriter<::demo::greeter::HelloReply>* writer) override { + (void)context; + ::demo::greeter::HelloReply reply; + reply.set_reply("Hello, " + request->name()); + writer->Write(reply); + reply.set_reply("Welcome, " + request->name()); + writer->Write(reply); + return ::grpc::Status::OK; +} +``` + +Generated clients return standard gRPC C++ streaming helpers: + +```cpp +demo::greeter::HelloRequest request; +request.set_name("Fory"); + +::grpc::ClientContext context; +auto reader = stub->LotsOfReplies(&context, request); +demo::greeter::HelloReply reply; +while (reader->Read(&reply)) { + std::cout << reply.reply() << std::endl; +} +::grpc::Status status = reader->Finish(); +``` + +Complete client streams with `WritesDone()` and always call `Finish()` to obtain +the final status. + +The generated descriptors preserve the exact IDL service and method names for +the gRPC path. + +## gRPC Runtime Behavior + +The generated service companion only supplies Fory serialization and gRPC C++ +bindings. Operational behavior remains standard gRPC C++ behavior: + +- Deadlines and cancellations +- TLS and authentication +- Status codes and metadata +- Channel and server lifecycle +- Synchronous streaming backpressure + +## Troubleshooting + +### Missing gRPC C++ Headers or Symbols + +Add the gRPC C++ dependency shown above to the target that compiles the +generated service files, and compile the generated `_service_grpc.cc` file +exactly once. + +### `UNIMPLEMENTED` + +Confirm that the generated server adapter was registered with +`ServerBuilder::RegisterService(...)`, and that the client and server were +generated from the same package, service, and method names. + +### Protobuf Clients Cannot Decode the Service + +Fory gRPC companions do not use protobuf wire encoding for messages. Use a +Fory-generated client for Fory-generated services, or provide a separate +protobuf service endpoint for generic protobuf clients. \ No newline at end of file diff --git a/docs/guide/cpp/index.md b/docs/guide/cpp/index.md index 5f73481b61..def5a76573 100644 --- a/docs/guide/cpp/index.md +++ b/docs/guide/cpp/index.md @@ -282,3 +282,4 @@ std::thread t2([&]() { - [Supported Types](supported-types.md) - All supported types - [Custom Serializers](custom-serializers.md) - Extend serialization behavior - [Row Format](row-format.md) - Zero-copy row-based format +- [gRPC Support](grpc-support.md) - Fory payloads over gRPC C++ diff --git a/integration_tests/grpc_tests/cpp/BUILD b/integration_tests/grpc_tests/cpp/BUILD new file mode 100644 index 0000000000..703604df67 --- /dev/null +++ b/integration_tests/grpc_tests/cpp/BUILD @@ -0,0 +1,38 @@ +# 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. + +load("@rules_cc//cc:defs.bzl", "cc_binary", "cc_library") + +cc_library( + name = "generated_services", + srcs = glob(["generated/*_service_grpc.cc"]), + hdrs = glob(["generated/*.h"]), + includes = ["generated"], + deps = [ + "//cpp/fory/serialization:fory_serialization", + "@grpc//:grpc++", + ], +) + +cc_binary( + name = "grpc_interop", + srcs = ["interop/main.cc"], + deps = [ + ":generated_services", + "@grpc//:grpc++", + ], +) diff --git a/integration_tests/grpc_tests/cpp/interop/main.cc b/integration_tests/grpc_tests/cpp/interop/main.cc new file mode 100644 index 0000000000..303b9ff309 --- /dev/null +++ b/integration_tests/grpc_tests/cpp/interop/main.cc @@ -0,0 +1,753 @@ +// 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. + +#include "grpc_fbs_service_grpc.h" +#include "grpc_fdl_service_grpc.h" +#include "grpc_pb_service_grpc.h" + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +void CheckStatus(const std::string &method, const ::grpc::Status &status) { + if (!status.ok()) { + throw std::runtime_error(method + ": " + status.error_message()); + } +} + +template +void CheckEqual(const std::string &method, const T &actual, const T &expected) { + if (!(actual == expected)) { + throw std::runtime_error(method + ": response mismatch"); + } +} + +template +void CheckEqual(const std::string &method, const std::vector &actual, + const std::vector &expected) { + if (actual != expected) { + throw std::runtime_error(method + ": response stream mismatch"); + } +} + +std::string Join(const std::vector &values) { + std::string joined; + for (std::size_t index = 0; index < values.size(); ++index) { + if (index != 0) { + joined += "+"; + } + joined += values[index]; + } + return joined; +} + +template +Request MakeStringRequest(std::string id, std::int32_t count, + std::string body) { + Request request; + request.set_id(std::move(id)); + request.set_count(count); + request.set_payload(std::move(body)); + return request; +} + +template +Response MakeStringResponse(const Request &request, const std::string &tag, + std::int32_t offset) { + Response response; + response.set_id(tag + ":" + request.id()); + response.set_count(request.count() + offset); + response.set_payload(tag + ":" + request.payload()); + return response; +} + +template +Response AggregateStringRequests(const std::vector &requests) { + std::vector ids; + std::vector bodies; + std::int32_t count = 0; + for (const auto &request : requests) { + ids.push_back(request.id()); + bodies.push_back(request.payload()); + count += request.count(); + } + Response response; + response.set_id("client:" + Join(ids)); + response.set_count(count); + response.set_payload("client:" + Join(bodies)); + return response; +} + +grpc_pb::GrpcPbRequest MakePbRequest(std::string id, std::uint32_t count, + grpc_pb::GrpcPbRequest::Payload body) { + grpc_pb::GrpcPbRequest request; + request.set_id(std::move(id)); + request.set_count(count); + *request.mutable_payload() = std::move(body); + return request; +} + +grpc_pb::GrpcPbResponse MakePbResponse(const grpc_pb::GrpcPbRequest &request, + const std::string &tag, + std::uint32_t offset) { + grpc_pb::GrpcPbResponse response; + response.set_id(tag + ":" + request.id()); + response.set_count(request.count() + offset); + if (request.has_payload()) { + if (request.payload().is_text()) { + *response.mutable_payload() = grpc_pb::GrpcPbResponse::Payload::text( + tag + ":" + request.payload().text()); + } else { + *response.mutable_payload() = grpc_pb::GrpcPbResponse::Payload::number( + request.payload().number() + offset); + } + } + return response; +} + +grpc_pb::GrpcPbResponse +AggregatePbRequests(const std::vector &requests) { + std::vector ids; + std::uint32_t count = 0; + for (const auto &request : requests) { + ids.push_back(request.id()); + count += request.count(); + } + grpc_pb::GrpcPbResponse response; + response.set_id("client:" + Join(ids)); + response.set_count(count); + *response.mutable_payload() = + grpc_pb::GrpcPbResponse::Payload::text("client:" + Join(ids)); + return response; +} + +template +void ExerciseMessages(Stub *stub, const std::vector &requests, + ResponseFactory response_factory, + AggregateFactory aggregate_factory) { + const Request &first = requests.front(); + + Response unary_response; + ::grpc::ClientContext unary_context; + CheckStatus("UnaryMessage", + stub->UnaryMessage(&unary_context, first, &unary_response)); + CheckEqual("UnaryMessage", unary_response, + response_factory(first, "unary", 10)); + + std::vector server_responses; + ::grpc::ClientContext server_context; + auto reader = stub->ServerStreamMessage(&server_context, first); + Response server_response; + while (reader->Read(&server_response)) { + server_responses.push_back(server_response); + } + CheckStatus("ServerStreamMessage", reader->Finish()); + CheckEqual("ServerStreamMessage", server_responses, + std::vector{response_factory(first, "server-0", 0), + response_factory(first, "server-1", 1), + response_factory(first, "server-2", 2)}); + + Response client_response; + ::grpc::ClientContext client_context; + auto writer = stub->ClientStreamMessage(&client_context, &client_response); + for (const auto &request : requests) { + if (!writer->Write(request)) { + throw std::runtime_error("ClientStreamMessage: write failed"); + } + } + writer->WritesDone(); + CheckStatus("ClientStreamMessage", writer->Finish()); + CheckEqual("ClientStreamMessage", client_response, + aggregate_factory(requests)); + + std::vector bidi_responses; + ::grpc::ClientContext bidi_context; + auto stream = stub->BidiStreamMessage(&bidi_context); + for (const auto &request : requests) { + if (!stream->Write(request)) { + throw std::runtime_error("BidiStreamMessage: write failed"); + } + } + stream->WritesDone(); + Response bidi_response; + while (stream->Read(&bidi_response)) { + bidi_responses.push_back(bidi_response); + } + CheckStatus("BidiStreamMessage", stream->Finish()); + std::vector expected_bidi; + for (std::size_t index = 0; index < requests.size(); ++index) { + expected_bidi.push_back(response_factory( + requests[index], "bidi-" + std::to_string(index), index)); + } + CheckEqual("BidiStreamMessage", bidi_responses, expected_bidi); +} + +template +void ExerciseUnions(Stub *stub, const std::vector &requests, + RequestUnionFactory request_union_factory, + ResponseUnionFactory response_union_factory, + AggregateFactory aggregate_factory) { + std::vector unions; + for (const auto &request : requests) { + unions.push_back(request_union_factory(request)); + } + + Union unary_response; + ::grpc::ClientContext unary_context; + CheckStatus("UnaryUnion", stub->UnaryUnion(&unary_context, unions.front(), + &unary_response)); + CheckEqual("UnaryUnion", unary_response, + response_union_factory(requests.front(), "unary", 10)); + + std::vector server_responses; + ::grpc::ClientContext server_context; + auto reader = stub->ServerStreamUnion(&server_context, unions.front()); + Union server_response; + while (reader->Read(&server_response)) { + server_responses.push_back(server_response); + } + CheckStatus("ServerStreamUnion", reader->Finish()); + CheckEqual("ServerStreamUnion", server_responses, + std::vector{ + response_union_factory(requests.front(), "server-0", 0), + response_union_factory(requests.front(), "server-1", 1), + response_union_factory(requests.front(), "server-2", 2)}); + + Union client_response; + ::grpc::ClientContext client_context; + auto writer = stub->ClientStreamUnion(&client_context, &client_response); + for (const auto &value : unions) { + if (!writer->Write(value)) { + throw std::runtime_error("ClientStreamUnion: write failed"); + } + } + writer->WritesDone(); + CheckStatus("ClientStreamUnion", writer->Finish()); + CheckEqual("ClientStreamUnion", client_response, aggregate_factory(requests)); + + std::vector bidi_responses; + ::grpc::ClientContext bidi_context; + auto stream = stub->BidiStreamUnion(&bidi_context); + for (const auto &value : unions) { + if (!stream->Write(value)) { + throw std::runtime_error("BidiStreamUnion: write failed"); + } + } + stream->WritesDone(); + Union bidi_response; + while (stream->Read(&bidi_response)) { + bidi_responses.push_back(bidi_response); + } + CheckStatus("BidiStreamUnion", stream->Finish()); + std::vector expected_bidi; + for (std::size_t index = 0; index < requests.size(); ++index) { + expected_bidi.push_back(response_union_factory( + requests[index], "bidi-" + std::to_string(index), index)); + } + CheckEqual("BidiStreamUnion", bidi_responses, expected_bidi); +} + +template +::grpc::Status ReadAndAggregate(Reader *reader, Response *response) { + std::vector requests; + Request request; + while (reader->Read(&request)) { + requests.push_back(request); + } + *response = AggregateStringRequests(requests); + return ::grpc::Status::OK; +} + +template +::grpc::Status EchoStringStream(Stream *stream) { + Request request; + std::int32_t index = 0; + while (stream->Read(&request)) { + if (!stream->Write(MakeStringResponse( + request, "bidi-" + std::to_string(index), index))) { + return ::grpc::Status(::grpc::StatusCode::UNKNOWN, + "response write failed"); + } + ++index; + } + return ::grpc::Status::OK; +} + +class FdlService final : public grpc_fdl::service::FdlGrpcService { +public: + ::grpc::Status UnaryMessage(::grpc::ServerContext *, + const grpc_fdl::GrpcFdlRequest *request, + grpc_fdl::GrpcFdlResponse *response) override { + *response = + MakeStringResponse(*request, "unary", 10); + return ::grpc::Status::OK; + } + + ::grpc::Status ServerStreamMessage( + ::grpc::ServerContext *, const grpc_fdl::GrpcFdlRequest *request, + ::grpc::ServerWriter *writer) override { + for (std::int32_t index = 0; index < 3; ++index) { + if (!writer->Write(MakeStringResponse( + *request, "server-" + std::to_string(index), index))) { + return ::grpc::Status(::grpc::StatusCode::UNKNOWN, + "response write failed"); + } + } + return ::grpc::Status::OK; + } + + ::grpc::Status + ClientStreamMessage(::grpc::ServerContext *, + ::grpc::ServerReader *reader, + grpc_fdl::GrpcFdlResponse *response) override { + return ReadAndAggregate(reader, response); + } + + ::grpc::Status BidiStreamMessage( + ::grpc::ServerContext *, + ::grpc::ServerReaderWriter *stream) override { + return EchoStringStream(stream); + } + + ::grpc::Status UnaryUnion(::grpc::ServerContext *, + const grpc_fdl::GrpcFdlUnion *request, + grpc_fdl::GrpcFdlUnion *response) override { + const auto *value = request->as_request(); + if (value == nullptr) { + return InvalidUnion(); + } + *response = grpc_fdl::GrpcFdlUnion::response( + MakeStringResponse(*value, "unary", 10)); + return ::grpc::Status::OK; + } + + ::grpc::Status ServerStreamUnion( + ::grpc::ServerContext *, const grpc_fdl::GrpcFdlUnion *request, + ::grpc::ServerWriter *writer) override { + const auto *value = request->as_request(); + if (value == nullptr) { + return InvalidUnion(); + } + for (std::int32_t index = 0; index < 3; ++index) { + if (!writer->Write(grpc_fdl::GrpcFdlUnion::response( + MakeStringResponse( + *value, "server-" + std::to_string(index), index)))) { + return ::grpc::Status(::grpc::StatusCode::UNKNOWN, + "response write failed"); + } + } + return ::grpc::Status::OK; + } + + ::grpc::Status + ClientStreamUnion(::grpc::ServerContext *, + ::grpc::ServerReader *reader, + grpc_fdl::GrpcFdlUnion *response) override { + std::vector requests; + grpc_fdl::GrpcFdlUnion value; + while (reader->Read(&value)) { + const auto *request = value.as_request(); + if (request == nullptr) { + return InvalidUnion(); + } + requests.push_back(*request); + } + *response = grpc_fdl::GrpcFdlUnion::response( + AggregateStringRequests(requests)); + return ::grpc::Status::OK; + } + + ::grpc::Status BidiStreamUnion( + ::grpc::ServerContext *, + ::grpc::ServerReaderWriter + *stream) override { + grpc_fdl::GrpcFdlUnion value; + std::int32_t index = 0; + while (stream->Read(&value)) { + const auto *request = value.as_request(); + if (request == nullptr) { + return InvalidUnion(); + } + if (!stream->Write(grpc_fdl::GrpcFdlUnion::response( + MakeStringResponse( + *request, "bidi-" + std::to_string(index), index)))) { + return ::grpc::Status(::grpc::StatusCode::UNKNOWN, + "response write failed"); + } + ++index; + } + return ::grpc::Status::OK; + } + +private: + static ::grpc::Status InvalidUnion() { + return ::grpc::Status(::grpc::StatusCode::INVALID_ARGUMENT, + "expected GrpcFdlUnion request"); + } +}; + +class FbsService final : public grpc_fbs::service::FbsGrpcService { +public: + ::grpc::Status UnaryMessage(::grpc::ServerContext *, + const grpc_fbs::GrpcFbsRequest *request, + grpc_fbs::GrpcFbsResponse *response) override { + *response = + MakeStringResponse(*request, "unary", 10); + return ::grpc::Status::OK; + } + + ::grpc::Status ServerStreamMessage( + ::grpc::ServerContext *, const grpc_fbs::GrpcFbsRequest *request, + ::grpc::ServerWriter *writer) override { + for (std::int32_t index = 0; index < 3; ++index) { + if (!writer->Write(MakeStringResponse( + *request, "server-" + std::to_string(index), index))) { + return ::grpc::Status(::grpc::StatusCode::UNKNOWN, + "response write failed"); + } + } + return ::grpc::Status::OK; + } + + ::grpc::Status + ClientStreamMessage(::grpc::ServerContext *, + ::grpc::ServerReader *reader, + grpc_fbs::GrpcFbsResponse *response) override { + return ReadAndAggregate(reader, response); + } + + ::grpc::Status BidiStreamMessage( + ::grpc::ServerContext *, + ::grpc::ServerReaderWriter *stream) override { + return EchoStringStream(stream); + } + + ::grpc::Status UnaryUnion(::grpc::ServerContext *, + const grpc_fbs::GrpcFbsUnion *request, + grpc_fbs::GrpcFbsUnion *response) override { + const auto *value = request->as_grpc_fbs_request(); + if (value == nullptr) { + return InvalidUnion(); + } + *response = grpc_fbs::GrpcFbsUnion::grpc_fbs_response( + MakeStringResponse(*value, "unary", 10)); + return ::grpc::Status::OK; + } + + ::grpc::Status ServerStreamUnion( + ::grpc::ServerContext *, const grpc_fbs::GrpcFbsUnion *request, + ::grpc::ServerWriter *writer) override { + const auto *value = request->as_grpc_fbs_request(); + if (value == nullptr) { + return InvalidUnion(); + } + for (std::int32_t index = 0; index < 3; ++index) { + if (!writer->Write(grpc_fbs::GrpcFbsUnion::grpc_fbs_response( + MakeStringResponse( + *value, "server-" + std::to_string(index), index)))) { + return ::grpc::Status(::grpc::StatusCode::UNKNOWN, + "response write failed"); + } + } + return ::grpc::Status::OK; + } + + ::grpc::Status + ClientStreamUnion(::grpc::ServerContext *, + ::grpc::ServerReader *reader, + grpc_fbs::GrpcFbsUnion *response) override { + std::vector requests; + grpc_fbs::GrpcFbsUnion value; + while (reader->Read(&value)) { + const auto *request = value.as_grpc_fbs_request(); + if (request == nullptr) { + return InvalidUnion(); + } + requests.push_back(*request); + } + *response = grpc_fbs::GrpcFbsUnion::grpc_fbs_response( + AggregateStringRequests(requests)); + return ::grpc::Status::OK; + } + + ::grpc::Status BidiStreamUnion( + ::grpc::ServerContext *, + ::grpc::ServerReaderWriter + *stream) override { + grpc_fbs::GrpcFbsUnion value; + std::int32_t index = 0; + while (stream->Read(&value)) { + const auto *request = value.as_grpc_fbs_request(); + if (request == nullptr) { + return InvalidUnion(); + } + if (!stream->Write(grpc_fbs::GrpcFbsUnion::grpc_fbs_response( + MakeStringResponse( + *request, "bidi-" + std::to_string(index), index)))) { + return ::grpc::Status(::grpc::StatusCode::UNKNOWN, + "response write failed"); + } + ++index; + } + return ::grpc::Status::OK; + } + +private: + static ::grpc::Status InvalidUnion() { + return ::grpc::Status(::grpc::StatusCode::INVALID_ARGUMENT, + "expected GrpcFbsUnion request"); + } +}; + +class PbService final : public grpc_pb::service::PbGrpcService { +public: + ::grpc::Status UnaryMessage(::grpc::ServerContext *, + const grpc_pb::GrpcPbRequest *request, + grpc_pb::GrpcPbResponse *response) override { + *response = MakePbResponse(*request, "unary", 10); + return ::grpc::Status::OK; + } + + ::grpc::Status ServerStreamMessage( + ::grpc::ServerContext *, const grpc_pb::GrpcPbRequest *request, + ::grpc::ServerWriter *writer) override { + for (std::uint32_t index = 0; index < 3; ++index) { + if (!writer->Write(MakePbResponse( + *request, "server-" + std::to_string(index), index))) { + return ::grpc::Status(::grpc::StatusCode::UNKNOWN, + "response write failed"); + } + } + return ::grpc::Status::OK; + } + + ::grpc::Status + ClientStreamMessage(::grpc::ServerContext *, + ::grpc::ServerReader *reader, + grpc_pb::GrpcPbResponse *response) override { + std::vector requests; + grpc_pb::GrpcPbRequest request; + while (reader->Read(&request)) { + requests.push_back(request); + } + *response = AggregatePbRequests(requests); + return ::grpc::Status::OK; + } + + ::grpc::Status BidiStreamMessage( + ::grpc::ServerContext *, + ::grpc::ServerReaderWriter *stream) override { + grpc_pb::GrpcPbRequest request; + std::uint32_t index = 0; + while (stream->Read(&request)) { + if (!stream->Write(MakePbResponse( + request, "bidi-" + std::to_string(index), index))) { + return ::grpc::Status(::grpc::StatusCode::UNKNOWN, + "response write failed"); + } + ++index; + } + return ::grpc::Status::OK; + } +}; + +void ExerciseFdl(const std::shared_ptr<::grpc::Channel> &channel) { + auto stub = grpc_fdl::service::grpc::FdlGrpcServiceStub::NewStub(channel); + const std::vector requests{ + MakeStringRequest("fdl-a", 1, "alpha"), + MakeStringRequest("fdl-b", 2, "beta")}; + ExerciseMessages( + stub.get(), requests, + [](const auto &request, const auto &tag, auto offset) { + return MakeStringResponse(request, tag, + offset); + }, + [](const auto &values) { + return AggregateStringRequests(values); + }); + + const std::vector union_requests{ + MakeStringRequest("fdl-u-a", 3, "union-alpha"), + MakeStringRequest("fdl-u-b", 4, "union-beta")}; + ExerciseUnions( + stub.get(), union_requests, + [](const auto &request) { + return grpc_fdl::GrpcFdlUnion::request(request); + }, + [](const auto &request, const auto &tag, auto offset) { + return grpc_fdl::GrpcFdlUnion::response( + MakeStringResponse(request, tag, + offset)); + }, + [](const auto &values) { + return grpc_fdl::GrpcFdlUnion::response( + AggregateStringRequests(values)); + }); +} + +void ExerciseFbs(const std::shared_ptr<::grpc::Channel> &channel) { + auto stub = grpc_fbs::service::grpc::FbsGrpcServiceStub::NewStub(channel); + const std::vector requests{ + MakeStringRequest("fbs-a", 5, "alpha"), + MakeStringRequest("fbs-b", 6, "beta")}; + ExerciseMessages( + stub.get(), requests, + [](const auto &request, const auto &tag, auto offset) { + return MakeStringResponse(request, tag, + offset); + }, + [](const auto &values) { + return AggregateStringRequests(values); + }); + + const std::vector union_requests{ + MakeStringRequest("fbs-u-a", 7, "union-alpha"), + MakeStringRequest("fbs-u-b", 8, "union-beta")}; + ExerciseUnions( + stub.get(), union_requests, + [](const auto &request) { + return grpc_fbs::GrpcFbsUnion::grpc_fbs_request(request); + }, + [](const auto &request, const auto &tag, auto offset) { + return grpc_fbs::GrpcFbsUnion::grpc_fbs_response( + MakeStringResponse(request, tag, + offset)); + }, + [](const auto &values) { + return grpc_fbs::GrpcFbsUnion::grpc_fbs_response( + AggregateStringRequests(values)); + }); +} + +void ExercisePb(const std::shared_ptr<::grpc::Channel> &channel) { + auto stub = grpc_pb::service::grpc::PbGrpcServiceStub::NewStub(channel); + const std::vector requests{ + MakePbRequest("pb-a", 9, grpc_pb::GrpcPbRequest::Payload::text("alpha")), + MakePbRequest("pb-b", 10, grpc_pb::GrpcPbRequest::Payload::number(42))}; + ExerciseMessages( + stub.get(), requests, + [](const auto &request, const auto &tag, auto offset) { + return MakePbResponse(request, tag, offset); + }, + [](const auto &values) { return AggregatePbRequests(values); }); +} + +std::string RequiredArgument(const std::vector &args, + const std::string &name) { + for (std::size_t index = 0; index < args.size(); ++index) { + if (args[index] == name) { + if (index + 1 == args.size()) { + throw std::invalid_argument("missing value for " + name); + } + return args[index + 1]; + } + } + throw std::invalid_argument("missing required argument " + name); +} + +void RunClient(const std::string &target) { + auto channel = + ::grpc::CreateChannel(target, ::grpc::InsecureChannelCredentials()); + if (!channel->WaitForConnected(std::chrono::system_clock::now() + + std::chrono::seconds(30))) { + throw std::runtime_error("timed out connecting to " + target); + } + ExerciseFdl(channel); + ExerciseFbs(channel); + ExercisePb(channel); +} + +void RunServer(const std::string &port_file) { + FdlService fdl_impl; + FbsService fbs_impl; + PbService pb_impl; + grpc_fdl::service::grpc::FdlGrpcServiceServiceGrpc fdl_service(&fdl_impl); + grpc_fbs::service::grpc::FbsGrpcServiceServiceGrpc fbs_service(&fbs_impl); + grpc_pb::service::grpc::PbGrpcServiceServiceGrpc pb_service(&pb_impl); + + ::grpc::ServerBuilder builder; + int port = 0; + builder.AddListeningPort("127.0.0.1:0", ::grpc::InsecureServerCredentials(), + &port); + builder.RegisterService(&fdl_service); + builder.RegisterService(&fbs_service); + builder.RegisterService(&pb_service); + std::unique_ptr<::grpc::Server> server = builder.BuildAndStart(); + if (server == nullptr || port == 0) { + throw std::runtime_error("failed to start gRPC server"); + } + + std::ofstream output(port_file); + output << port; + output.close(); + if (!output) { + throw std::runtime_error("failed to write port file " + port_file); + } + server->Wait(); +} + +} // namespace + +int main(int argc, char **argv) { + try { + const std::vector args(argv + 1, argv + argc); + if (args.empty()) { + throw std::invalid_argument( + "usage: grpc_interop client --target HOST:PORT | server " + "--port-file PATH"); + } + if (args.front() == "client") { + RunClient(RequiredArgument(args, "--target")); + return 0; + } + if (args.front() == "server") { + RunServer(RequiredArgument(args, "--port-file")); + return 0; + } + throw std::invalid_argument("unknown mode " + args.front()); + } catch (const std::exception &error) { + std::cerr << error.what() << std::endl; + return 1; + } +} diff --git a/integration_tests/grpc_tests/generate_grpc.py b/integration_tests/grpc_tests/generate_grpc.py index f2950c1fe5..931fe9b285 100644 --- a/integration_tests/grpc_tests/generate_grpc.py +++ b/integration_tests/grpc_tests/generate_grpc.py @@ -36,6 +36,7 @@ "python_sync": TEST_DIR / "python/grpc_sync_tests/generated", "go": TEST_DIR / "go/generated", "rust": TEST_DIR / "rust/generated/src", + "cpp": TEST_DIR / "cpp/generated", "csharp": TEST_DIR / "csharp/generated", "kotlin": TEST_DIR / "kotlin/src/main/kotlin/generated", "dart": TEST_DIR / "dart/lib/generated", @@ -80,6 +81,7 @@ def main() -> int: f"--python_out={OUTPUTS['python']}", f"--go_out={go_pkg_out}", f"--rust_out={OUTPUTS['rust']}", + f"--cpp_out={OUTPUTS['cpp']}", f"--csharp_out={OUTPUTS['csharp']}", f"--kotlin_out={OUTPUTS['kotlin']}", f"--dart_out={OUTPUTS['dart']}", diff --git a/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/CppGrpcTest.java b/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/CppGrpcTest.java new file mode 100644 index 0000000000..4797fbabe5 --- /dev/null +++ b/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/CppGrpcTest.java @@ -0,0 +1,67 @@ +/* + * 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. + */ + +package org.apache.fory.grpc_tests; + +import io.grpc.Server; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.testng.annotations.Test; + +public class CppGrpcTest extends GrpcTestBase { + + @Test + public void testJavaServerCppClient() throws Exception { + Server server = startJavaAllSchemasServer(); + try { + runPeer("cpp-grpc-client", cppCommand("client", "--target", + "127.0.0.1:" + server.getPort())); + } finally { + server.shutdownNow(); + server.awaitTermination(10, TimeUnit.SECONDS); + } + } + + @Test + public void testCppServerJavaClient() throws Exception { + exercisePeerServer("cpp-grpc", "C++", "fory-grpc-cpp-", + cppCommand("server"), this::exerciseAllSchemas); + } + + private PeerCommand cppCommand(String... args) { + Path cppRoot = grpcRoot().resolve("cpp"); + Path binary = repoRoot() + .resolve("bazel-bin") + .resolve("integration_tests") + .resolve("grpc_tests") + .resolve("cpp") + .resolve("grpc_interop"); + List command = new ArrayList<>(); + command.add(binary.toString()); + command.addAll(Arrays.asList(args)); + PeerCommand peerCommand = newPeerCommand(cppRoot, command); + putEnv(peerCommand, "ENABLE_FORY_DEBUG_OUTPUT", "1"); + setLocalhostNoProxy(peerCommand); + clearProxyEnv(peerCommand); + return peerCommand; + } +} diff --git a/integration_tests/grpc_tests/run_tests.sh b/integration_tests/grpc_tests/run_tests.sh index 784f2f05ee..47b7bbc7a4 100755 --- a/integration_tests/grpc_tests/run_tests.sh +++ b/integration_tests/grpc_tests/run_tests.sh @@ -21,7 +21,7 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" ROOT_DIR="$(cd "${SCRIPT_DIR}/../.." && pwd)" -TEST_CLASSES="${1:-PythonAsyncGrpcTest,PythonSyncGrpcTest,RustGrpcTest,GoGrpcTest,KotlinGrpcTest,DartGrpcTest}" +TEST_CLASSES="${1:-PythonAsyncGrpcTest,PythonSyncGrpcTest,RustGrpcTest,GoGrpcTest,CppGrpcTest,KotlinGrpcTest,DartGrpcTest}" has_test_class() { [[ ",${TEST_CLASSES}," == *",$1,"* ]] @@ -41,6 +41,12 @@ fi if has_test_class "RustGrpcTest"; then cargo build --manifest-path "${SCRIPT_DIR}/rust/Cargo.toml" --workspace --quiet fi +if has_test_class "CppGrpcTest"; then + ( + cd "${SCRIPT_DIR}/cpp" + bazel build //integration_tests/grpc_tests/cpp:grpc_interop --config=x86_64 + ) +fi if has_test_class "KotlinGrpcTest"; then cd "${SCRIPT_DIR}/kotlin" mvn --no-transfer-progress -DskipTests package