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 <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
