You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fory.apache.org by ch...@apache.org on 2026/06/09 05:21:10 UTC
(fory) branch main updated: feat(go): implement grpc stub generation (#3698)
This is an automated email from the ASF dual-hosted git repository.
chaokunyang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fory.git
The following commit(s) were added to refs/heads/main by this push:
new 7ec4a175b feat(go): implement grpc stub generation (#3698)
7ec4a175b is described below
commit 7ec4a175bce944de4e9ef2e85f2065f4250dee42
Author: Ayush Kumar <ay...@gmail.com>
AuthorDate: Tue Jun 9 10:51:05 2026 +0530
feat(go): implement grpc stub generation (#3698)
## Why?
Go implementation only provides type generation and misses the gRPC
service stub generations for now. this PR adds it.
## Related issues
Closes #3274
## What it does ?
### Implemented till now -
#### `ImportTracker` class in `base.py`
1. which returns the foreign imports defined in a RpcMethod inside a
schema file's service interface.
#### `generate_services()`
1. which returns a service generated file
2. License header, package header using `get_package_name()`
3. the imports block of go using `_build_import_block(tracker)`
`_build_import_block(tracker)` function returns the import block with
fixed grpc, context imports which are required by every single generated
file for each service. and a foreign type imports statement returned by
the `tracker.go_imports()`. at end after all the service code generation
is done, this function inserts the import block to the saved placeholder
index.
- [ ] Substantial AI assistance was used in this PR: `yes` / `no`
- [ ] If `yes`, I included a completed [AI Contribution
Checklist](https://github.com/apache/fory/blob/main/AI_POLICY.md#9-contributor-checklist-for-ai-assisted-prs)
in this PR description and the required `AI Usage Disclosure`.
- [ ] If `yes`, my PR description includes the required `ai_review`
summary and screenshot evidence of the final clean AI review results
from both fresh reviewers on the current PR diff or current HEAD after
the latest code changes.
## Does this PR introduce any user-facing change?
- [ ] Does this PR introduce any public API change?
- [ ] Does this PR introduce any binary protocol compatibility change?
## Benchmark
---
compiler/fory_compiler/generators/go.py | 7 +-
compiler/fory_compiler/generators/services/base.py | 54 ++
compiler/fory_compiler/generators/services/go.py | 720 +++++++++++++++++++++
.../fory_compiler/tests/test_service_codegen.py | 26 +-
integration_tests/grpc_tests/generate_grpc.py | 8 +
integration_tests/grpc_tests/go/go.mod | 35 +
integration_tests/grpc_tests/go/go.sum | 24 +
integration_tests/grpc_tests/go/main.go | 299 +++++++++
.../apache/fory/grpc_tests/GrpcInteropTest.java | 94 +++
integration_tests/grpc_tests/run_tests.sh | 4 +-
10 files changed, 1268 insertions(+), 3 deletions(-)
diff --git a/compiler/fory_compiler/generators/go.py b/compiler/fory_compiler/generators/go.py
index b25e63c79..c4cc57e73 100644
--- a/compiler/fory_compiler/generators/go.py
+++ b/compiler/fory_compiler/generators/go.py
@@ -21,6 +21,7 @@ from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple, Union as TypingUnion
from fory_compiler.generators.base import BaseGenerator, GeneratedFile
+from fory_compiler.generators.services.go import GoServiceGeneratorMixin
from fory_compiler.frontend.utils import parse_idl_file
from fory_compiler.ir.ast import (
Message,
@@ -38,7 +39,7 @@ from fory_compiler.ir.ast import (
from fory_compiler.ir.types import PrimitiveKind
-class GoGenerator(BaseGenerator):
+class GoGenerator(GoServiceGeneratorMixin, BaseGenerator):
"""Generates Go structs with fory tags."""
language_name = "go"
@@ -206,6 +207,10 @@ class GoGenerator(BaseGenerator):
# Generate a single Go file with all types
files.append(self.generate_file())
+ # Generate gRPC service stubs if requested
+ if self.options.grpc:
+ files.extend(self.generate_services())
+
return files
def get_package_name(self) -> str:
diff --git a/compiler/fory_compiler/generators/services/base.py b/compiler/fory_compiler/generators/services/base.py
new file mode 100644
index 000000000..2bc2bcfb2
--- /dev/null
+++ b/compiler/fory_compiler/generators/services/base.py
@@ -0,0 +1,54 @@
+# 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.
+
+"Shared utilities for gRPC service stub generators."
+
+from enum import Enum
+from typing import List, Dict
+from fory_compiler.ir.ast import RpcMethod
+
+
+class StreamingMode(Enum):
+ UNARY = 1
+ CLIENT_STREAMING = 2
+ SERVER_STREAMING = 3
+ BIDIRECTIONAL = 4
+
+
+def streaming_mode(method: RpcMethod) -> StreamingMode:
+ """Defines the type of rpc streaming patterns."""
+ if not method.client_streaming and not method.server_streaming:
+ return StreamingMode.UNARY
+ elif method.client_streaming and not method.server_streaming:
+ return StreamingMode.CLIENT_STREAMING
+ elif not method.client_streaming and method.server_streaming:
+ return StreamingMode.SERVER_STREAMING
+ else:
+ return StreamingMode.BIDIRECTIONAL
+
+
+class ImportTracker:
+ """Accumulates cross-package Go imports for generated service stubs."""
+
+ def __init__(self):
+ self._imports: Dict[str, str] = {}
+
+ def add(self, alias: str, import_path: str) -> None:
+ self._imports[alias] = import_path
+
+ def go_imports(self) -> List[str]:
+ return sorted(self._imports.values())
diff --git a/compiler/fory_compiler/generators/services/go.py b/compiler/fory_compiler/generators/services/go.py
new file mode 100644
index 000000000..399dbe685
--- /dev/null
+++ b/compiler/fory_compiler/generators/services/go.py
@@ -0,0 +1,720 @@
+# 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.
+
+"""Go gRPC service code generator."""
+
+from typing import List
+from fory_compiler.generators.services.base import (
+ ImportTracker,
+ StreamingMode,
+ streaming_mode,
+)
+from fory_compiler.generators.base import GeneratedFile
+from fory_compiler.ir.ast import Service, NamedType
+
+
+class GoServiceGeneratorMixin:
+ """Generates Go gRPC service stubs."""
+
+ def generate_services(self) -> List[GeneratedFile]:
+ local_services = [
+ s for s in self.schema.services if not self.is_imported_type(s)
+ ]
+ if not local_services:
+ return []
+ return [self._generate_grpc_file(local_services)]
+
+ def _generate_grpc_file(self, services: List[Service]) -> GeneratedFile:
+ """Generate one _grpc.go file containing all services in the schema."""
+ lines: List[str] = []
+ tracker = ImportTracker()
+
+ # License header
+ lines.append(self.get_license_header("//"))
+ lines.append("")
+
+ # Package declaration
+ lines.append(f"package {self.get_package_name()}")
+ lines.append("")
+
+ # Save placeholder index so imports can be inserted after all code is
+ # generated (imports are collected lazily via tracker as types are resolved).
+ import_placeholder_index = len(lines)
+
+ # CodecV2 is emitted once per file, shared by all services in the schema.
+ lines.extend(self._generate_codec())
+
+ for service in services:
+ lines.extend(self._generate_client_interface(service, tracker))
+ lines.extend(self._generate_client_struct(service))
+ lines.extend(self._generate_new_client(service))
+ lines.extend(self._generate_client_methods(service, tracker))
+ lines.extend(self._generate_stream_types(service, tracker))
+ lines.extend(self._generate_server_interface(service, tracker))
+ lines.extend(self._generate_unimplemented_server(service, tracker))
+ lines.extend(self._generate_server_stream_types(service, tracker))
+ lines.extend(self._generate_service_desc(service, tracker))
+ lines.extend(self._generate_register_server(service))
+
+ import_lines = self._build_import_block(tracker)
+ for i, line in enumerate(import_lines):
+ lines.insert(import_placeholder_index + i, line)
+
+ return GeneratedFile(
+ path=f"{self.get_file_name()}_grpc.go", content="\n".join(lines)
+ )
+
+ def _build_import_block(self, tracker: ImportTracker) -> List[str]:
+ imports = [
+ '"context"',
+ '"google.golang.org/grpc"',
+ '"google.golang.org/grpc/codes"',
+ '"google.golang.org/grpc/mem"',
+ '"google.golang.org/grpc/status"',
+ '"github.com/apache/fory/go/fory"',
+ ]
+
+ for alias, path in tracker._imports.items():
+ imports.append(f'{alias} "{path}"')
+
+ sorted_imports = sorted(set(imports))
+
+ lines = ["import ("]
+ for imp in sorted_imports:
+ lines.append(f"\t{imp}")
+ lines.append(")")
+ lines.append("")
+
+ return lines
+
+ def _resolve_go_type(self, named_type: NamedType, tracker: ImportTracker) -> str:
+ type_ref = self.schema.resolve_type_name(named_type.name)
+ type_def = self.schema.get_type(type_ref)
+ if type_def is not None and self.is_imported_type(type_def):
+ info = self._import_info_for_type(type_def)
+ if info:
+ alias, import_path, _ = info
+ tracker.add(alias, import_path)
+ return f"*{alias}.{type_ref}"
+ return f"*{type_ref}"
+
+ def _generate_client_interface(
+ self, service: Service, tracker: ImportTracker
+ ) -> List[str]:
+ lines: List[str] = []
+ lines.append(
+ f"// {service.name}Client is the client API for {service.name} service."
+ )
+ lines.append(f"type {service.name}Client interface {{")
+ for method in service.methods:
+ req_type = self._resolve_go_type(method.request_type, tracker)
+ res_type = self._resolve_go_type(method.response_type, tracker)
+ mode = streaming_mode(method)
+ if mode is StreamingMode.UNARY:
+ signature = f"(ctx context.Context, in {req_type}, opts ...grpc.CallOption) ({res_type}, error)"
+ lines.append(f"\t{self.to_pascal_case(method.name)}{signature}")
+ elif mode is StreamingMode.SERVER_STREAMING:
+ signature = f"(ctx context.Context, in {req_type}, opts ...grpc.CallOption) ({service.name}_{self.to_pascal_case(method.name)}Client, error)"
+ lines.append(f"\t{self.to_pascal_case(method.name)}{signature}")
+ else:
+ signature = f"(ctx context.Context, opts ...grpc.CallOption) ({service.name}_{self.to_pascal_case(method.name)}Client, error)"
+ lines.append(f"\t{self.to_pascal_case(method.name)}{signature}")
+
+ lines.append("}")
+ lines.append("")
+ return lines
+
+ def _generate_client_struct(self, service: Service) -> List[str]:
+ lines: List[str] = []
+ lines.append(f"type {self.to_camel_case(service.name)}Client struct {{")
+ lines.append("\tcc grpc.ClientConnInterface")
+ lines.append("\tfory *fory.Fory")
+ lines.append("}")
+ lines.append("")
+ return lines
+
+ def _generate_new_client(self, service: Service) -> List[str]:
+ lines: List[str] = []
+ lines.append(
+ f"func New{service.name}Client(cc grpc.ClientConnInterface, f *fory.Fory) {service.name}Client {{"
+ )
+ lines.append(
+ f"\treturn &{self.to_camel_case(service.name)}Client{{cc: cc, fory: f}}"
+ )
+ lines.append("}")
+ lines.append("")
+ return lines
+
+ def _generate_codec(self) -> List[str]:
+ lines: List[str] = []
+ lines.append(
+ "// CodecV2 implements grpc/encoding.CodecV2 using Fory serialization."
+ )
+ lines.append(
+ "// Pass a configured *fory.Fory instance with all message types registered."
+ )
+ lines.append("type CodecV2 struct {")
+ lines.append("\tFory *fory.Fory")
+ lines.append("}")
+ lines.append("")
+ lines.append(
+ "// Marshal serializes v with Fory. The result is copied before being handed"
+ )
+ lines.append(
+ "// to gRPC because Fory reuses its internal write buffer across calls —"
+ )
+ lines.append(
+ "// streaming handlers may buffer multiple frames before sending, and without"
+ )
+ lines.append("// a copy all frames would alias the last serialized value.")
+ lines.append("func (c CodecV2) Marshal(v any) (mem.BufferSlice, error) {")
+ lines.append("\tb, err := c.Fory.Marshal(v)")
+ lines.append("\tif err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\tout := make([]byte, len(b))")
+ lines.append("\tcopy(out, b)")
+ lines.append("\treturn mem.BufferSlice{mem.NewBuffer(&out, nil)}, nil")
+ lines.append("}")
+ lines.append("")
+ lines.append(
+ "// Unmarshal deserializes the gRPC frame into v. Each buffer segment is"
+ )
+ lines.append(
+ "// copied into a fresh slice because the transport may reclaim the"
+ )
+ lines.append("// underlying memory before Fory finishes reading it.")
+ lines.append("func (c CodecV2) Unmarshal(data mem.BufferSlice, v any) error {")
+ lines.append("\tb := make([]byte, data.Len())")
+ lines.append("\tn := 0")
+ lines.append("\tfor _, buf := range data {")
+ lines.append("\t\tn += copy(b[n:], buf.ReadOnlyData())")
+ lines.append("\t}")
+ lines.append("\treturn c.Fory.Unmarshal(b, v)")
+ lines.append("}")
+ lines.append("")
+ lines.append(
+ '// Name returns "fory", the codec identifier used with grpc.ForceCodecV2.'
+ )
+ lines.append('func (CodecV2) Name() string { return "fory" }')
+ lines.append("")
+ return lines
+
+ def _generate_client_methods(
+ self, service: Service, tracker: ImportTracker
+ ) -> List[str]:
+ lines: List[str] = []
+ stream_index = 0
+ for method in service.methods:
+ req_type = self._resolve_go_type(method.request_type, tracker)
+ res_type = self._resolve_go_type(method.response_type, tracker)
+ mode = streaming_mode(method)
+ if mode is StreamingMode.UNARY:
+ lines.append(
+ f"func (c *{self.to_camel_case(service.name)}Client) {self.to_pascal_case(method.name)}(ctx context.Context, in {req_type}, opts ...grpc.CallOption) ({res_type}, error) {{"
+ )
+ lines.append(f"\tout := new({res_type[1:]})")
+ lines.append(
+ "\tcallOpts := append([]grpc.CallOption{grpc.ForceCodecV2(CodecV2{Fory: c.fory})}, opts...)"
+ )
+ lines.append(
+ f'\terr := c.cc.Invoke(ctx, "{self.get_grpc_method_path(service, method)}", in, out, callOpts...)'
+ )
+ lines.append("\tif err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\treturn out, nil")
+ lines.append("}")
+ lines.append("")
+ elif mode is StreamingMode.SERVER_STREAMING:
+ lines.append(
+ f"func (c *{self.to_camel_case(service.name)}Client) {self.to_pascal_case(method.name)}(ctx context.Context, in {req_type}, opts ...grpc.CallOption) ({service.name}_{self.to_pascal_case(method.name)}Client, error) {{"
+ )
+ lines.append(
+ "\tcallOpts := append([]grpc.CallOption{grpc.ForceCodecV2(CodecV2{Fory: c.fory})}, opts...)"
+ )
+ lines.append(
+ f'\tstream, err := c.cc.NewStream(ctx, &_{service.name}_serviceDesc.Streams[{stream_index}], "{self.get_grpc_method_path(service, method)}", callOpts...)'
+ )
+ lines.append("\tif err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append(
+ f"\tx := &{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client{{stream}}"
+ )
+ lines.append("\tif err := x.SendMsg(in); err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\tif err := x.CloseSend(); err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\treturn x, nil")
+ lines.append("}")
+ lines.append("")
+ stream_index += 1
+ else:
+ lines.append(
+ f"func (c *{self.to_camel_case(service.name)}Client) {self.to_pascal_case(method.name)}(ctx context.Context, opts ...grpc.CallOption) ({service.name}_{self.to_pascal_case(method.name)}Client, error) {{"
+ )
+ lines.append(
+ "\tcallOpts := append([]grpc.CallOption{grpc.ForceCodecV2(CodecV2{Fory: c.fory})}, opts...)"
+ )
+ lines.append(
+ f'\tstream, err := c.cc.NewStream(ctx, &_{service.name}_serviceDesc.Streams[{stream_index}], "{self.get_grpc_method_path(service, method)}", callOpts...)'
+ )
+ lines.append("\tif err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append(
+ f"\treturn &{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client{{stream}}, nil"
+ )
+ lines.append("}")
+ lines.append("")
+ stream_index += 1
+ return lines
+
+ def _generate_stream_types(
+ self, service: Service, tracker: ImportTracker
+ ) -> List[str]:
+ lines: List[str] = []
+ for method in service.methods:
+ req_type = self._resolve_go_type(method.request_type, tracker)
+ res_type = self._resolve_go_type(method.response_type, tracker)
+ mode = streaming_mode(method)
+ if mode is StreamingMode.UNARY:
+ continue
+ if mode is StreamingMode.CLIENT_STREAMING:
+ # type interface
+ lines.append(
+ f"type {service.name}_{self.to_pascal_case(method.name)}Client interface {{"
+ )
+ lines.append(f"\tSend({req_type}) error")
+ lines.append(f"\tCloseAndRecv() ({res_type}, error)")
+ lines.append("\tgrpc.ClientStream")
+ lines.append("}")
+ lines.append("")
+
+ # struct
+ lines.append(
+ f"type {self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client struct {{"
+ )
+ lines.append("\tgrpc.ClientStream")
+ lines.append("}")
+ lines.append("")
+
+ # methods
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client) Send(m {req_type}) error {{"
+ )
+ lines.append("\treturn x.ClientStream.SendMsg(m)")
+ lines.append("}")
+ lines.append("")
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client) CloseAndRecv() ({res_type}, error) {{"
+ )
+ lines.append("\tif err := x.ClientStream.CloseSend(); err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append(f"\tm := new({res_type[1:]})")
+ lines.append("\tif err := x.ClientStream.RecvMsg(m); err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\treturn m, nil")
+ lines.append("}")
+ lines.append("")
+ elif mode is StreamingMode.SERVER_STREAMING:
+ # type interface
+ lines.append(
+ f"type {service.name}_{self.to_pascal_case(method.name)}Client interface {{"
+ )
+ lines.append(f"\tRecv() ({res_type}, error)")
+ lines.append("\tgrpc.ClientStream")
+ lines.append("}")
+ lines.append("")
+
+ # struct
+ lines.append(
+ f"type {self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client struct {{"
+ )
+ lines.append("\tgrpc.ClientStream")
+ lines.append("}")
+ lines.append("")
+
+ # methods
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client) Recv() ({res_type}, error) {{"
+ )
+ lines.append(f"\tm := new({res_type[1:]})")
+ lines.append("\tif err := x.ClientStream.RecvMsg(m); err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\treturn m, nil")
+ lines.append("}")
+ lines.append("")
+ else:
+ # interface
+ lines.append(
+ f"type {service.name}_{self.to_pascal_case(method.name)}Client interface {{"
+ )
+ lines.append(f"\tSend({req_type}) error")
+ lines.append(f"\tRecv() ({res_type}, error)")
+ lines.append("\tgrpc.ClientStream")
+ lines.append("}")
+ lines.append("")
+
+ # struct
+ lines.append(
+ f"type {self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client struct {{"
+ )
+ lines.append("\tgrpc.ClientStream")
+ lines.append("}")
+ lines.append("")
+
+ # methods
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client) Send(m {req_type}) error {{"
+ )
+ lines.append("\treturn x.ClientStream.SendMsg(m)")
+ lines.append("}")
+ lines.append("")
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Client) Recv() ({res_type}, error) {{"
+ )
+ lines.append(f"\tm := new({res_type[1:]})")
+ lines.append("\tif err := x.ClientStream.RecvMsg(m); err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\treturn m, nil")
+ lines.append("}")
+ lines.append("")
+ return lines
+
+ def _generate_server_interface(
+ self, service: Service, tracker: ImportTracker
+ ) -> List[str]:
+ lines: List[str] = []
+ lines.append(
+ f"// {service.name}Server is the server API for {service.name} service."
+ )
+ lines.append(
+ f"// All implementations must embed Unimplemented{service.name}Server"
+ )
+ lines.append("// for forward compatibility.")
+ lines.append(f"type {service.name}Server interface {{")
+ for method in service.methods:
+ req_type = self._resolve_go_type(method.request_type, tracker)
+ res_type = self._resolve_go_type(method.response_type, tracker)
+ mode = streaming_mode(method)
+ if mode is StreamingMode.UNARY:
+ lines.append(
+ f"\t{self.to_pascal_case(method.name)}(context.Context, {req_type}) ({res_type}, error)"
+ )
+ elif mode is StreamingMode.SERVER_STREAMING:
+ lines.append(
+ f"\t{self.to_pascal_case(method.name)}({req_type}, {service.name}_{self.to_pascal_case(method.name)}Server) error"
+ )
+ else:
+ lines.append(
+ f"\t{self.to_pascal_case(method.name)}({service.name}_{self.to_pascal_case(method.name)}Server) error"
+ )
+ lines.append(f"\tmustEmbedUnimplemented{service.name}Server()")
+ lines.append("}")
+ lines.append("")
+ return lines
+
+ def _generate_unimplemented_server(
+ self, service: Service, tracker: ImportTracker
+ ) -> List[str]:
+ lines: List[str] = []
+ lines.append(
+ f"// Unimplemented{service.name}Server must be embedded to have forward compatible implementation."
+ )
+ lines.append(f"type Unimplemented{service.name}Server struct {{}}")
+ lines.append("")
+ lines.append(
+ f"func (Unimplemented{service.name}Server) mustEmbedUnimplemented{service.name}Server() {{}}"
+ )
+ lines.append("")
+ for method in service.methods:
+ req_type = self._resolve_go_type(method.request_type, tracker)
+ res_type = self._resolve_go_type(method.response_type, tracker)
+ mode = streaming_mode(method)
+ if mode is StreamingMode.UNARY:
+ lines.append(
+ f"func (Unimplemented{service.name}Server) {self.to_pascal_case(method.name)}(context.Context, {req_type}) ({res_type}, error) {{"
+ )
+ lines.append(
+ f'\treturn nil, status.Errorf(codes.Unimplemented, "method {self.to_pascal_case(method.name)} not implemented")'
+ )
+ lines.append("}")
+ lines.append("")
+ elif mode is StreamingMode.SERVER_STREAMING:
+ lines.append(
+ f"func (Unimplemented{service.name}Server) {self.to_pascal_case(method.name)}({req_type}, {service.name}_{self.to_pascal_case(method.name)}Server) error {{"
+ )
+ lines.append(
+ f'\treturn status.Errorf(codes.Unimplemented, "method {self.to_pascal_case(method.name)} not implemented")'
+ )
+ lines.append("}")
+ lines.append("")
+ else:
+ lines.append(
+ f"func (Unimplemented{service.name}Server) {self.to_pascal_case(method.name)}({service.name}_{self.to_pascal_case(method.name)}Server) error {{"
+ )
+ lines.append(
+ f'\treturn status.Errorf(codes.Unimplemented, "method {self.to_pascal_case(method.name)} not implemented")'
+ )
+ lines.append("}")
+ lines.append("")
+ return lines
+
+ def _generate_register_server(self, service: Service) -> List[str]:
+ lines: List[str] = []
+ lines.append(
+ f"func Register{service.name}Server(s grpc.ServiceRegistrar, srv {service.name}Server) {{"
+ )
+ lines.append(f"\ts.RegisterService(&_{service.name}_serviceDesc, srv)")
+ lines.append("}")
+ lines.append("")
+ return lines
+
+ def _generate_server_stream_types(
+ self, service: Service, tracker: ImportTracker
+ ) -> List[str]:
+ lines: List[str] = []
+ for method in service.methods:
+ req_type = self._resolve_go_type(method.request_type, tracker)
+ res_type = self._resolve_go_type(method.response_type, tracker)
+ mode = streaming_mode(method)
+ if mode is StreamingMode.UNARY:
+ continue
+ elif mode is StreamingMode.CLIENT_STREAMING:
+ # type interface
+ lines.append(
+ f"type {service.name}_{self.to_pascal_case(method.name)}Server interface {{"
+ )
+ lines.append(f"\tRecv() ({req_type}, error)")
+ lines.append(f"\tSendAndClose({res_type}) error")
+ lines.append("\tgrpc.ServerStream")
+ lines.append("}")
+ lines.append("")
+
+ # struct
+ lines.append(
+ f"type {self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server struct {{"
+ )
+ lines.append("\tgrpc.ServerStream")
+ lines.append("}")
+ lines.append("")
+
+ # methods
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server) Recv() ({req_type}, error) {{"
+ )
+ lines.append(f"\tm := new({req_type[1:]})")
+ lines.append("\tif err := x.ServerStream.RecvMsg(m); err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\treturn m, nil")
+ lines.append("}")
+ lines.append("")
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server) SendAndClose(m {res_type}) error {{"
+ )
+ lines.append("\treturn x.ServerStream.SendMsg(m)")
+ lines.append("}")
+ lines.append("")
+ elif mode is StreamingMode.SERVER_STREAMING:
+ # type interface
+ lines.append(
+ f"type {service.name}_{self.to_pascal_case(method.name)}Server interface {{"
+ )
+ lines.append(f"\tSend({res_type}) error")
+ lines.append("\tgrpc.ServerStream")
+ lines.append("}")
+ lines.append("")
+
+ # struct
+ lines.append(
+ f"type {self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server struct {{"
+ )
+ lines.append("\tgrpc.ServerStream")
+ lines.append("}")
+ lines.append("")
+
+ # methods
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server) Send(m {res_type}) error {{"
+ )
+ lines.append("\treturn x.ServerStream.SendMsg(m)")
+ lines.append("}")
+ lines.append("")
+ else:
+ # type interface
+ lines.append(
+ f"type {service.name}_{self.to_pascal_case(method.name)}Server interface {{"
+ )
+ lines.append(f"\tSend({res_type}) error")
+ lines.append(f"\tRecv() ({req_type}, error)")
+ lines.append("\tgrpc.ServerStream")
+ lines.append("}")
+ lines.append("")
+
+ # struct
+ lines.append(
+ f"type {self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server struct {{"
+ )
+ lines.append("\tgrpc.ServerStream")
+ lines.append("}")
+ lines.append("")
+
+ # methods
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server) Send(m {res_type}) error {{"
+ )
+ lines.append("\treturn x.ServerStream.SendMsg(m)")
+ lines.append("}")
+ lines.append("")
+ lines.append(
+ f"func (x *{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server) Recv() ({req_type}, error) {{"
+ )
+ lines.append(f"\tm := new({req_type[1:]})")
+ lines.append("\tif err := x.ServerStream.RecvMsg(m); err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\treturn m, nil")
+ lines.append("}")
+ lines.append("")
+ return lines
+
+ def _generate_service_desc(
+ self, service: Service, tracker: ImportTracker
+ ) -> List[str]:
+ lines: List[str] = []
+ for method in service.methods:
+ req_type = self._resolve_go_type(method.request_type, tracker)
+ mode = streaming_mode(method)
+ # handlers
+ if mode is StreamingMode.UNARY:
+ lines.append(
+ f"func _{service.name}_{self.to_pascal_case(method.name)}_Handler(srv interface{{}}, ctx context.Context, dec func(interface{{}}) error, interceptor grpc.UnaryServerInterceptor) (interface{{}}, error) {{"
+ )
+ lines.append(f"\tin := new({req_type[1:]})")
+ lines.append("\tif err := dec(in); err != nil {")
+ lines.append("\t\treturn nil, err")
+ lines.append("\t}")
+ lines.append("\tif interceptor == nil {")
+ lines.append(
+ f"\t\treturn srv.({service.name}Server).{self.to_pascal_case(method.name)}(ctx, in)"
+ )
+ lines.append("\t}")
+ lines.append("\tinfo := &grpc.UnaryServerInfo{")
+ lines.append("\t\tServer:\tsrv,")
+ lines.append(
+ f'\t\tFullMethod:\t"{self.get_grpc_method_path(service, method)}",'
+ )
+ lines.append("\t}")
+ lines.append(
+ "\thandler := func(ctx context.Context, req interface{}) (interface{}, error) {"
+ )
+ lines.append(
+ f"\t\treturn srv.({service.name}Server).{self.to_pascal_case(method.name)}(ctx, req.({req_type}))"
+ )
+ lines.append("\t}")
+ lines.append("\treturn interceptor(ctx, in, info, handler)")
+ lines.append("}")
+ lines.append("")
+ elif mode is StreamingMode.SERVER_STREAMING:
+ lines.append(
+ f"func _{service.name}_{self.to_pascal_case(method.name)}_Handler(srv interface{{}}, stream grpc.ServerStream) error {{"
+ )
+ lines.append(f"\tm := new({req_type[1:]})")
+ lines.append("\tif err := stream.RecvMsg(m); err != nil {")
+ lines.append("\t\treturn err")
+ lines.append("\t}")
+ lines.append(
+ f"\treturn srv.({service.name}Server).{self.to_pascal_case(method.name)}(m, &{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server{{stream}})"
+ )
+ lines.append("}")
+ lines.append("")
+ else:
+ lines.append(
+ f"func _{service.name}_{self.to_pascal_case(method.name)}_Handler(srv interface{{}}, stream grpc.ServerStream) error {{"
+ )
+ lines.append(
+ f"\treturn srv.({service.name}Server).{self.to_pascal_case(method.name)}(&{self.to_camel_case(service.name)}{self.to_pascal_case(method.name)}Server{{stream}})"
+ )
+ lines.append("}")
+ lines.append("")
+ # var
+ lines.append(f"var _{service.name}_serviceDesc = grpc.ServiceDesc{{")
+ lines.append(f'\tServiceName: "{self.get_grpc_service_name(service)}",')
+ lines.append(f"\tHandlerType: (*{service.name}Server)(nil),")
+ # unary type service descriptors
+ lines.append("\tMethods: []grpc.MethodDesc{")
+ lines.extend(self._generate_unary_type_desc(service))
+ lines.append("\t},")
+ # stream type service descriptors
+ lines.append("\tStreams: []grpc.StreamDesc{")
+ lines.extend(self._generate_stream_type_desc(service))
+ lines.append("\t},")
+ lines.append(f'\tMetadata: "{self.get_file_name()}.fdl",')
+ lines.append("}")
+ lines.append("")
+ return lines
+
+ def _generate_unary_type_desc(self, service: Service) -> List[str]:
+ lines: List[str] = []
+ for method in service.methods:
+ mode = streaming_mode(method)
+ if mode is StreamingMode.UNARY:
+ lines.append("\t\t{")
+ lines.append(
+ f'\t\t\tMethodName:\t"{self.to_pascal_case(method.name)}",'
+ )
+ lines.append(
+ f"\t\t\tHandler:\t_{service.name}_{self.to_pascal_case(method.name)}_Handler,"
+ )
+ lines.append("\t\t},")
+ return lines
+
+ def _generate_stream_type_desc(self, service: Service) -> List[str]:
+ lines: List[str] = []
+ for method in service.methods:
+ mode = streaming_mode(method)
+ if mode is StreamingMode.UNARY:
+ continue
+ else:
+ lines.append("\t\t{")
+ lines.append(
+ f'\t\t\tStreamName:\t"{self.to_pascal_case(method.name)}",'
+ )
+ lines.append(
+ f"\t\t\tHandler:\t_{service.name}_{self.to_pascal_case(method.name)}_Handler,"
+ )
+ if (
+ mode is StreamingMode.CLIENT_STREAMING
+ or mode is StreamingMode.BIDIRECTIONAL
+ ):
+ lines.append("\t\t\tClientStreams:\ttrue,")
+ if (
+ mode is StreamingMode.SERVER_STREAMING
+ or mode is StreamingMode.BIDIRECTIONAL
+ ):
+ lines.append("\t\t\tServerStreams:\ttrue,")
+ lines.append("\t\t},")
+ return lines
diff --git a/compiler/fory_compiler/tests/test_service_codegen.py b/compiler/fory_compiler/tests/test_service_codegen.py
index 2850974a4..58269d0f7 100644
--- a/compiler/fory_compiler/tests/test_service_codegen.py
+++ b/compiler/fory_compiler/tests/test_service_codegen.py
@@ -129,7 +129,7 @@ def test_service_definition_does_not_affect_message_codegen():
def test_generate_services_returns_empty_list_for_unsupported_generators():
schema = parse_fdl(_GREETER_WITH_SERVICE)
for generator_cls in GENERATOR_CLASSES:
- if generator_cls in (JavaGenerator, PythonGenerator):
+ if generator_cls in (JavaGenerator, PythonGenerator, GoGenerator):
continue
options = GeneratorOptions(output_dir=Path("/tmp"))
generator = generator_cls(schema, options)
@@ -218,6 +218,30 @@ def test_grpc_streaming_method_shapes():
assert "self.client = channel.stream_unary(" in python
assert "self.bidi = channel.stream_stream(" in python
+ go = next(iter(generate_service_files(schema, GoGenerator).values()))
+ assert "ClientStreams:\ttrue" in go
+ assert "ServerStreams:\ttrue" in go
+ assert "grpc.ClientStream" in go
+ assert "grpc.ServerStream" in go
+
+
+def test_go_grpc_service_codegen():
+ schema = parse_fdl(_GREETER_WITH_SERVICE)
+ files = generate_service_files(schema, GoGenerator)
+ assert len(files) == 1
+ content = next(iter(files.values()))
+ assert "func NewGreeterClient(" in content
+ assert "func RegisterGreeterServer(" in content
+ assert "type GreeterClient interface" in content
+ assert "type GreeterServer interface" in content
+ assert "type UnimplementedGreeterServer struct" in content
+ assert "CodecV2{Fory: c.fory}" in content
+ assert "type CodecV2 struct" in content
+ assert "func (c CodecV2) Marshal(v any) (mem.BufferSlice, error)" in content
+ assert "func (c CodecV2) Unmarshal(data mem.BufferSlice, v any) error" in content
+ assert '"/demo.greeter.Greeter/SayHello"' in content
+ assert "mustEmbedUnimplementedGreeterServer()" in content
+
def test_java_outer_classname_service_references_nested_model_types():
schema = parse_fdl(
diff --git a/integration_tests/grpc_tests/generate_grpc.py b/integration_tests/grpc_tests/generate_grpc.py
index 22ea0a446..5d3131640 100644
--- a/integration_tests/grpc_tests/generate_grpc.py
+++ b/integration_tests/grpc_tests/generate_grpc.py
@@ -33,6 +33,7 @@ SCHEMAS = [
OUTPUTS = {
"java": TEST_DIR / "java/src/main/java/generated",
"python": TEST_DIR / "python/grpc_tests/generated",
+ "go": TEST_DIR / "go/generated",
}
@@ -57,6 +58,12 @@ def main() -> int:
)
for schema in SCHEMAS:
+ # Each Go schema goes into its own subdir (named after the package) so
+ # multiple schemas with different package names can share the same
+ # generated/ root without one package's files overwriting another's.
+ pkg_name = schema.stem.replace(".", "_")
+ go_pkg_out = OUTPUTS["go"] / pkg_name
+ go_pkg_out.mkdir(parents=True, exist_ok=True)
subprocess.check_call(
[
sys.executable,
@@ -66,6 +73,7 @@ def main() -> int:
str(schema),
f"--java_out={OUTPUTS['java']}",
f"--python_out={OUTPUTS['python']}",
+ f"--go_out={go_pkg_out}",
"--grpc",
],
env=env,
diff --git a/integration_tests/grpc_tests/go/go.mod b/integration_tests/grpc_tests/go/go.mod
new file mode 100644
index 000000000..461aef088
--- /dev/null
+++ b/integration_tests/grpc_tests/go/go.mod
@@ -0,0 +1,35 @@
+// 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.
+
+module github.com/apache/fory/integration_tests/grpc_tests/go
+
+go 1.25.0
+
+require (
+ github.com/apache/fory/go/fory v0.0.0
+ google.golang.org/grpc v1.68.0
+)
+
+require (
+ golang.org/x/net v0.29.0 // indirect
+ golang.org/x/sys v0.25.0 // indirect
+ golang.org/x/text v0.18.0 // indirect
+ google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
+ google.golang.org/protobuf v1.34.2 // indirect
+)
+
+replace github.com/apache/fory/go/fory => ../../../go/fory
diff --git a/integration_tests/grpc_tests/go/go.sum b/integration_tests/grpc_tests/go/go.sum
new file mode 100644
index 000000000..177fd3c66
--- /dev/null
+++ b/integration_tests/grpc_tests/go/go.sum
@@ -0,0 +1,24 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo=
+golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0=
+golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
+golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
+golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU=
+google.golang.org/grpc v1.68.0 h1:aHQeeJbo8zAkAa3pRzrVjZlbz6uSfeOXlJNQM0RAbz0=
+google.golang.org/grpc v1.68.0/go.mod h1:fmSPC5AsjSBCK54MyHRx48kpOti1/jRfOlwEWywNjWA=
+google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
+google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/integration_tests/grpc_tests/go/main.go b/integration_tests/grpc_tests/go/main.go
new file mode 100644
index 000000000..caa9e3909
--- /dev/null
+++ b/integration_tests/grpc_tests/go/main.go
@@ -0,0 +1,299 @@
+// 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.
+
+// Binary grpc-interop is the Go peer for Java-driven gRPC integration tests.
+// It is invoked as a subprocess by GrpcInteropTest.java and supports two modes:
+//
+// server --port-file <path> start a gRPC server and write the bound port to the file
+// client --target <addr> connect to addr and exercise all four streaming modes
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "io"
+ "log"
+ "net"
+ "os"
+ "strings"
+
+ "github.com/apache/fory/go/fory"
+ grpc_fdl "github.com/apache/fory/integration_tests/grpc_tests/go/generated/grpc_fdl"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
+ "google.golang.org/grpc/encoding"
+)
+
+// --- helpers ----------------------------------------------------------------
+
+func newFory() *fory.Fory {
+ f := fory.New(fory.WithXlang(true), fory.WithRefTracking(true), fory.WithCompatible(true))
+ if err := grpc_fdl.RegisterTypes(f); err != nil {
+ log.Fatalf("RegisterTypes: %v", err)
+ }
+ return f
+}
+
+func fdlResponse(req *grpc_fdl.GrpcFdlRequest, tag string, offset int) *grpc_fdl.GrpcFdlResponse {
+ return &grpc_fdl.GrpcFdlResponse{
+ Id: fmt.Sprintf("%s:%s", tag, req.Id),
+ Count: req.Count + int32(offset),
+ Payload: fmt.Sprintf("%s:%s", tag, req.Payload),
+ }
+}
+
+func fdlAggregate(requests []*grpc_fdl.GrpcFdlRequest) *grpc_fdl.GrpcFdlResponse {
+ ids := make([]string, len(requests))
+ payloads := make([]string, len(requests))
+ var count int32
+ for i, req := range requests {
+ ids[i] = req.Id
+ payloads[i] = req.Payload
+ count += req.Count
+ }
+ return &grpc_fdl.GrpcFdlResponse{
+ Id: "client:" + strings.Join(ids, "+"),
+ Count: count,
+ Payload: "client:" + strings.Join(payloads, "+"),
+ }
+}
+
+// protoFallbackCodec overrides the built-in "proto" codec (v1 registry) with
+// Fory so the server can decode requests from Java clients, which send with the
+// default content-type (application/grpc) rather than application/grpc+fory.
+type protoFallbackCodec struct{ grpc_fdl.CodecV2 }
+
+func (protoFallbackCodec) Name() string { return "proto" }
+
+func (c protoFallbackCodec) Marshal(v interface{}) ([]byte, error) {
+ b, err := c.Fory.Marshal(v)
+ if err != nil {
+ return nil, err
+ }
+ out := make([]byte, len(b))
+ copy(out, b)
+ return out, nil
+}
+
+func (c protoFallbackCodec) Unmarshal(data []byte, v interface{}) error {
+ return c.Fory.Unmarshal(data, v)
+}
+
+// --- server -----------------------------------------------------------------
+
+type fdlService struct {
+ grpc_fdl.UnimplementedFdlGrpcServiceServer
+}
+
+func (s *fdlService) UnaryMessage(_ context.Context, req *grpc_fdl.GrpcFdlRequest) (*grpc_fdl.GrpcFdlResponse, error) {
+ return fdlResponse(req, "unary", 10), nil
+}
+
+func (s *fdlService) ServerStreamMessage(req *grpc_fdl.GrpcFdlRequest, stream grpc_fdl.FdlGrpcService_ServerStreamMessageServer) error {
+ for i := 0; i < 3; i++ {
+ if err := stream.Send(fdlResponse(req, fmt.Sprintf("server-%d", i), i)); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func (s *fdlService) ClientStreamMessage(stream grpc_fdl.FdlGrpcService_ClientStreamMessageServer) error {
+ var requests []*grpc_fdl.GrpcFdlRequest
+ for {
+ req, err := stream.Recv()
+ if err == io.EOF {
+ return stream.SendAndClose(fdlAggregate(requests))
+ }
+ if err != nil {
+ return err
+ }
+ requests = append(requests, req)
+ }
+}
+
+func (s *fdlService) BidiStreamMessage(stream grpc_fdl.FdlGrpcService_BidiStreamMessageServer) error {
+ index := 0
+ for {
+ req, err := stream.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ if err := stream.Send(fdlResponse(req, fmt.Sprintf("bidi-%d", index), index)); err != nil {
+ return err
+ }
+ index++
+ }
+}
+
+func runServer(portFile string, f *fory.Fory) error {
+ lis, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ return fmt.Errorf("listen: %w", err)
+ }
+ codec := grpc_fdl.CodecV2{Fory: f}
+ // "fory" (v2): used by Go clients via grpc.ForceCodecV2.
+ encoding.RegisterCodecV2(codec)
+ // "proto" (v1): overrides the built-in proto codec so Java clients are
+ // also handled by Fory. RegisterCodecV2 writes to a separate registry and
+ // does not replace the v1 "proto" entry; RegisterCodec must be used.
+ encoding.RegisterCodec(protoFallbackCodec{codec})
+ s := grpc.NewServer()
+ grpc_fdl.RegisterFdlGrpcServiceServer(s, &fdlService{})
+
+ // Write the bound port so the Java test harness knows where to connect.
+ port := lis.Addr().(*net.TCPAddr).Port
+ if err := os.WriteFile(portFile, []byte(fmt.Sprintf("%d", port)), 0600); err != nil {
+ return fmt.Errorf("write port file: %w", err)
+ }
+
+ return s.Serve(lis)
+}
+
+// --- client -----------------------------------------------------------------
+
+func exerciseMessageStub(stub grpc_fdl.FdlGrpcServiceClient, requests []*grpc_fdl.GrpcFdlRequest) error {
+ ctx := context.Background()
+ first := requests[0]
+
+ // unary
+ got, err := stub.UnaryMessage(ctx, first)
+ if err != nil {
+ return fmt.Errorf("UnaryMessage: %w", err)
+ }
+ if want := fdlResponse(first, "unary", 10); *got != *want {
+ return fmt.Errorf("UnaryMessage: got %+v, want %+v", got, want)
+ }
+
+ // server streaming
+ ss, err := stub.ServerStreamMessage(ctx, first)
+ if err != nil {
+ return fmt.Errorf("ServerStreamMessage: %w", err)
+ }
+ for i := 0; i < 3; i++ {
+ got, err := ss.Recv()
+ if err != nil {
+ return fmt.Errorf("ServerStreamMessage Recv[%d]: %w", i, err)
+ }
+ if want := fdlResponse(first, fmt.Sprintf("server-%d", i), i); *got != *want {
+ return fmt.Errorf("ServerStreamMessage[%d]: got %+v, want %+v", i, got, want)
+ }
+ }
+ if _, err := ss.Recv(); err != io.EOF {
+ return fmt.Errorf("ServerStreamMessage: expected EOF, got %v", err)
+ }
+
+ // client streaming
+ cs, err := stub.ClientStreamMessage(ctx)
+ if err != nil {
+ return fmt.Errorf("ClientStreamMessage: %w", err)
+ }
+ for _, req := range requests {
+ if err := cs.Send(req); err != nil {
+ return fmt.Errorf("ClientStreamMessage Send: %w", err)
+ }
+ }
+ csResp, err := cs.CloseAndRecv()
+ if err != nil {
+ return fmt.Errorf("ClientStreamMessage CloseAndRecv: %w", err)
+ }
+ if want := fdlAggregate(requests); *csResp != *want {
+ return fmt.Errorf("ClientStreamMessage: got %+v, want %+v", csResp, want)
+ }
+
+ // bidirectional streaming
+ bidi, err := stub.BidiStreamMessage(ctx)
+ if err != nil {
+ return fmt.Errorf("BidiStreamMessage: %w", err)
+ }
+ for i, req := range requests {
+ if err := bidi.Send(req); err != nil {
+ return fmt.Errorf("BidiStreamMessage Send[%d]: %w", i, err)
+ }
+ got, err := bidi.Recv()
+ if err != nil {
+ return fmt.Errorf("BidiStreamMessage Recv[%d]: %w", i, err)
+ }
+ if want := fdlResponse(req, fmt.Sprintf("bidi-%d", i), i); *got != *want {
+ return fmt.Errorf("BidiStreamMessage[%d]: got %+v, want %+v", i, got, want)
+ }
+ }
+ if err := bidi.CloseSend(); err != nil {
+ return fmt.Errorf("BidiStreamMessage CloseSend: %w", err)
+ }
+ if _, err := bidi.Recv(); err != io.EOF {
+ return fmt.Errorf("BidiStreamMessage: expected EOF after CloseSend, got %v", err)
+ }
+
+ return nil
+}
+
+func runClient(target string, f *fory.Fory) error {
+ conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(insecure.NewCredentials()))
+ if err != nil {
+ return fmt.Errorf("dial %s: %w", target, err)
+ }
+ defer conn.Close()
+
+ stub := grpc_fdl.NewFdlGrpcServiceClient(conn, f)
+ requests := []*grpc_fdl.GrpcFdlRequest{
+ {Id: "fdl-a", Count: 1, Payload: "alpha"},
+ {Id: "fdl-b", Count: 2, Payload: "beta"},
+ }
+ return exerciseMessageStub(stub, requests)
+}
+
+// --- entry point ------------------------------------------------------------
+
+func main() {
+ serverCmd := flag.NewFlagSet("server", flag.ExitOnError)
+ serverPortFile := serverCmd.String("port-file", "", "path to write bound port")
+
+ clientCmd := flag.NewFlagSet("client", flag.ExitOnError)
+ clientTarget := clientCmd.String("target", "", "host:port to connect to")
+
+ if len(os.Args) < 2 {
+ log.Fatal("usage: grpc-interop <server|client> [flags]")
+ }
+
+ f := newFory()
+
+ switch os.Args[1] {
+ case "server":
+ serverCmd.Parse(os.Args[2:])
+ if *serverPortFile == "" {
+ log.Fatal("--port-file is required")
+ }
+ if err := runServer(*serverPortFile, f); err != nil {
+ log.Fatalf("server error: %v", err)
+ }
+ case "client":
+ clientCmd.Parse(os.Args[2:])
+ if *clientTarget == "" {
+ log.Fatal("--target is required")
+ }
+ if err := runClient(*clientTarget, f); err != nil {
+ log.Fatalf("client error: %v", err)
+ }
+ default:
+ log.Fatalf("unknown command %q", os.Args[1])
+ }
+}
diff --git a/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/GrpcInteropTest.java b/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/GrpcInteropTest.java
index 791c2a6a3..c494c6dc2 100644
--- a/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/GrpcInteropTest.java
+++ b/integration_tests/grpc_tests/java/src/test/java/org/apache/fory/grpc_tests/GrpcInteropTest.java
@@ -111,6 +111,18 @@ public class GrpcInteropTest {
assertFdlUnions(blocking, async, unions);
}
+ // exerciseFdlMessages exercises the four FDL message streaming modes without union methods.
+ private void exerciseFdlMessages(ManagedChannel channel) throws InterruptedException {
+ grpc_fdl.FdlGrpcServiceGrpc.FdlGrpcServiceBlockingStub blocking =
+ grpc_fdl.FdlGrpcServiceGrpc.newBlockingStub(channel);
+ grpc_fdl.FdlGrpcServiceGrpc.FdlGrpcServiceStub async =
+ grpc_fdl.FdlGrpcServiceGrpc.newStub(channel);
+
+ List<grpc_fdl.GrpcFdlRequest> messages =
+ Arrays.asList(fdlRequest("fdl-a", 1, "alpha"), fdlRequest("fdl-b", 2, "beta"));
+ assertFdlMessages(blocking, async, messages);
+ }
+
private void assertFdlMessages(
grpc_fdl.FdlGrpcServiceGrpc.FdlGrpcServiceBlockingStub blocking,
grpc_fdl.FdlGrpcServiceGrpc.FdlGrpcServiceStub async,
@@ -266,6 +278,88 @@ public class GrpcInteropTest {
pbResponse(requests.get(0), "bidi-0", 0), pbResponse(requests.get(1), "bidi-1", 1)));
}
+ @Test
+ public void testJavaServerGoClient() throws Exception {
+ Server server =
+ ServerBuilder.forPort(0)
+ .addService(new FdlService())
+ .build()
+ .start();
+ try {
+ runGo("go-grpc-client", "client", "--target", "127.0.0.1:" + server.getPort());
+ } finally {
+ server.shutdownNow();
+ server.awaitTermination(10, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testGoServerJavaClient() throws Exception {
+ Path portFile = Files.createTempFile("fory-grpc-go-", ".port");
+ Files.deleteIfExists(portFile);
+ PeerCommand command = goCommand("server", "--port-file", portFile.toString());
+ Process process = startPeer(command);
+ PeerOutputCollector outputCollector =
+ new PeerOutputCollector(process.getInputStream(), "go-grpc-server");
+ outputCollector.start();
+ try {
+ int port = waitForPort(process, outputCollector, portFile);
+ ManagedChannel channel =
+ ManagedChannelBuilder.forAddress("127.0.0.1", port).usePlaintext().build();
+ try {
+ exerciseFdlMessages(channel);
+ } finally {
+ channel.shutdownNow();
+ channel.awaitTermination(10, TimeUnit.SECONDS);
+ }
+ } finally {
+ process.destroy();
+ process.waitFor(10, TimeUnit.SECONDS);
+ if (process.isAlive()) {
+ process.destroyForcibly();
+ process.waitFor(10, TimeUnit.SECONDS);
+ }
+ outputCollector.awaitOutput();
+ Files.deleteIfExists(portFile);
+ }
+ }
+
+ private PeerCommand goCommand(String... args) {
+ Path grpcRoot = repoRoot().resolve("integration_tests").resolve("grpc_tests");
+ Path goRoot = grpcRoot.resolve("go");
+ List<String> command = new ArrayList<>();
+ command.add(goRoot.resolve("grpc-interop").toString());
+ command.addAll(Arrays.asList(args));
+ PeerCommand peerCommand = new PeerCommand();
+ peerCommand.command = command;
+ peerCommand.workDir = goRoot;
+ peerCommand.environment.put("NO_PROXY", "127.0.0.1,localhost");
+ peerCommand.environment.put("no_proxy", "127.0.0.1,localhost");
+ return peerCommand;
+ }
+
+ private void runGo(String peer, String... args) throws IOException, InterruptedException {
+ Process process = startPeer(goCommand(args));
+ PeerOutputCollector outputCollector = new PeerOutputCollector(process.getInputStream(), peer);
+ outputCollector.start();
+ boolean finished = process.waitFor(180, TimeUnit.SECONDS);
+ if (!finished) {
+ process.destroyForcibly();
+ process.waitFor(10, TimeUnit.SECONDS);
+ Assert.fail("Peer process timed out for " + peer + peerOutput(outputCollector));
+ }
+ int exitCode = process.exitValue();
+ if (exitCode != 0) {
+ Assert.fail(
+ "Peer process failed for "
+ + peer
+ + " with exit code "
+ + exitCode
+ + peerOutput(outputCollector));
+ }
+ outputCollector.awaitOutput();
+ }
+
private PeerCommand pythonCommand(String... args) {
Path repoRoot = repoRoot();
Path grpcRoot = repoRoot.resolve("integration_tests").resolve("grpc_tests");
diff --git a/integration_tests/grpc_tests/run_tests.sh b/integration_tests/grpc_tests/run_tests.sh
index 7cb2a4ad6..033d27c56 100755
--- a/integration_tests/grpc_tests/run_tests.sh
+++ b/integration_tests/grpc_tests/run_tests.sh
@@ -27,7 +27,9 @@ python -m pip install -v -e "${ROOT_DIR}/python"
python "${SCRIPT_DIR}/generate_grpc.py"
+cd "${SCRIPT_DIR}/go"
+go build -o grpc-interop .
cd "${ROOT_DIR}/integration_tests/grpc_tests/java"
-ENABLE_FORY_DEBUG_OUTPUT=1 mvn -T16 --no-transfer-progress \
+mvn -T16 --no-transfer-progress \
-Dtest=GrpcInteropTest \
test
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@fory.apache.org
For additional commands, e-mail: commits-help@fory.apache.org