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 0d8f412c2 feat(scala): add generated grpc service support for scala
(#3762)
0d8f412c2 is described below
commit 0d8f412c2eb3b0496986c569708c55bf7fdd742c
Author: Shawn Yang <[email protected]>
AuthorDate: Sat Jun 13 15:02:10 2026 +0800
feat(scala): add generated grpc service support for scala (#3762)
## Why?
Scala schema generation already emits Scala 3 model types and modules,
but schemas with service definitions could not generate Scala gRPC
companions. This left Scala users without generated client/server
bindings for Fory-encoded gRPC services even though Java, Python, Go,
Rust, and Kotlin already support `--grpc`.
## What does this PR do?
- Adds Scala gRPC service companion generation for `foryc
--scala_out=... --grpc`, including service descriptors, server base
classes, generated clients, Fory-backed grpc-java marshallers, unary
calls, and streaming method shapes.
- Adds Scala compiler preflight validation for package/import
combinations and generated file path collisions, including service
companion paths.
- Adds transport-free Scala RPC handle traits, `RpcFuture` and
`RpcIterator`, for generated unary and server-streaming client
convenience APIs.
- Extends service codegen tests to cover Scala marshalling output,
streaming shapes, imported type references, collision handling, keyword
escaping, protobuf/FlatBuffers service inputs, and end-to-end `--grpc`
output creation.
- Documents Scala gRPC generation across the compiler docs and Scala
guide, including dependencies, client/server examples, streaming
behavior, and cross-IDL support.
## Related issues
#3266
## AI Contribution Checklist
- [ ] Substantial AI assistance was used in this PR: `yes` / `no`
- [ ] If `yes`, I included a completed [AI Contribution
Checklist](https://github.com/apache/fory/blob/main/AI_POLICY.md#9-contributor-checklist-for-ai-assisted-prs)
in this PR description and the required `AI Usage Disclosure`.
- [ ] If `yes`, my PR description includes the required `ai_review`
summary and screenshot evidence of the final clean AI review results
from both fresh reviewers on the current PR diff or current HEAD after
the latest code changes.
## Does this PR introduce any user-facing change?
- [ ] Does this PR introduce any public API change?
- [ ] Does this PR introduce any binary protocol compatibility change?
User-facing changes:
- Adds Scala `--grpc` service companion generation and new Scala RPC
handle traits used by generated clients.
- Does not change the Fory binary protocol.
## Benchmark
---
compiler/fory_compiler/cli.py | 69 ++
compiler/fory_compiler/generators/scala.py | 126 +++-
.../fory_compiler/generators/services/scala.py | 810 +++++++++++++++++++++
.../fory_compiler/tests/test_service_codegen.py | 207 +++++-
docs/compiler/compiler-guide.md | 23 +-
docs/compiler/flatbuffers-idl.md | 6 +-
docs/compiler/generated-code.md | 86 +++
docs/compiler/index.md | 6 +-
docs/compiler/protobuf-idl.md | 27 +-
docs/compiler/schema-idl.md | 2 +-
docs/guide/java/grpc-support.md | 5 +-
docs/guide/scala/grpc-support.md | 397 ++++++++++
docs/guide/scala/index.md | 1 +
docs/guide/scala/schema-idl.md | 4 +
.../org/apache/fory/scala/rpc/RpcFuture.scala | 40 +
.../org/apache/fory/scala/rpc/RpcIterator.scala | 36 +
.../org/apache/fory/scala/rpc/RpcHandleTest.scala | 71 ++
17 files changed, 1859 insertions(+), 57 deletions(-)
diff --git a/compiler/fory_compiler/cli.py b/compiler/fory_compiler/cli.py
index a654f6a4e..7326470df 100644
--- a/compiler/fory_compiler/cli.py
+++ b/compiler/fory_compiler/cli.py
@@ -35,6 +35,10 @@ from fory_compiler.generators.kotlin import (
kotlin_output_paths,
kotlin_package_for_schema,
)
+from fory_compiler.generators.scala import (
+ scala_output_paths,
+ scala_package_for_schema,
+)
class ImportError(Exception):
@@ -317,6 +321,68 @@ def validate_kotlin_generation(
return validate_kotlin_output_paths(graph, grpc=grpc)
+def validate_scala_import_packages(graph: List[Tuple[Path, Schema]]) -> bool:
+ """Check package combinations that Scala source cannot compile."""
+ packages = {scala_package_for_schema(schema) for _, schema in graph}
+ if None not in packages or len(packages) <= 1:
+ return True
+ details = ", ".join(
+ f"{package or '<default>'}: {path}"
+ for path, schema in sorted(graph, key=lambda item: str(item[0]))
+ for package in [scala_package_for_schema(schema)]
+ )
+ print(
+ "Error: Scala imports cannot mix default-package schemas with named "
+ f"Scala packages; found {details}",
+ file=sys.stderr,
+ )
+ return False
+
+
+def validate_scala_output_paths(
+ graph: List[Tuple[Path, Schema]],
+ grpc: bool = False,
+) -> bool:
+ """Check Scala output paths for the current generation run."""
+ outputs: Dict[str, List[str]] = {}
+ for path, schema in graph:
+ for output_path, owner in scala_output_paths(schema,
include_services=grpc):
+ outputs.setdefault(output_path, []).append(f"{path} {owner}")
+ collisions = {
+ output_path: paths for output_path, paths in outputs.items() if
len(paths) > 1
+ }
+ if not collisions:
+ return True
+ details = ", ".join(
+ f"{output_path}: {', '.join(paths)}"
+ for output_path, paths in sorted(collisions.items())
+ )
+ print(
+ "Error: Scala generated file path collision; rename schema files, "
+ f"schema types, or services, or use distinct Scala packages.
Collisions: {details}",
+ file=sys.stderr,
+ )
+ return False
+
+
+def validate_scala_generation(
+ files: List[Path],
+ import_paths: List[Path],
+ grpc: bool = False,
+) -> bool:
+ """Preflight Scala package and helper paths before writing output."""
+ cache: Dict[Path, Schema] = {}
+ graph: List[Tuple[Path, Schema]] = []
+ for file_path in files:
+ file_graph = collect_schema_graph(file_path, import_paths, cache,
set())
+ if file_graph is None:
+ return False
+ graph.extend(file_graph)
+ if not validate_scala_import_packages(graph):
+ return False
+ return validate_scala_output_paths(graph, grpc=grpc)
+
+
def _find_go_module_root(base_go_out: Path) -> Optional[Path]:
base_go_out = base_go_out.resolve()
for candidate in (base_go_out, *base_go_out.parents):
@@ -899,6 +965,9 @@ def cmd_compile(args: argparse.Namespace) -> int:
if "kotlin" in lang_output_dirs:
if not validate_kotlin_generation(args.files, import_paths,
grpc=args.grpc):
return 1
+ if "scala" in lang_output_dirs:
+ if not validate_scala_generation(args.files, import_paths,
grpc=args.grpc):
+ return 1
# Create output directories
for out_dir in lang_output_dirs.values():
diff --git a/compiler/fory_compiler/generators/scala.py
b/compiler/fory_compiler/generators/scala.py
index bba6a1a6c..c978c54f1 100644
--- a/compiler/fory_compiler/generators/scala.py
+++ b/compiler/fory_compiler/generators/scala.py
@@ -23,6 +23,7 @@ from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple
from fory_compiler.generators.base import BaseGenerator, GeneratedFile
+from fory_compiler.generators.services.scala import ScalaServiceGeneratorMixin
from fory_compiler.ir.ast import (
ArrayType,
Enum,
@@ -40,7 +41,103 @@ from fory_compiler.ir.construction import analyze_shapes
from fory_compiler.ir.types import PrimitiveKind
-class ScalaGenerator(BaseGenerator):
+def _to_pascal_case(name: str) -> str:
+ if not name:
+ return name
+ if "_" in name:
+ return "".join(word.capitalize() for word in name.lower().split("_"))
+ if name.isupper():
+ return name.capitalize()
+ return name[0].upper() + name[1:]
+
+
+def scala_package_for_schema(schema: Schema) -> Optional[str]:
+ """Return the Scala source package for a schema."""
+ return schema.package
+
+
+def scala_package_path(schema: Schema) -> str:
+ """Return the generated Scala package directory."""
+ package = scala_package_for_schema(schema)
+ return package.replace(".", "/") if package else ""
+
+
+def scala_module_name(schema: Schema) -> str:
+ """Return the generated Scala schema module name."""
+ package = scala_package_for_schema(schema)
+ if package:
+ return _to_pascal_case(package.split(".")[-1]) + "ForyModule"
+ if schema.source_file and not schema.source_file.startswith("<"):
+ cleaned = "".join(
+ char if char.isascii() and (char.isalnum() or char == "_") else "_"
+ for char in Path(schema.source_file).stem
+ )
+ prefix = _to_pascal_case(cleaned) if cleaned else "Schema"
+ if not prefix or not (prefix[0].isalpha() or prefix[0] == "_"):
+ prefix = f"Schema{prefix}"
+ return prefix + "ForyModule"
+ return "ForyModule"
+
+
+def scala_source_path(schema: Schema, type_name: str) -> str:
+ """Return a generated Scala source path for a top-level owner."""
+ path = scala_package_path(schema)
+ file_name = f"{type_name}.scala"
+ return f"{path}/{file_name}" if path else file_name
+
+
+def scala_module_file_path(schema: Schema) -> str:
+ """Return the generated Scala schema module path."""
+ return scala_source_path(schema, scala_module_name(schema))
+
+
+def _is_schema_local(type_def: object, schema: Schema) -> bool:
+ if not schema.source_file or schema.source_file.startswith("<"):
+ return True
+ location = getattr(type_def, "location", None)
+ file_path = getattr(location, "file", None) if location else None
+ if not file_path:
+ return True
+ try:
+ return Path(file_path).resolve() == Path(schema.source_file).resolve()
+ except Exception:
+ return file_path == schema.source_file
+
+
+def scala_output_paths(
+ schema: Schema,
+ local_only: bool = False,
+ include_services: bool = False,
+) -> List[Tuple[str, str]]:
+ """Return generated Scala output paths and their owning schema elements."""
+ outputs: List[Tuple[str, str]] = []
+
+ def add_type(type_def: object, kind: str) -> None:
+ if local_only and not _is_schema_local(type_def, schema):
+ return
+ outputs.append((scala_source_path(schema, type_def.name), kind))
+
+ for enum in schema.enums:
+ add_type(enum, f"enum {enum.name}")
+ for union in schema.unions:
+ add_type(union, f"union {union.name}")
+ for message in schema.messages:
+ add_type(message, f"message {message.name}")
+ outputs.append((scala_module_file_path(schema), "schema module"))
+ if include_services:
+ for service in schema.services:
+ if local_only and not _is_schema_local(service, schema):
+ continue
+ outputs.append(
+ (
+ scala_source_path(schema, f"{service.name}Grpc"),
+ f"service {service.name}",
+ )
+ )
+ return outputs
+
+
+class ScalaGenerator(ScalaServiceGeneratorMixin, BaseGenerator):
"""Generates Scala 3 models with Fory macro-derived serializers."""
language_name = "scala"
@@ -136,11 +233,10 @@ class ScalaGenerator(BaseGenerator):
self._construction_shapes = analyze_shapes(schema)
def get_scala_package(self) -> Optional[str]:
- return self.schema.package
+ return scala_package_for_schema(self.schema)
def get_scala_package_path(self) -> str:
- package = self.get_scala_package()
- return package.replace(".", "/") if package else ""
+ return scala_package_path(self.schema)
def get_module_name(self) -> str:
return self._module_name(self.schema)
@@ -167,7 +263,7 @@ class ScalaGenerator(BaseGenerator):
return path_str
def _scala_package_for_schema(self, schema: Schema) -> Optional[str]:
- return schema.package
+ return scala_package_for_schema(schema)
def _validate_import_packages(self) -> None:
schemas = self._schema_graph()
@@ -184,25 +280,7 @@ class ScalaGenerator(BaseGenerator):
)
def _module_name(self, schema: Schema) -> str:
- package = self._scala_package_for_schema(schema)
- if package:
- return self.to_pascal_case(package.split(".")[-1]) + "ForyModule"
- prefix = self._module_prefix_from_source(schema)
- if prefix:
- return prefix + "ForyModule"
- return "ForyModule"
-
- def _module_prefix_from_source(self, schema: Schema) -> str:
- if not schema.source_file or schema.source_file.startswith("<"):
- return ""
- cleaned = "".join(
- char if char.isascii() and (char.isalnum() or char == "_") else "_"
- for char in Path(schema.source_file).stem
- )
- prefix = self.to_pascal_case(cleaned) if cleaned else "Schema"
- if not prefix or not (prefix[0].isalpha() or prefix[0] == "_"):
- prefix = f"Schema{prefix}"
- return prefix
+ return scala_module_name(schema)
def _scala_package_for_type(self, type_def: object) -> Optional[str]:
location = getattr(type_def, "location", None)
diff --git a/compiler/fory_compiler/generators/services/scala.py
b/compiler/fory_compiler/generators/services/scala.py
new file mode 100644
index 000000000..630b6aa9d
--- /dev/null
+++ b/compiler/fory_compiler/generators/services/scala.py
@@ -0,0 +1,810 @@
+# 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.
+
+"""Scala gRPC service generator helpers."""
+
+from __future__ import annotations
+
+from typing import Dict, List
+
+from fory_compiler.generators.base import GeneratedFile
+from fory_compiler.generators.services.base import StreamingMode,
streaming_mode
+from fory_compiler.ir.ast import NamedType, RpcMethod, Service
+
+
+class ScalaServiceGeneratorMixin:
+ """Generates Scala gRPC service companions."""
+
+ def generate_services(self) -> List[GeneratedFile]:
+ """Generate Scala gRPC service companions for local service
definitions."""
+ local_services = [
+ service
+ for service in self.schema.services
+ if not self.is_imported_type(service)
+ ]
+ if not local_services:
+ return []
+ self.check_scala_grpc_service_collisions(local_services)
+ self.check_scala_grpc_method_collisions(local_services)
+ return [
+ self.generate_scala_grpc_service_file(service) for service in
local_services
+ ]
+
+ def scala_generated_output_paths(
+ self, include_services: bool = False
+ ) -> List[tuple[str, str]]:
+ """Return generated Scala paths for the current schema graph."""
+ from fory_compiler.generators.scala import scala_output_paths
+
+ outputs: List[tuple[str, str]] = []
+ for index, (_source, schema) in enumerate(self._schema_graph()):
+ outputs.extend(
+ scala_output_paths(
+ schema,
+ local_only=index == 0,
+ include_services=include_services,
+ )
+ )
+ return outputs
+
+ def check_scala_grpc_service_collisions(self, services: List[Service]) ->
None:
+ generated_paths: Dict[str, str] = {}
+ for path, owner in
self.scala_generated_output_paths(include_services=True):
+ prior = generated_paths.get(path)
+ if prior is not None:
+ raise ValueError(
+ "Scala generated file path collision; rename schema files,
"
+ "schema types, or services, or use distinct Scala
packages. "
+ f"Collisions: {path}: {prior}, {owner}"
+ )
+ generated_paths[path] = owner
+
+ seen_service_objects: Dict[str, str] = {}
+ for service in services:
+ owner = self.scala_grpc_owner_name(service)
+ prior = seen_service_objects.get(owner)
+ if prior is not None:
+ raise ValueError(
+ f"Scala gRPC service object collision: {prior} and "
+ f"{service.name} both generate {owner}"
+ )
+ seen_service_objects[owner] = service.name
+
+ def check_scala_grpc_method_collisions(self, services: List[Service]) ->
None:
+ top_level_reserved = {
+ "SERVICE_NAME",
+ "FORY",
+ "completePromise",
+ "internalError",
+ "interrupted",
+ "marshaller",
+ "newClient",
+ "readBytes",
+ "readUnknownLengthBytes",
+ "serviceDescriptor",
+ "serviceDescriptorValue",
+ "unimplemented",
+ }
+ helper_names = {
+ "ForyMarshaller",
+ "KnownLengthByteArrayInputStream",
+ "RpcFutureAdapter",
+ "RpcIteratorAdapter",
+ }
+ for service in services:
+ seen_top_level: Dict[str, str] = {}
+ seen_service_methods: Dict[str, str] = {}
+ seen_client_methods: Dict[str, str] = {}
+ for method in service.methods:
+ base_name = self.scala_grpc_method_name(method)
+ top_level_names = [
+ self.scala_grpc_method_descriptor(method),
+ self.scala_grpc_method_backing_field(method),
+ ]
+ service_method_names = [base_name]
+ client_method_names = [base_name]
+ mode = streaming_mode(method)
+ if mode in (StreamingMode.UNARY,
StreamingMode.SERVER_STREAMING):
+ client_method_names.append(
+ self.scala_grpc_blocking_method_name(method)
+ )
+ if mode is StreamingMode.UNARY:
+ client_method_names.append(
+ self.scala_grpc_future_method_name(method)
+ )
+ for name in top_level_names:
+ if name in top_level_reserved or name in helper_names:
+ raise ValueError(
+ f"Scala gRPC method name collision in service "
+ f"{service.name}: {method.name} generates reserved
"
+ f"member {name}"
+ )
+ prior = seen_top_level.get(name)
+ if prior is not None:
+ raise ValueError(
+ f"Scala gRPC method name collision in service "
+ f"{service.name}: {prior} and {method.name} both "
+ f"generate top-level member {name}"
+ )
+ seen_top_level[name] = method.name
+ for name in service_method_names:
+ prior = seen_service_methods.get(name)
+ if prior is not None:
+ raise ValueError(
+ f"Scala gRPC method name collision in service "
+ f"{service.name}: {prior} and {method.name} both "
+ f"generate service method {name}"
+ )
+ seen_service_methods[name] = method.name
+ for name in client_method_names:
+ prior = seen_client_methods.get(name)
+ if prior is not None:
+ raise ValueError(
+ f"Scala gRPC method name collision in service "
+ f"{service.name}: {prior} and {method.name} both "
+ f"generate client method {name}"
+ )
+ seen_client_methods[name] = method.name
+
+ def generate_scala_grpc_service_file(self, service: Service) ->
GeneratedFile:
+ imports = {"org.apache.fory.scala.rpc.{RpcFuture, RpcIterator}"}
+ lines = self.source_header(imports)
+ owner = self.scala_grpc_owner_name(service)
+ lines.append(f"object {owner} {{")
+ lines.append(
+ f' final val SERVICE_NAME: String =
"{self.scala_string_literal(self.get_grpc_service_name(service))}"'
+ )
+ lines.append("")
+ lines.append(
+ " private val FORY: org.apache.fory.ThreadSafeFory = "
+ f"{self.get_module_name()}.getFory"
+ )
+ lines.append("")
+ lines.extend(self.generate_scala_grpc_service_descriptor(service))
+ for method in service.methods:
+ lines.extend(self.generate_scala_grpc_method_descriptor(method))
+ lines.extend(self.generate_scala_grpc_factories(service))
+ lines.extend(self.generate_scala_grpc_service_base(service))
+ lines.extend(self.generate_scala_grpc_client(service))
+ lines.extend(self.generate_scala_grpc_marshaller())
+ lines.append("}")
+ return self.source_file(owner, lines)
+
+ def generate_scala_grpc_service_descriptor(self, service: Service) ->
List[str]:
+ lines = [
+ " private lazy val serviceDescriptorValue:
io.grpc.ServiceDescriptor =",
+ " io.grpc.ServiceDescriptor.newBuilder(SERVICE_NAME)",
+ ]
+ for method in service.methods:
+ lines.append(
+ f"
.addMethod({self.scala_grpc_method_descriptor(method)})"
+ )
+ lines.extend(
+ [
+ " .build()",
+ "",
+ " def serviceDescriptor: io.grpc.ServiceDescriptor =",
+ " serviceDescriptorValue",
+ "",
+ ]
+ )
+ return lines
+
+ def generate_scala_grpc_method_descriptor(self, method: RpcMethod) ->
List[str]:
+ request_type = self.scala_grpc_type(method.request_type)
+ response_type = self.scala_grpc_type(method.response_type)
+ descriptor = self.scala_grpc_method_descriptor(method)
+ backing_field = self.scala_grpc_method_backing_field(method)
+ method_type = self.scala_grpc_method_type(method)
+ return [
+ f" private lazy val {backing_field}:
io.grpc.MethodDescriptor[{request_type}, {response_type}] =",
+ " io.grpc.MethodDescriptor",
+ f" .newBuilder[{request_type}, {response_type}]()",
+ f"
.setType(io.grpc.MethodDescriptor.MethodType.{method_type})",
+ " .setFullMethodName(",
+ " io.grpc.MethodDescriptor.generateFullMethodName(",
+ " SERVICE_NAME,",
+ f'
"{self.scala_string_literal(method.name)}"))',
+ " .setSampledToLocalTracing(true)",
+ f"
.setRequestMarshaller(marshaller(classOf[{request_type}]))",
+ f"
.setResponseMarshaller(marshaller(classOf[{response_type}]))",
+ " .build()",
+ "",
+ f" def {descriptor}: io.grpc.MethodDescriptor[{request_type},
{response_type}] =",
+ f" {backing_field}",
+ "",
+ ]
+
+ def generate_scala_grpc_factories(self, service: Service) -> List[str]:
+ client = self.scala_grpc_client_name(service)
+ return [
+ f" def newClient(channel: io.grpc.Channel): {client} =",
+ f" new {client}(channel, io.grpc.CallOptions.DEFAULT)",
+ "",
+ " def newClient(",
+ " channel: io.grpc.Channel,",
+ f" callOptions: io.grpc.CallOptions): {client} =",
+ f" new {client}(channel, callOptions)",
+ "",
+ ]
+
+ def generate_scala_grpc_service_base(self, service: Service) -> List[str]:
+ base_name = self.scala_grpc_service_base_name(service)
+ lines = [f" abstract class {base_name} extends
io.grpc.BindableService {{"]
+ for method in service.methods:
+ lines.extend(self.generate_scala_grpc_server_method(method))
+ lines.extend(self.generate_scala_grpc_bind_service(service))
+ lines.append(" }")
+ lines.append("")
+ return lines
+
+ def generate_scala_grpc_server_method(self, method: RpcMethod) ->
List[str]:
+ method_name = self.safe_identifier(self.scala_grpc_method_name(method))
+ request_type = self.scala_grpc_type(method.request_type)
+ response_type = self.scala_grpc_type(method.response_type)
+ descriptor = self.scala_grpc_method_descriptor(method)
+ mode = streaming_mode(method)
+ lines: List[str] = [""]
+ if mode is StreamingMode.UNARY:
+ lines.extend(
+ [
+ f" def {method_name}(request: {request_type}):
{response_type} =",
+ f' throw
unimplemented("{self.scala_string_literal(method.name)}")',
+ "",
+ " def "
+ f"{method_name}(request: {request_type}, "
+ f"responseObserver:
io.grpc.stub.StreamObserver[{response_type}]): Unit = {{",
+ " try {",
+ f" val response = {method_name}(request)",
+ " responseObserver.onNext(response)",
+ " responseObserver.onCompleted()",
+ " } catch {",
+ " case e: io.grpc.StatusException =>",
+ " responseObserver.onError(e)",
+ " case e: io.grpc.StatusRuntimeException
=>",
+ " responseObserver.onError(e)",
+ " case scala.util.control.NonFatal(e) =>",
+ " responseObserver.onError(",
+ f' internalError("Scala gRPC method
{self.scala_string_literal(method.name)} failed", e))',
+ " }",
+ " }",
+ ]
+ )
+ elif mode is StreamingMode.SERVER_STREAMING:
+ lines.extend(
+ [
+ " def "
+ f"{method_name}(request: {request_type}, "
+ f"responseObserver:
io.grpc.stub.StreamObserver[{response_type}]): Unit =",
+ "
io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall(",
+ f" {descriptor}, responseObserver)",
+ ]
+ )
+ else:
+ lines.extend(
+ [
+ f" def {method_name}(",
+ f" responseObserver:
io.grpc.stub.StreamObserver[{response_type}]",
+ f" ): io.grpc.stub.StreamObserver[{request_type}]
=",
+ "
io.grpc.stub.ServerCalls.asyncUnimplementedStreamingCall(",
+ f" {descriptor}, responseObserver)",
+ ]
+ )
+ return lines
+
+ def generate_scala_grpc_bind_service(self, service: Service) -> List[str]:
+ lines = [
+ "",
+ " final override def bindService():
io.grpc.ServerServiceDefinition = {",
+ " val builder =
io.grpc.ServerServiceDefinition.builder(serviceDescriptor)",
+ ]
+ for method in service.methods:
+ lines.extend(self.generate_scala_grpc_bind_method(service, method))
+ lines.extend([" builder.build()", " }"])
+ return lines
+
+ def generate_scala_grpc_bind_method(
+ self, service: Service, method: RpcMethod
+ ) -> List[str]:
+ base_name = self.scala_grpc_service_base_name(service)
+ method_name = self.safe_identifier(self.scala_grpc_method_name(method))
+ request_type = self.scala_grpc_type(method.request_type)
+ response_type = self.scala_grpc_type(method.response_type)
+ descriptor = self.scala_grpc_method_descriptor(method)
+ mode = streaming_mode(method)
+ if mode is StreamingMode.UNARY:
+ call = "asyncUnaryCall"
+ trait_name = "UnaryMethod"
+ invoke = [
+ f"
{base_name}.this.{method_name}(request, responseObserver)",
+ ]
+ elif mode is StreamingMode.SERVER_STREAMING:
+ call = "asyncServerStreamingCall"
+ trait_name = "ServerStreamingMethod"
+ invoke = [
+ f"
{base_name}.this.{method_name}(request, responseObserver)",
+ ]
+ elif mode is StreamingMode.CLIENT_STREAMING:
+ call = "asyncClientStreamingCall"
+ trait_name = "ClientStreamingMethod"
+ invoke = [
+ f"
{base_name}.this.{method_name}(responseObserver)",
+ ]
+ else:
+ call = "asyncBidiStreamingCall"
+ trait_name = "BidiStreamingMethod"
+ invoke = [
+ f"
{base_name}.this.{method_name}(responseObserver)",
+ ]
+ if mode in (StreamingMode.UNARY, StreamingMode.SERVER_STREAMING):
+ return [
+ " builder.addMethod(",
+ f" {descriptor},",
+ f" io.grpc.stub.ServerCalls.{call}(",
+ f" new
io.grpc.stub.ServerCalls.{trait_name}[{request_type}, {response_type}] {{",
+ " override def invoke(",
+ f" request: {request_type},",
+ f" responseObserver:
io.grpc.stub.StreamObserver[{response_type}]): Unit =",
+ *invoke,
+ " }))",
+ "",
+ ]
+ return [
+ " builder.addMethod(",
+ f" {descriptor},",
+ f" io.grpc.stub.ServerCalls.{call}(",
+ f" new
io.grpc.stub.ServerCalls.{trait_name}[{request_type}, {response_type}] {{",
+ " override def invoke(",
+ f" responseObserver:
io.grpc.stub.StreamObserver[{response_type}]",
+ f" ):
io.grpc.stub.StreamObserver[{request_type}] =",
+ *invoke,
+ " }))",
+ "",
+ ]
+
+ def generate_scala_grpc_client(self, service: Service) -> List[str]:
+ client = self.scala_grpc_client_name(service)
+ owner = self.scala_grpc_owner_name(service)
+ lines = [
+ f" final class {client} private[{owner}] (",
+ " channel: io.grpc.Channel,",
+ " callOptions: io.grpc.CallOptions)",
+ f" extends io.grpc.stub.AbstractStub[{client}](channel,
callOptions) {{",
+ " protected override def build(",
+ " channel: io.grpc.Channel,",
+ f" callOptions: io.grpc.CallOptions): {client} =",
+ f" new {client}(channel, callOptions)",
+ ]
+ for method in service.methods:
+ lines.extend(self.generate_scala_grpc_client_method(method))
+ lines.append(" }")
+ lines.append("")
+ return lines
+
+ def generate_scala_grpc_client_method(self, method: RpcMethod) ->
List[str]:
+ method_name = self.safe_identifier(self.scala_grpc_method_name(method))
+ request_type = self.scala_grpc_type(method.request_type)
+ response_type = self.scala_grpc_type(method.response_type)
+ descriptor = self.scala_grpc_method_descriptor(method)
+ mode = streaming_mode(method)
+ lines: List[str] = [""]
+ if mode is StreamingMode.UNARY:
+ future_method = self.scala_grpc_future_method_name(method)
+ blocking_method = self.scala_grpc_blocking_method_name(method)
+ lines.extend(
+ [
+ f" def {method_name}(request: {request_type}):
RpcFuture[{response_type}] =",
+ " new RpcFutureAdapter(",
+ "
io.grpc.stub.ClientCalls.futureUnaryCall(",
+ f" getChannel().newCall({descriptor},
getCallOptions()),",
+ " request))",
+ "",
+ " def "
+ f"{method_name}(request: {request_type}, "
+ f"responseObserver:
io.grpc.stub.StreamObserver[{response_type}]): Unit =",
+ " io.grpc.stub.ClientCalls.asyncUnaryCall(",
+ f" getChannel().newCall({descriptor},
getCallOptions()),",
+ " request,",
+ " responseObserver)",
+ "",
+ f" def {blocking_method}(request: {request_type}):
{response_type} =",
+ " io.grpc.stub.ClientCalls.blockingUnaryCall(",
+ f" getChannel(), {descriptor},
getCallOptions(), request)",
+ "",
+ f" def {future_method}(",
+ f" request: {request_type}):
com.google.common.util.concurrent.ListenableFuture[{response_type}] =",
+ " io.grpc.stub.ClientCalls.futureUnaryCall(",
+ f" getChannel().newCall({descriptor},
getCallOptions()),",
+ " request)",
+ ]
+ )
+ elif mode is StreamingMode.SERVER_STREAMING:
+ blocking_method = self.scala_grpc_blocking_method_name(method)
+ lines.extend(
+ [
+ f" def {method_name}(request: {request_type}):
RpcIterator[{response_type}] =",
+ " new RpcIteratorAdapter(",
+ f" getChannel().newCall({descriptor},
getCallOptions()),",
+ " request)",
+ "",
+ " def "
+ f"{method_name}(request: {request_type}, "
+ f"responseObserver:
io.grpc.stub.StreamObserver[{response_type}]): Unit =",
+ "
io.grpc.stub.ClientCalls.asyncServerStreamingCall(",
+ f" getChannel().newCall({descriptor},
getCallOptions()),",
+ " request,",
+ " responseObserver)",
+ "",
+ f" def {blocking_method}(request: {request_type}):
java.util.Iterator[{response_type}] =",
+ "
io.grpc.stub.ClientCalls.blockingServerStreamingCall(",
+ f" getChannel(), {descriptor},
getCallOptions(), request)",
+ ]
+ )
+ elif mode is StreamingMode.CLIENT_STREAMING:
+ lines.extend(
+ [
+ f" def {method_name}(",
+ f" responseObserver:
io.grpc.stub.StreamObserver[{response_type}]",
+ f" ): io.grpc.stub.StreamObserver[{request_type}]
=",
+ "
io.grpc.stub.ClientCalls.asyncClientStreamingCall(",
+ f" getChannel().newCall({descriptor},
getCallOptions()),",
+ " responseObserver)",
+ ]
+ )
+ else:
+ lines.extend(
+ [
+ f" def {method_name}(",
+ f" responseObserver:
io.grpc.stub.StreamObserver[{response_type}]",
+ f" ): io.grpc.stub.StreamObserver[{request_type}]
=",
+ "
io.grpc.stub.ClientCalls.asyncBidiStreamingCall(",
+ f" getChannel().newCall({descriptor},
getCallOptions()),",
+ " responseObserver)",
+ ]
+ )
+ return lines
+
+ def generate_scala_grpc_marshaller(self) -> List[str]:
+ return [
+ " private final class RpcFutureAdapter[A](",
+ " delegate:
com.google.common.util.concurrent.ListenableFuture[A])",
+ " extends RpcFuture[A] {",
+ " @volatile private var futureView:
scala.concurrent.Future[A] = null",
+ "",
+ " override def asFuture: scala.concurrent.Future[A] = {",
+ " var local = futureView",
+ " if (local eq null) {",
+ " this.synchronized {",
+ " local = futureView",
+ " if (local eq null) {",
+ " val promise =
scala.concurrent.Promise[A]()",
+ " delegate.addListener(",
+ " new Runnable {",
+ " override def run(): Unit =",
+ " completePromise(delegate,
promise)",
+ " },",
+ "
com.google.common.util.concurrent.MoreExecutors.directExecutor())",
+ " local = promise.future",
+ " futureView = local",
+ " }",
+ " }",
+ " }",
+ " local",
+ " }",
+ "",
+ " override def cancel(): Boolean = delegate.cancel(true)",
+ "",
+ " override def isCancelled: Boolean = delegate.isCancelled",
+ "",
+ " override def isDone: Boolean = delegate.isDone",
+ " }",
+ "",
+ " private final class RpcIteratorAdapter[ReqT <: AnyRef, RespT
<: AnyRef](",
+ " call: io.grpc.ClientCall[ReqT, RespT],",
+ " request: ReqT)",
+ " extends RpcIterator[RespT] {",
+ " private[this] val lock = new Object",
+ " private[this] var nextValue: RespT =
null.asInstanceOf[RespT]",
+ " private[this] var hasValue = false",
+ " private[this] var completed = false",
+ " private[this] var closed = false",
+ " private[this] var cancelled = false",
+ " private[this] var failure: Throwable = null",
+ "",
+ " start(request)",
+ "",
+ " override def hasNext: Boolean = {",
+ " var cancelCall = false",
+ " var thrown: Throwable = null",
+ " val result = lock.synchronized {",
+ " cancelCall = awaitValueLocked()",
+ " if (hasValue) true",
+ " else if (failure ne null) {",
+ " thrown = failure",
+ " false",
+ " } else false",
+ " }",
+ " if (cancelCall) {",
+ ' call.cancel("RpcIterator interrupted",
thrown)',
+ " }",
+ " if (thrown ne null) {",
+ " throw thrown",
+ " }",
+ " result",
+ " }",
+ "",
+ " override def next(): RespT = {",
+ " var requestMore = false",
+ " var cancelCall = false",
+ " var thrown: Throwable = null",
+ " var empty = false",
+ " val result = lock.synchronized {",
+ " cancelCall = awaitValueLocked()",
+ " if (!hasValue) {",
+ " if (failure ne null) {",
+ " thrown = failure",
+ " } else {",
+ " empty = true",
+ " }",
+ " null.asInstanceOf[RespT]",
+ " } else {",
+ " val value = nextValue",
+ " nextValue = null.asInstanceOf[RespT]",
+ " hasValue = false",
+ " requestMore = !closed && (failure eq null)",
+ " value",
+ " }",
+ " }",
+ " if (cancelCall) {",
+ ' call.cancel("RpcIterator interrupted", thrown)',
+ " }",
+ " if (thrown ne null) {",
+ " throw thrown",
+ " }",
+ " if (empty) {",
+ ' throw new NoSuchElementException("next on empty
RpcIterator")',
+ " }",
+ " if (requestMore) {",
+ " call.request(1)",
+ " }",
+ " result",
+ " }",
+ "",
+ " override def cancel(): Unit = {",
+ " var shouldCancel = false",
+ " lock.synchronized {",
+ " if (!cancelled && !completed) {",
+ " cancelled = true",
+ " shouldCancel = true",
+ " }",
+ " if (!closed) {",
+ " closed = true",
+ " hasValue = false",
+ " nextValue = null.asInstanceOf[RespT]",
+ " lock.notifyAll()",
+ " }",
+ " }",
+ " if (shouldCancel) {",
+ ' call.cancel("RpcIterator cancelled", null)',
+ " }",
+ " }",
+ "",
+ " override def isClosed: Boolean =",
+ " lock.synchronized { closed }",
+ "",
+ " private def start(request: ReqT): Unit = {",
+ " call.start(",
+ " new io.grpc.ClientCall.Listener[RespT] {",
+ " override def onMessage(message: RespT): Unit
=",
+ " lock.synchronized {",
+ " if (!closed) {",
+ " if (hasValue) {",
+ " failure =
io.grpc.Status.INTERNAL",
+ '
.withDescription("Received response without iterator demand")',
+ " .asRuntimeException()",
+ " closed = true",
+ " } else {",
+ " nextValue = message",
+ " hasValue = true",
+ " }",
+ " lock.notifyAll()",
+ " }",
+ " }",
+ "",
+ " override def onClose(",
+ " status: io.grpc.Status,",
+ " trailers: io.grpc.Metadata): Unit =",
+ " lock.synchronized {",
+ " if (!status.isOk && !cancelled) {",
+ " failure =
status.asRuntimeException(trailers)",
+ " } else {",
+ " completed = true",
+ " }",
+ " closed = true",
+ " lock.notifyAll()",
+ " }",
+ " },",
+ " new io.grpc.Metadata())",
+ " call.request(1)",
+ " call.sendMessage(request)",
+ " call.halfClose()",
+ " }",
+ "",
+ " private def awaitValueLocked(): Boolean = {",
+ " var cancelCall = false",
+ " while (!hasValue && !completed && (failure eq null)
&& !closed) {",
+ " try {",
+ " lock.wait()",
+ " } catch {",
+ " case e: InterruptedException =>",
+ " failure = interrupted(e)",
+ " closed = true",
+ " cancelled = true",
+ " cancelCall = true",
+ " }",
+ " }",
+ " cancelCall",
+ " }",
+ " }",
+ "",
+ " private def completePromise[A](",
+ " delegate:
com.google.common.util.concurrent.ListenableFuture[A],",
+ " promise: scala.concurrent.Promise[A]): Unit =",
+ " try {",
+ " promise.success(delegate.get())",
+ " } catch {",
+ " case e: java.util.concurrent.CancellationException
=>",
+ " promise.failure(e)",
+ " case e: java.util.concurrent.ExecutionException if
e.getCause ne null =>",
+ " promise.failure(e.getCause)",
+ " case e: InterruptedException =>",
+ " Thread.currentThread().interrupt()",
+ " promise.failure(e)",
+ " case scala.util.control.NonFatal(e) =>",
+ " promise.failure(e)",
+ " }",
+ "",
+ " private def marshaller[T <: AnyRef](",
+ " typ: Class[T]): io.grpc.MethodDescriptor.Marshaller[T] =",
+ " new ForyMarshaller(FORY, typ)",
+ "",
+ " private final class ForyMarshaller[T <: AnyRef](",
+ " fory: org.apache.fory.ThreadSafeFory,",
+ " typ: Class[T])",
+ " extends io.grpc.MethodDescriptor.Marshaller[T] {",
+ " override def stream(value: T): java.io.InputStream =",
+ " try {",
+ " new
KnownLengthByteArrayInputStream(fory.serialize(value))",
+ " } catch {",
+ " case scala.util.control.NonFatal(e) =>",
+ ' throw internalError("Fory serialization
failed", e)',
+ " }",
+ "",
+ " override def parse(stream: java.io.InputStream): T =",
+ " try {",
+ " fory.deserialize(readBytes(stream), typ)",
+ " } catch {",
+ " case e: java.io.IOException =>",
+ ' throw internalError("Fory deserialization
failed", e)',
+ " case scala.util.control.NonFatal(e) =>",
+ ' throw internalError("Fory deserialization
failed", e)',
+ " }",
+ " }",
+ "",
+ " private final class KnownLengthByteArrayInputStream(bytes:
Array[Byte])",
+ " extends java.io.ByteArrayInputStream(bytes)",
+ " with io.grpc.KnownLength",
+ "",
+ " private def readBytes(stream: java.io.InputStream):
Array[Byte] = {",
+ " if (stream.isInstanceOf[io.grpc.KnownLength]) {",
+ " val size = stream.available()",
+ " val bytes = new Array[Byte](size)",
+ " var offset = 0",
+ " while (offset < size) {",
+ " val read = stream.read(bytes, offset, size -
offset)",
+ " if (read == -1) {",
+ " throw new java.io.EOFException(",
+ ' "Expected " + size + " bytes, got " +
offset)',
+ " }",
+ " offset += read",
+ " }",
+ " bytes",
+ " } else {",
+ " readUnknownLengthBytes(stream)",
+ " }",
+ " }",
+ "",
+ " private def readUnknownLengthBytes(stream:
java.io.InputStream): Array[Byte] = {",
+ " val out = new java.io.ByteArrayOutputStream()",
+ " val buffer = new Array[Byte](8192)",
+ " var read = stream.read(buffer)",
+ " while (read != -1) {",
+ " out.write(buffer, 0, read)",
+ " read = stream.read(buffer)",
+ " }",
+ " out.toByteArray()",
+ " }",
+ "",
+ " private def internalError(",
+ " description: String,",
+ " cause: Throwable): io.grpc.StatusRuntimeException =",
+ " io.grpc.Status.INTERNAL",
+ " .withDescription(description)",
+ " .withCause(cause)",
+ " .asRuntimeException()",
+ "",
+ " private def interrupted(e: InterruptedException):
io.grpc.StatusRuntimeException = {",
+ " Thread.currentThread().interrupt()",
+ " io.grpc.Status.CANCELLED",
+ ' .withDescription("Interrupted while waiting for RPC
response")',
+ " .withCause(e)",
+ " .asRuntimeException()",
+ " }",
+ "",
+ " private def unimplemented(methodName: String):
io.grpc.StatusRuntimeException =",
+ " io.grpc.Status.UNIMPLEMENTED",
+ ' .withDescription("Method " + SERVICE_NAME + "/" +
methodName + " is unimplemented")',
+ " .asRuntimeException()",
+ "",
+ ]
+
+ def scala_grpc_type(self, named_type: NamedType) -> str:
+ type_name = self.schema.resolve_type_name(named_type.name)
+ return self.resolve_scala_type_name(type_name, None)
+
+ def scala_grpc_owner_name(self, service: Service) -> str:
+ return f"{service.name}Grpc"
+
+ def scala_grpc_service_base_name(self, service: Service) -> str:
+ return f"{service.name}ImplBase"
+
+ def scala_grpc_client_name(self, service: Service) -> str:
+ return f"{service.name}Client"
+
+ def scala_grpc_method_name(self, method: RpcMethod) -> str:
+ return self.to_camel_case(method.name)
+
+ def scala_grpc_method_descriptor(self, method: RpcMethod) -> str:
+ return f"{self.scala_grpc_method_name(method)}Method"
+
+ def scala_grpc_method_backing_field(self, method: RpcMethod) -> str:
+ return f"{self.scala_grpc_method_name(method)}MethodValue"
+
+ def scala_grpc_blocking_method_name(self, method: RpcMethod) -> str:
+ return f"{self.scala_grpc_method_name(method)}Blocking"
+
+ def scala_grpc_future_method_name(self, method: RpcMethod) -> str:
+ return f"{self.scala_grpc_method_name(method)}Future"
+
+ def scala_grpc_method_type(self, method: RpcMethod) -> str:
+ mode = streaming_mode(method)
+ if mode is StreamingMode.BIDIRECTIONAL:
+ return "BIDI_STREAMING"
+ if mode is StreamingMode.CLIENT_STREAMING:
+ return "CLIENT_STREAMING"
+ if mode is StreamingMode.SERVER_STREAMING:
+ return "SERVER_STREAMING"
+ return "UNARY"
+
+ def scala_string_literal(self, value: str) -> str:
+ return (
+ value.replace("\\", "\\\\")
+ .replace('"', '\\"')
+ .replace("\n", "\\n")
+ .replace("\r", "\\r")
+ .replace("\t", "\\t")
+ )
diff --git a/compiler/fory_compiler/tests/test_service_codegen.py
b/compiler/fory_compiler/tests/test_service_codegen.py
index f2e794ba7..1d76c729a 100644
--- a/compiler/fory_compiler/tests/test_service_codegen.py
+++ b/compiler/fory_compiler/tests/test_service_codegen.py
@@ -23,7 +23,7 @@ from typing import Dict, Tuple, Type
import pytest
-from fory_compiler.cli import compile_file, resolve_imports
+from fory_compiler.cli import compile_file, resolve_imports,
validate_scala_generation
from fory_compiler.frontend.fdl.lexer import Lexer
from fory_compiler.frontend.fdl.parser import Parser
from fory_compiler.frontend.fbs.lexer import Lexer as FbsLexer
@@ -40,6 +40,7 @@ from fory_compiler.generators.java import JavaGenerator
from fory_compiler.generators.kotlin import KotlinGenerator
from fory_compiler.generators.python import PythonGenerator
from fory_compiler.generators.rust import RustGenerator
+from fory_compiler.generators.scala import ScalaGenerator
from fory_compiler.generators.swift import SwiftGenerator
from fory_compiler.ir.ast import Schema
from fory_compiler.ir.validator import SchemaValidator
@@ -53,6 +54,7 @@ GENERATOR_CLASSES: Tuple[Type[BaseGenerator], ...] = (
GoGenerator,
CSharpGenerator,
SwiftGenerator,
+ ScalaGenerator,
KotlinGenerator,
)
@@ -138,6 +140,7 @@ def
test_generate_services_returns_empty_list_for_unsupported_generators():
PythonGenerator,
GoGenerator,
RustGenerator,
+ ScalaGenerator,
KotlinGenerator,
):
continue
@@ -219,6 +222,37 @@ def
test_kotlin_grpc_service_codegen_contains_fory_marshaller():
assert "MessageLite" not in content
+def test_scala_grpc_marshaller():
+ schema = parse_fdl(_GREETER_WITH_SERVICE)
+ files = generate_service_files(schema, ScalaGenerator)
+ assert set(files) == {"demo/greeter/GreeterGrpc.scala"}
+ content = files["demo/greeter/GreeterGrpc.scala"]
+ assert "object GreeterGrpc" in content
+ assert 'SERVICE_NAME: String = "demo.greeter.Greeter"' in content
+ assert "private val FORY: org.apache.fory.ThreadSafeFory" in content
+ assert "GreeterForyModule.getFory" in content
+ assert "import org.apache.fory.scala.rpc.{RpcFuture, RpcIterator}" in
content
+ assert (
+ "private lazy val sayHelloMethodValue:
io.grpc.MethodDescriptor[HelloRequest, HelloReply]"
+ in content
+ )
+ assert "def sayHello(request: HelloRequest): RpcFuture[HelloReply]" in
content
+ assert "def sayHelloBlocking(request: HelloRequest): HelloReply" in content
+ assert (
+ "def sayHelloFuture(\n request: HelloRequest):
com.google.common.util.concurrent.ListenableFuture[HelloReply]"
+ in content
+ )
+ assert "protected override def build(" in content
+ assert "private final class ForyMarshaller[T <: AnyRef]" in content
+ assert "fory.serialize(value)" in content
+ assert "fory.deserialize(readBytes(stream), typ)" in content
+ assert "with io.grpc.KnownLength" in content
+ assert "ProtoUtils" not in content
+ assert "MessageLite" not in content
+ assert "ScalaPB" not in content
+ assert "org.apache.fory.scala.grpc" not in content
+
+
def test_grpc_streaming_method_shapes():
schema = parse_fdl(
dedent(
@@ -287,6 +321,29 @@ def test_grpc_streaming_method_shapes():
assert "io.grpc.kotlin.ServerCalls.clientStreamingServerMethodDefinition"
in kotlin
assert "io.grpc.kotlin.ServerCalls.bidiStreamingServerMethodDefinition" in
kotlin
+ scala = next(iter(generate_service_files(schema, ScalaGenerator).values()))
+ assert "io.grpc.MethodDescriptor.MethodType.UNARY" in scala
+ assert "io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING" in scala
+ assert "io.grpc.MethodDescriptor.MethodType.CLIENT_STREAMING" in scala
+ assert "io.grpc.MethodDescriptor.MethodType.BIDI_STREAMING" in scala
+ assert "def unary(request: Req): RpcFuture[Res]" in scala
+ assert "def server(request: Req): RpcIterator[Res]" in scala
+ assert (
+ "def client(\n responseObserver:
io.grpc.stub.StreamObserver[Res]\n ): io.grpc.stub.StreamObserver[Req]"
+ in scala
+ )
+ assert (
+ "def bidi(\n responseObserver:
io.grpc.stub.StreamObserver[Payload]\n ):
io.grpc.stub.StreamObserver[Payload]"
+ in scala
+ )
+ assert "io.grpc.stub.ClientCalls.asyncServerStreamingCall" in scala
+ assert "io.grpc.stub.ClientCalls.asyncClientStreamingCall" in scala
+ assert "io.grpc.stub.ClientCalls.asyncBidiStreamingCall" in scala
+ assert "new RpcIteratorAdapter(" in scala
+ assert "call.request(1)" in scala
+ assert "call.sendMessage(request)" in scala
+ assert "call.halfClose()" in scala
+
def test_go_grpc_service_codegen():
schema = parse_fdl(_GREETER_WITH_SERVICE)
@@ -432,6 +489,13 @@ def
test_grpc_services_use_imported_java_type_references(tmp_path: Path):
assert "marshaller(common.Shared::class.java)" in kotlin
assert "public open suspend fun get(request: common.Shared): Local" in
kotlin
+ scala_files = generate_service_files(schema, ScalaGenerator)
+ assert set(scala_files) == {"api/ApiServiceGrpc.scala"}
+ scala = scala_files["api/ApiServiceGrpc.scala"]
+ assert "io.grpc.MethodDescriptor[common.Shared, Local]" in scala
+ assert "marshaller(classOf[common.Shared])" in scala
+ assert "def get(request: common.Shared): RpcFuture[Local]" in scala
+
def test_proto_grpc_services_use_imported_qualified_type_references(tmp_path:
Path):
common = tmp_path / "common.proto"
@@ -490,6 +554,11 @@ def
test_proto_grpc_services_use_imported_qualified_type_references(tmp_path: Pa
assert "io.grpc.MethodDescriptor<common.Shared, Local>" in kotlin
assert "marshaller(common.Shared::class.java)" in kotlin
+ scala_files = generate_service_files(schema, ScalaGenerator)
+ scala = scala_files["api/ApiServiceGrpc.scala"]
+ assert "io.grpc.MethodDescriptor[common.Shared, Local]" in scala
+ assert "marshaller(classOf[common.Shared])" in scala
+
def test_proto_grpc_absolute_rpc_type_uses_package_type_not_nested_shadow():
schema = parse_proto(
@@ -530,6 +599,11 @@ def
test_proto_grpc_absolute_rpc_type_uses_package_type_not_nested_shadow():
assert "io.grpc.MethodDescriptor<Request, Response>" in kotlin
assert "io.grpc.MethodDescriptor<demo.Request, Response>" not in kotlin
+ scala_files = generate_service_files(schema, ScalaGenerator)
+ scala = scala_files["demo/ApiServiceGrpc.scala"]
+ assert "io.grpc.MethodDescriptor[Request, Response]" in scala
+ assert "io.grpc.MethodDescriptor[demo.Request, Response]" not in scala
+
def test_proto_grpc_absolute_rpc_type_prefers_longest_package_prefix(tmp_path:
Path):
common = tmp_path / "common.proto"
@@ -586,6 +660,11 @@ def
test_proto_grpc_absolute_rpc_type_prefers_longest_package_prefix(tmp_path: P
assert "io.grpc.MethodDescriptor<alpha.beta.C, alpha.beta.C>" in kotlin
assert "io.grpc.MethodDescriptor<beta.C, beta.C>" not in kotlin
+ scala_files = generate_service_files(schema, ScalaGenerator)
+ scala = scala_files["alpha/ApiServiceGrpc.scala"]
+ assert "io.grpc.MethodDescriptor[alpha.beta.C, alpha.beta.C]" in scala
+ assert "io.grpc.MethodDescriptor[beta.C, beta.C]" not in scala
+
def test_java_grpc_service_class_collision_fails():
schema = parse_fdl(
@@ -634,6 +713,64 @@ def test_kotlin_grpc_service_class_collision_fails():
KotlinGenerator(schema, GeneratorOptions(output_dir=Path("/tmp"),
grpc=True))
+def test_scala_grpc_class_collision():
+ schema = parse_fdl(
+ dedent(
+ """
+ package demo.collision;
+
+ message GreeterGrpc {}
+ message Req {}
+ message Res {}
+
+ service Greeter {
+ rpc Call (Req) returns (Res);
+ }
+ """
+ )
+ )
+ generator = ScalaGenerator(
+ schema, GeneratorOptions(output_dir=Path("/tmp"), grpc=True)
+ )
+ with pytest.raises(ValueError, match="Scala generated file path
collision"):
+ generator.generate_services()
+
+
+def test_scala_grpc_preflight_collision(tmp_path: Path, capsys):
+ common = tmp_path / "common.fdl"
+ common.write_text(
+ dedent(
+ """
+ package demo.collision;
+
+ message GreeterGrpc {}
+ """
+ )
+ )
+ main = tmp_path / "main.fdl"
+ main.write_text(
+ dedent(
+ """
+ package demo.collision;
+
+ import "common.fdl";
+
+ message Req {}
+ message Res {}
+
+ service Greeter {
+ rpc Call (Req) returns (Res);
+ }
+ """
+ )
+ )
+
+ assert validate_scala_generation([main], [tmp_path], grpc=True) is False
+ err = capsys.readouterr().err
+ assert "Scala generated file path collision" in err
+ assert "GreeterGrpc.scala" in err
+
+
def test_java_grpc_service_class_collision_with_imported_type_fails(tmp_path:
Path):
common = tmp_path / "common.fdl"
common.write_text(
@@ -786,6 +923,44 @@ def test_grpc_method_name_collisions_fail():
else:
raise AssertionError("Expected Kotlin gRPC method name collision")
+ scala_generator = ScalaGenerator(
+ schema, GeneratorOptions(output_dir=Path("/tmp"), grpc=True)
+ )
+ try:
+ scala_generator.generate_services()
+ except ValueError as e:
+ assert "Scala gRPC method name collision" in str(e)
+ assert "Foo and foo" in str(e)
+ else:
+ raise AssertionError("Expected Scala gRPC method name collision")
+
+
+def test_scala_grpc_method_scopes():
+ schema = parse_fdl(
+ dedent(
+ """
+ package demo.scope;
+
+ message Req {}
+ message Res {}
+
+ service Greeter {
+ rpc Cancel (Req) returns (Res);
+ rpc Close (Req) returns (stream Res);
+ rpc ReadBytes (Req) returns (Res);
+ rpc CompletePromise (Req) returns (Res);
+ }
+ """
+ )
+ )
+
+ files = generate_service_files(schema, ScalaGenerator)
+ content = files["demo/scope/GreeterGrpc.scala"]
+ assert "def cancel(request: Req): RpcFuture[Res]" in content
+ assert "def close(request: Req): RpcIterator[Res]" in content
+ assert "def readBytes(request: Req): RpcFuture[Res]" in content
+ assert "def completePromise(request: Req): RpcFuture[Res]" in content
+
def test_java_python_grpc_method_keywords_are_safe_names():
schema = parse_fdl(
@@ -819,6 +994,11 @@ def test_java_python_grpc_method_keywords_are_safe_names():
assert "public suspend fun `class`(" in kotlin
assert "implementation = ::`class`" in kotlin
+ scala = next(iter(generate_service_files(schema, ScalaGenerator).values()))
+ assert "def `class`(request: Req): RpcFuture[Res]" in scala
+ assert "def `class`(request: Req, responseObserver:" in scala
+ assert 'SERVICE_NAME,\n "Class"' in scala
+
def test_python_grpc_service_registration_collisions_fail():
schema = parse_fdl(
@@ -907,10 +1087,14 @@ def test_proto_and_fbs_grpc_service_codegen():
)
proto_java = generate_service_files(proto_schema, JavaGenerator)
proto_python = generate_service_files(proto_schema, PythonGenerator)
+ proto_scala = generate_service_files(proto_schema, ScalaGenerator)
assert "demo/proto/ProtoSvcGrpc.java" in proto_java
assert "demo_proto_grpc.py" in proto_python
+ assert "demo/proto/ProtoSvcGrpc.scala" in proto_scala
assert "MethodType.SERVER_STREAMING" in
proto_java["demo/proto/ProtoSvcGrpc.java"]
assert "channel.unary_stream(" in proto_python["demo_proto_grpc.py"]
+ assert "MethodType.SERVER_STREAMING" in
proto_scala["demo/proto/ProtoSvcGrpc.scala"]
+ assert "RpcIterator[Res]" in proto_scala["demo/proto/ProtoSvcGrpc.scala"]
fbs_schema = parse_fbs(
dedent(
@@ -928,10 +1112,16 @@ def test_proto_and_fbs_grpc_service_codegen():
)
fbs_java = generate_service_files(fbs_schema, JavaGenerator)
fbs_python = generate_service_files(fbs_schema, PythonGenerator)
+ fbs_scala = generate_service_files(fbs_schema, ScalaGenerator)
assert "demo/fbs/FbsSvcGrpc.java" in fbs_java
assert "demo_fbs_grpc.py" in fbs_python
+ assert "demo/fbs/FbsSvcGrpc.scala" in fbs_scala
assert 'SERVICE_NAME = "demo.fbs.FbsSvc"' in
fbs_java["demo/fbs/FbsSvcGrpc.java"]
assert '"/demo.fbs.FbsSvc/Call"' in fbs_python["demo_fbs_grpc.py"]
+ assert (
+ 'SERVICE_NAME: String = "demo.fbs.FbsSvc"'
+ in fbs_scala["demo/fbs/FbsSvcGrpc.scala"]
+ )
def test_service_schema_produces_one_file_per_message_per_language():
@@ -943,10 +1133,20 @@ def
test_service_schema_produces_one_file_per_message_per_language():
)
-def test_compile_service_schema_with_grpc_flag(tmp_path: Path, capsys):
+def test_grpc_flag_compiles_services(tmp_path: Path, capsys):
example_path = Path(__file__).resolve().parents[2] / "examples" /
"service.fdl"
lang_dirs = {}
- for lang in ("java", "python", "rust", "go", "cpp", "csharp", "swift",
"kotlin"):
+ for lang in (
+ "java",
+ "python",
+ "rust",
+ "go",
+ "cpp",
+ "csharp",
+ "swift",
+ "scala",
+ "kotlin",
+ ):
lang_dirs[lang] = tmp_path / lang
ok = compile_file(example_path, lang_dirs, grpc=True, generated_outputs={})
output = capsys.readouterr().out
@@ -960,6 +1160,7 @@ def test_compile_service_schema_with_grpc_flag(tmp_path:
Path, capsys):
assert output.count("demo_greeter_grpc.go") == 1
assert (lang_dirs["rust"] / "demo_greeter_service.rs").exists()
assert (lang_dirs["rust"] / "demo_greeter_service_grpc.rs").exists()
+ assert (lang_dirs["scala"] / "demo" / "greeter" /
"GreeterGrpc.scala").exists()
assert (lang_dirs["kotlin"] / "demo" / "greeter" /
"GreeterGrpcKt.kt").exists()
diff --git a/docs/compiler/compiler-guide.md b/docs/compiler/compiler-guide.md
index 683bd3417..cd1701070 100644
--- a/docs/compiler/compiler-guide.md
+++ b/docs/compiler/compiler-guide.md
@@ -141,24 +141,25 @@ foryc schema.fdl --output ./src/generated
foryc user.fdl order.fdl product.fdl --output ./generated
```
-**Compile a simple schema containing service definitions (Java + Python + Rust
+ Kotlin models):**
+**Compile a simple schema containing service definitions (Java + Python + Rust
+ Scala + Kotlin models):**
```bash
-foryc compiler/examples/service.fdl --java_out=./generated/java
--python_out=./generated/python --rust_out=./generated/rust
--kotlin_out=./generated/kotlin
+foryc compiler/examples/service.fdl --java_out=./generated/java
--python_out=./generated/python --rust_out=./generated/rust
--scala_out=./generated/scala --kotlin_out=./generated/kotlin
```
-**Generate Java, Python, Rust, and Kotlin gRPC service companions:**
+**Generate Java, Python, Rust, Scala, and Kotlin gRPC service companions:**
```bash
-foryc compiler/examples/service.fdl --java_out=./generated/java
--python_out=./generated/python --rust_out=./generated/rust
--kotlin_out=./generated/kotlin --grpc
+foryc compiler/examples/service.fdl --java_out=./generated/java
--python_out=./generated/python --rust_out=./generated/rust
--scala_out=./generated/scala --kotlin_out=./generated/kotlin --grpc
```
The generated gRPC service code uses Fory to serialize request and response
payloads. Java output imports grpc-java APIs, Python output imports `grpc`, and
-Rust output imports `tonic` and `bytes`. Kotlin output imports grpc-java and
-grpc-kotlin APIs and uses coroutine stubs. Applications that compile or run
-those generated service files must provide their own gRPC dependencies. Fory
-packages do not add a hard gRPC dependency for this feature.
+Rust output imports `tonic` and `bytes`. Scala output imports grpc-java APIs.
+Kotlin output imports grpc-java and grpc-kotlin APIs and uses coroutine stubs.
+Applications that compile or run those generated service files must provide
+their own gRPC dependencies. Fory packages do not add a hard gRPC dependency
for
+this feature.
**Use import search paths:**
@@ -194,6 +195,9 @@ foryc schema.fdl --java_out=./gen/java -I proto/ -I common/
# Generate Scala 3 code to a specific directory
foryc schema.fdl --scala_out=./src/main/scala
+# Generate Scala 3 models and gRPC service companions
+foryc service.fdl --scala_out=./src/main/scala --grpc
+
# Generate Kotlin code to a specific directory
foryc schema.fdl --kotlin_out=./src/main/kotlin
```
@@ -415,6 +419,7 @@ generated/
├── User.scala
├── Status.scala
├── Animal.scala
+ ├── ExampleServiceGrpc.scala
└── ExampleForyModule.scala
```
@@ -425,6 +430,8 @@ generated/
- Enums use Scala 3 `enum`
- Unions use Scala 3 ADT `enum` with `@ForyUnion`, `@ForyCase`, and an
`Unknown`
- Schema module object included
+- With `--grpc`, one `<ServiceName>Grpc.scala` service companion is generated
+ per local service definition
### Kotlin
diff --git a/docs/compiler/flatbuffers-idl.md b/docs/compiler/flatbuffers-idl.md
index fa29946ac..203a64c40 100644
--- a/docs/compiler/flatbuffers-idl.md
+++ b/docs/compiler/flatbuffers-idl.md
@@ -126,8 +126,8 @@ message Container {
FlatBuffers `rpc_service` definitions are translated to Fory services. With
`--grpc`, the compiler emits gRPC service companions for supported outputs such
-as Java, Python, Go, Rust, and Kotlin. These companions use Fory serialization
for
-request and response payloads.
+as Java, Python, Go, Rust, Scala, and Kotlin. These companions use Fory
+serialization for request and response payloads.
```fbs
rpc_service SearchService {
@@ -137,7 +137,7 @@ rpc_service SearchService {
```
```bash
-foryc api.fbs --java_out=./generated/java --python_out=./generated/python
--rust_out=./generated/rust --kotlin_out=./generated/kotlin --grpc
+foryc api.fbs --java_out=./generated/java --python_out=./generated/python
--rust_out=./generated/rust --scala_out=./generated/scala
--kotlin_out=./generated/kotlin --grpc
```
Generated service code imports grpc APIs, so applications must provide
grpc-java,
diff --git a/docs/compiler/generated-code.md b/docs/compiler/generated-code.md
index d197fb86f..dfdc403a0 100644
--- a/docs/compiler/generated-code.md
+++ b/docs/compiler/generated-code.md
@@ -1589,6 +1589,92 @@ object AddressbookForyModule extends
org.apache.fory.ForyModule {
}
```
+### gRPC Service Companions
+
+When a schema contains services and the compiler is run with `--grpc`, Scala
+generation emits one `<ServiceName>Grpc.scala` companion per local service
+definition. The companion lives in the same Scala package as the generated
+models and schema module.
+
+For a service such as:
+
+```protobuf
+service AddressBookService {
+ rpc Lookup (Person) returns (AddressBook);
+ rpc Watch (Person) returns (stream AddressBook);
+ rpc Upload (stream Person) returns (AddressBook);
+ rpc Chat (stream Person) returns (stream AddressBook);
+}
+```
+
+the generated companion contains:
+
+- `SERVICE_NAME` and grpc-java method descriptors
+- `AddressBookServiceImplBase` for server implementations
+- `AddressBookServiceClient` for client calls
+- Fory-backed grpc-java marshallers for request and response payloads
+
+The generated Scala client keeps grpc-java available per method while adding
+Scala-friendly convenience methods for the shapes where a direct Scala handle
+can preserve the needed lifecycle controls:
+
+| RPC shape | Scala convenience
method | grpc-java-style method |
+| ------------------------------------------------------- |
----------------------------------- |
------------------------------------------------ |
+| `rpc Lookup (Person) returns (AddressBook)` | `lookup(request):
RpcFuture[Resp]` | async observer, blocking, and `ListenableFuture` |
+| `rpc Watch (Person) returns (stream AddressBook)` | `watch(request):
RpcIterator[Resp]` | async observer and blocking iterator |
+| `rpc Upload (stream Person) returns (AddressBook)` | None
| request `StreamObserver` |
+| `rpc Chat (stream Person) returns (stream AddressBook)` | None
| request and response `StreamObserver` |
+
+Unary client convenience methods return `org.apache.fory.scala.rpc.RpcFuture`:
+
+```scala
+val client = AddressBookServiceGrpc.newClient(channel)
+val call = client.lookup(person)
+call.asFuture.foreach(handleAddressBook)(scala.concurrent.ExecutionContext.global)
+```
+
+Server-streaming client convenience methods return
+`org.apache.fory.scala.rpc.RpcIterator`:
+
+```scala
+val stream = client.watch(person)
+try {
+ while (stream.hasNext) {
+ handleAddressBook(stream.next())
+ }
+} finally {
+ stream.close()
+}
+```
+
+Close or cancel the `RpcIterator` when the client stops before consuming the
+whole stream. The generated adapter cancels the underlying gRPC call so the
+server is not left writing a response stream the client no longer reads.
+
+Client-streaming and bidirectional methods use grpc-java `StreamObserver` APIs:
+
+```scala
+val requestStream = client.upload(
+ new io.grpc.stub.StreamObserver[AddressBook] {
+ override def onNext(value: AddressBook): Unit = handleAddressBook(value)
+ override def onError(t: Throwable): Unit = handleError(t)
+ override def onCompleted(): Unit = ()
+ }
+)
+requestStream.onNext(person)
+requestStream.onCompleted()
+```
+
+Server implementations mirror grpc-java. Unary methods can override the direct
+request-to-response method generated by Scala, but streaming methods override
+observer-based methods and must call `onNext`, `onError`, and `onCompleted`
+according to grpc-java lifecycle rules.
+
+Applications compiling generated Scala gRPC companions must provide grpc-java
+dependencies such as `grpc-api`, `grpc-stub`, and a transport like
+`grpc-netty-shaded`. The `fory-scala` artifact does not add grpc-java as a hard
+dependency.
+
## Cross-Language Notes
### Type ID Behavior
diff --git a/docs/compiler/index.md b/docs/compiler/index.md
index f9fea15b4..e635bcde0 100644
--- a/docs/compiler/index.md
+++ b/docs/compiler/index.md
@@ -23,7 +23,7 @@ Fory IDL is a schema definition language for Apache Fory that
enables type-safe
cross-language serialization. Define your data structures once and generate
native data structure code for Java, Python, C++, Go, Rust,
JavaScript/TypeScript, C#, Swift, Dart, Scala, and Kotlin. Fory IDL can also
-describe RPC services; for Java, Python, Go, Rust, and Kotlin, the compiler can
+describe RPC services; for Java, Python, Go, Rust, Scala, and Kotlin, the
compiler can
generate gRPC service companions that use Fory serialization for request and
response payloads.
@@ -88,10 +88,10 @@ service AnimalService {
}
```
-Generate Java, Python, Rust, and Kotlin models plus gRPC service companions
with:
+Generate Java, Python, Rust, Scala, and Kotlin models plus gRPC service
companions with:
```bash
-foryc animals.fdl --java_out=./generated/java --python_out=./generated/python
--rust_out=./generated/rust --kotlin_out=./generated/kotlin --grpc
+foryc animals.fdl --java_out=./generated/java --python_out=./generated/python
--rust_out=./generated/rust --scala_out=./generated/scala
--kotlin_out=./generated/kotlin --grpc
```
The generated service code uses normal gRPC APIs, but request and response
diff --git a/docs/compiler/protobuf-idl.md b/docs/compiler/protobuf-idl.md
index 4ce825e37..acf7f7f5a 100644
--- a/docs/compiler/protobuf-idl.md
+++ b/docs/compiler/protobuf-idl.md
@@ -41,17 +41,17 @@ how protobuf concepts map to Fory, and how to use
protobuf-only Fory extension o
## Protobuf vs Fory at a Glance
-| Aspect | Protocol Buffers | Fory
|
-| ------------------ | ----------------------------- |
------------------------------------------ |
-| Primary purpose | RPC/message contracts | High-performance object
serialization |
-| Encoding model | Tag-length-value | Fory binary protocol
|
-| Reference tracking | Not built-in | First-class (`ref`)
|
-| Circular refs | Not supported | Supported
|
-| Unknown fields | Preserved | Not preserved
|
-| Generated types | Protobuf-specific model types | Native language
constructs |
-| gRPC ecosystem | Native |
Java/Python/Go/Rust/Kotlin service codegen |
-
-Fory can generate Java, Python, Go, Rust, and Kotlin gRPC service companions
with
+| Aspect | Protocol Buffers | Fory
|
+| ------------------ | ----------------------------- |
------------------------------------------------ |
+| Primary purpose | RPC/message contracts | High-performance object
serialization |
+| Encoding model | Tag-length-value | Fory binary protocol
|
+| Reference tracking | Not built-in | First-class (`ref`)
|
+| Circular refs | Not supported | Supported
|
+| Unknown fields | Preserved | Not preserved
|
+| Generated types | Protobuf-specific model types | Native language
constructs |
+| gRPC ecosystem | Native |
Java/Python/Go/Rust/Scala/Kotlin service codegen |
+
+Fory can generate Java, Python, Go, Rust, Scala, and Kotlin gRPC service
companions with
`--grpc`. Those services use normal gRPC transports but serialize request and
response payloads with Fory rather than protobuf. For broad gRPC ecosystem
tooling, schema reflection, and protobuf-native interceptors, protobuf remains
@@ -314,12 +314,13 @@ languages.
For supported service outputs, add `--grpc` to emit gRPC companion code:
```bash
-foryc api.proto --java_out=./generated/java --python_out=./generated/python
--rust_out=./generated/rust --kotlin_out=./generated/kotlin --grpc
+foryc api.proto --java_out=./generated/java --python_out=./generated/python
--rust_out=./generated/rust --scala_out=./generated/scala
--kotlin_out=./generated/kotlin --grpc
```
Generated Java service files compile against grpc-java, generated Python
service
modules import `grpc`, generated Rust service files import `tonic` and `bytes`,
-and generated Kotlin service files compile against grpc-java and grpc-kotlin.
+generated Scala service files compile against grpc-java, and generated Kotlin
+service files compile against grpc-java and grpc-kotlin.
Add those dependencies in your application build; Fory packages do not add gRPC
as a hard dependency. Protobuf `oneof` fields are translated to Fory union
fields inside request and response messages. Direct union RPC request or
diff --git a/docs/compiler/schema-idl.md b/docs/compiler/schema-idl.md
index 40f082108..4fcf7c52f 100644
--- a/docs/compiler/schema-idl.md
+++ b/docs/compiler/schema-idl.md
@@ -908,7 +908,7 @@ union_field := ['repeated'] field_type IDENTIFIER '='
INTEGER [field_options] ';
Services define RPC method contracts in Fory IDL. They are optional: schemas
with services still generate the normal data model types, and gRPC service code
is generated only when the compiler is run with `--grpc` for supported language
-outputs such as Java, Python, Go, Rust, and Kotlin.
+outputs such as Java, Python, Go, Rust, Scala, and Kotlin.
```protobuf
message GetPetRequest [id=200] {
diff --git a/docs/guide/java/grpc-support.md b/docs/guide/java/grpc-support.md
index 4a233852d..7bebb010f 100644
--- a/docs/guide/java/grpc-support.md
+++ b/docs/guide/java/grpc-support.md
@@ -30,8 +30,9 @@ Fory payload encoding. Use standard protobuf gRPC code
generation when your API
must be consumed by generic protobuf clients, reflection tools, or components
that expect protobuf message bytes.
-For Kotlin coroutine stubs and service bases, see
-[Kotlin gRPC Support](../kotlin/grpc-support.md).
+For Scala generated grpc-java companions, see
+[Scala gRPC Support](../scala/grpc-support.md). For Kotlin coroutine stubs and
+service bases, see [Kotlin gRPC Support](../kotlin/grpc-support.md).
## Add Dependencies
diff --git a/docs/guide/scala/grpc-support.md b/docs/guide/scala/grpc-support.md
new file mode 100644
index 000000000..0b97929db
--- /dev/null
+++ b/docs/guide/scala/grpc-support.md
@@ -0,0 +1,397 @@
+---
+title: gRPC Support
+sidebar_position: 6
+id: grpc_support
+license: |
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+---
+
+Fory can generate Scala 3 gRPC service companions for schemas that define
+services. The generated service code uses normal grpc-java channels, servers,
+deadlines, status codes, interceptors, and transport security, while request
+and response objects are serialized with Fory instead of protobuf.
+
+Use this mode when both sides of the RPC are generated from the same Fory IDL,
+protobuf IDL, or FlatBuffers IDL and you want gRPC transport semantics with
+Fory payload encoding. Use standard protobuf gRPC code generation when your API
+must be consumed by generic protobuf clients, reflection tools, or components
+that expect protobuf message bytes.
+
+## Add Dependencies
+
+Generated Scala service files compile against grpc-java. The `fory-scala`
+artifact does not add gRPC as a hard dependency, so add grpc-java dependencies
+in your application build and align the version with the rest of your service
+stack.
+
+```sbt
+libraryDependencies ++= Seq(
+ "org.apache.fory" %% "fory-scala" % "<fory-version>",
+ "io.grpc" % "grpc-api" % "<grpc-version>",
+ "io.grpc" % "grpc-stub" % "<grpc-version>",
+ "io.grpc" % "grpc-netty-shaded" % "<grpc-version>"
+)
+```
+
+Generated Scala models and gRPC service companions are Scala 3 source. The
+`fory-scala` artifact remains cross-built for Scala 2.13 and Scala 3, and the
+dependency-free `org.apache.fory.scala.rpc` handle traits are available from
+the shared artifact.
+
+## Define a Service
+
+Service definitions can come from Fory IDL, protobuf IDL, or FlatBuffers
+`rpc_service` definitions. A Fory IDL service looks like this:
+
+```protobuf
+package demo.greeter;
+
+message HelloRequest {
+ string name = 1;
+}
+
+message HelloReply {
+ string reply = 1;
+}
+
+service Greeter {
+ rpc SayHello (HelloRequest) returns (HelloReply);
+}
+```
+
+Generate Scala model and gRPC companion code with `--grpc`:
+
+```bash
+foryc service.fdl --scala_out=./generated/scala --grpc
+```
+
+For this schema, the Scala generator emits:
+
+| File | Purpose |
+| ------------------------- | -------------------------------------------- |
+| `HelloRequest.scala` | Fory model type for the request |
+| `HelloReply.scala` | Fory model type for the response |
+| `GreeterForyModule.scala` | Fory registration module for generated types |
+| `GreeterGrpc.scala` | grpc-java service base, client, and codecs |
+
+## Implement a Server
+
+Extend the generated `GreeterGrpc.GreeterImplBase` class and register it with a
+standard grpc-java `Server`. Unary RPCs can be implemented with a direct
+request-to-response method:
+
+```scala
+package demo.greeter
+
+import io.grpc.ServerBuilder
+
+final class GreeterService extends GreeterGrpc.GreeterImplBase {
+ override def sayHello(request: HelloRequest): HelloReply =
+ HelloReply(s"Hello, ${request.name}")
+}
+
+@main def runServer(): Unit = {
+ val server = ServerBuilder
+ .forPort(50051)
+ .addService(new GreeterService)
+ .build()
+ .start()
+ server.awaitTermination()
+}
+```
+
+Generated request and response types are registered by the generated code, so
+service implementations do not perform manual serializer registration.
+
+## Create a Client
+
+Use the generated client with an ordinary grpc-java channel:
+
+```scala
+package demo.greeter
+
+import io.grpc.ManagedChannelBuilder
+import scala.concurrent.Await
+import scala.concurrent.duration.DurationInt
+
+@main def runClient(): Unit = {
+ val channel = ManagedChannelBuilder
+ .forAddress("localhost", 50051)
+ .usePlaintext()
+ .build()
+ try {
+ val client = GreeterGrpc.newClient(channel)
+ val call = client.sayHello(HelloRequest("Fory"))
+ val reply = Await.result(call.asFuture, 30.seconds)
+ println(reply.reply)
+ } finally {
+ channel.shutdownNow()
+ }
+}
+```
+
+Unary Scala-friendly methods return `RpcFuture[A]`. Use `asFuture` for Scala
+composition, and call `cancel()` when the RPC should be cancelled before it
+completes. The same generated client also exposes grpc-java-style per-method
+variants such as observer-based async calls, blocking calls, and
+`ListenableFuture` unary calls.
+
+## Streaming RPCs
+
+Fory service definitions can use the same gRPC streaming shapes as grpc-java:
+
+```protobuf
+service Greeter {
+ rpc SayHello (HelloRequest) returns (HelloReply);
+ rpc LotsOfReplies (HelloRequest) returns (stream HelloReply);
+ rpc LotsOfGreetings (stream HelloRequest) returns (HelloReply);
+ rpc Chat (stream HelloRequest) returns (stream HelloReply);
+}
+```
+
+Generated Scala methods use these shapes:
+
+| IDL shape | Scala client convenience | grpc-java-style methods
|
+| ----------------------- | ------------------------ |
------------------------------------------------ |
+| Unary | `RpcFuture[Resp]` | Async observer,
blocking, and `ListenableFuture` |
+| Server streaming | `RpcIterator[Resp]` | Async observer and
blocking iterator |
+| Client streaming | None | `StreamObserver`
request stream |
+| Bidirectional streaming | None | `StreamObserver`
request and response streams |
+
+The Scala-friendly convenience layer covers the cases where a direct Scala
+handle can keep the important lifecycle controls. Unary calls use
+`RpcFuture[A]` so callers can compose with Scala `Future` without losing
+cancellation. Server-streaming calls use `RpcIterator[A]` so callers can
consume
+responses with the normal Scala `Iterator` contract while still being able to
+close the underlying RPC. Client-streaming and bidirectional streaming stay on
+grpc-java `StreamObserver` APIs because the request stream lifecycle,
+completion, cancellation, and flow-control rules are the grpc-java rules.
+
+### Server-Streaming Clients
+
+Use the Scala-friendly method when the client wants pull-style consumption:
+
+```scala
+val stream = client.lotsOfReplies(HelloRequest("Fory"))
+try {
+ while (stream.hasNext) {
+ val reply = stream.next()
+ println(reply.reply)
+ }
+} finally {
+ stream.close()
+}
+```
+
+`RpcIterator[A]` extends Scala `Iterator[A]` and `AutoCloseable`. A fully
+consumed stream closes when the server completes it. If the caller stops early,
+call `close()` or `cancel()` to release the gRPC call and notify the server
that
+the response stream is no longer needed.
+
+Use the observer overload when the client wants grpc-java async callbacks:
+
+```scala
+import io.grpc.stub.StreamObserver
+
+client.lotsOfReplies(
+ HelloRequest("Fory"),
+ new StreamObserver[HelloReply] {
+ override def onNext(value: HelloReply): Unit =
+ println(value.reply)
+
+ override def onError(t: Throwable): Unit =
+ t.printStackTrace()
+
+ override def onCompleted(): Unit =
+ println("done")
+ }
+)
+```
+
+The generated client also exposes a blocking grpc-java-style iterator through
+`lotsOfRepliesBlocking(request)`. Prefer the Scala-friendly `RpcIterator` when
+you need early cancellation; use the blocking iterator only when matching an
+existing grpc-java workflow.
+
+### Client-Streaming Clients
+
+For client-streaming RPCs, the generated method accepts a response observer and
+returns the request observer. Send every request with `onNext`, then call
+`onCompleted` exactly once when the client has finished sending:
+
+```scala
+import io.grpc.stub.StreamObserver
+
+val requests = client.lotsOfGreetings(
+ new StreamObserver[HelloReply] {
+ override def onNext(value: HelloReply): Unit =
+ println(value.reply)
+
+ override def onError(t: Throwable): Unit =
+ t.printStackTrace()
+
+ override def onCompleted(): Unit =
+ println("server completed")
+ }
+)
+
+requests.onNext(HelloRequest("Ada"))
+requests.onNext(HelloRequest("Grace"))
+requests.onCompleted()
+```
+
+If the client cannot finish sending requests, signal the failure with
+`requests.onError(error)`. Deadlines, cancellation, and call options are the
+standard grpc-java stub features, so they are configured on the generated
client
+stub before starting the call.
+
+### Bidirectional Clients
+
+Bidirectional streaming uses the same grpc-java request observer pattern, but
+responses can arrive while the client is still sending requests:
+
+```scala
+import io.grpc.stub.StreamObserver
+
+val requests = client.chat(
+ new StreamObserver[HelloReply] {
+ override def onNext(value: HelloReply): Unit =
+ println(value.reply)
+
+ override def onError(t: Throwable): Unit =
+ t.printStackTrace()
+
+ override def onCompleted(): Unit =
+ println("chat closed")
+ }
+)
+
+requests.onNext(HelloRequest("first"))
+requests.onNext(HelloRequest("second"))
+requests.onCompleted()
+```
+
+Use grpc-java observer subtypes such as `ClientResponseObserver`,
+`ClientCallStreamObserver`, or `ServerCallStreamObserver` when an application
+needs manual inbound flow control, readiness callbacks, cancellation handlers,
+or direct transport-level cancellation. The generated Scala methods accept the
+standard grpc-java observer types, so those advanced grpc-java patterns remain
+available without a separate Fory API.
+
+### Streaming Servers
+
+Unary server methods can use the direct Scala-friendly override shown earlier.
+Streaming server methods use grpc-java observers. A server-streaming
+implementation receives one request and writes zero or more responses:
+
+```scala
+import io.grpc.stub.StreamObserver
+import scala.util.control.NonFatal
+
+final class GreeterService extends GreeterGrpc.GreeterImplBase {
+ override def lotsOfReplies(
+ request: HelloRequest,
+ responseObserver: StreamObserver[HelloReply]
+ ): Unit =
+ try {
+ responseObserver.onNext(HelloReply(s"Hello, ${request.name}"))
+ responseObserver.onNext(HelloReply(s"Welcome, ${request.name}"))
+ responseObserver.onCompleted()
+ } catch {
+ case NonFatal(e) =>
+ responseObserver.onError(e)
+ }
+}
+```
+
+Client-streaming servers return an observer for incoming requests and write the
+single response when the request stream completes:
+
+```scala
+import io.grpc.stub.StreamObserver
+import scala.collection.mutable.ArrayBuffer
+
+final class GreeterService extends GreeterGrpc.GreeterImplBase {
+ override def lotsOfGreetings(
+ responseObserver: StreamObserver[HelloReply]
+ ): StreamObserver[HelloRequest] =
+ new StreamObserver[HelloRequest] {
+ private val names = ArrayBuffer.empty[String]
+
+ override def onNext(value: HelloRequest): Unit =
+ names += value.name
+
+ override def onError(t: Throwable): Unit =
+ names.clear()
+
+ override def onCompleted(): Unit = {
+ responseObserver.onNext(HelloReply(names.mkString("Hello ", ", ", "")))
+ responseObserver.onCompleted()
+ }
+ }
+}
+```
+
+Bidirectional servers also return an observer for incoming requests, but may
+emit responses from each `onNext` call:
+
+```scala
+import io.grpc.stub.StreamObserver
+
+final class GreeterService extends GreeterGrpc.GreeterImplBase {
+ override def chat(
+ responseObserver: StreamObserver[HelloReply]
+ ): StreamObserver[HelloRequest] =
+ new StreamObserver[HelloRequest] {
+ override def onNext(value: HelloRequest): Unit =
+ responseObserver.onNext(HelloReply(s"Echo: ${value.name}"))
+
+ override def onError(t: Throwable): Unit = ()
+
+ override def onCompleted(): Unit =
+ responseObserver.onCompleted()
+ }
+}
+```
+
+Server-streaming, client-streaming, and bidirectional server methods use
+grpc-java `StreamObserver` APIs because streaming completion, request flow
+control, cancellation, and backpressure follow grpc-java behavior.
+
+## Operations
+
+The generated service code only replaces request and response serialization.
+All normal gRPC operational features still belong to grpc-java:
+
+- Deadlines and cancellations
+- TLS and authentication
+- Name resolution and load balancing
+- Client and server interceptors
+- Status codes and metadata
+- Channel pooling and lifecycle management
+
+## Troubleshooting
+
+### Missing `io.grpc` or Guava Classes
+
+Add the grpc-java dependencies shown above. Generated Fory service files import
+grpc-java APIs, but `fory-scala` intentionally does not depend on gRPC.
+
+### Generic Protobuf Clients Cannot Read Payloads
+
+Fory-generated gRPC services use Fory bytes inside gRPC message frames, not
+protobuf message bytes. Use a Fory-generated client for Fory-generated
services,
+or provide a separate protobuf service endpoint for generic protobuf clients.
diff --git a/docs/guide/scala/index.md b/docs/guide/scala/index.md
index 80cc25744..80cea5a40 100644
--- a/docs/guide/scala/index.md
+++ b/docs/guide/scala/index.md
@@ -119,3 +119,4 @@ Fory Scala is built on top of Fory Java. Most configuration
options, features, a
- [Schema Metadata](schema-metadata.md) - Scala annotations, references, enum
IDs, and union metadata
- [Default Values](default-values.md) - Scala class default values support
- [Schema IDL And Xlang](schema-idl.md) - Scala 3 generated models and
macro-derived xlang serializers
+- [gRPC Support](grpc-support.md) - Scala 3 generated gRPC service companions
diff --git a/docs/guide/scala/schema-idl.md b/docs/guide/scala/schema-idl.md
index e9ae8c55d..d61d04294 100644
--- a/docs/guide/scala/schema-idl.md
+++ b/docs/guide/scala/schema-idl.md
@@ -44,6 +44,10 @@ Generated schema modules are also Fory modules. Use
`.withModule(...)` when
creating a custom Fory instance, or use the generated no-argument `toBytes` and
`fromBytes` helpers when the generated default Fory instance is sufficient.
+Schemas with service definitions can also generate Scala 3 gRPC service
+companions with `foryc --scala_out=... --grpc`. See
+[gRPC Support](grpc-support.md) for dependencies and client/server examples.
+
Generated helpers register message type identities before installing message
serializers. This two-phase order lets mutually recursive message graphs build
descriptor metadata through the normal `TypeResolver` path without temporary
diff --git a/scala/src/main/scala/org/apache/fory/scala/rpc/RpcFuture.scala
b/scala/src/main/scala/org/apache/fory/scala/rpc/RpcFuture.scala
new file mode 100644
index 000000000..973ad3722
--- /dev/null
+++ b/scala/src/main/scala/org/apache/fory/scala/rpc/RpcFuture.scala
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.fory.scala.rpc
+
+/** A cancellable unary RPC handle returned by generated Scala gRPC clients.
+ *
+ * `RpcFuture` is transport-free: it has no grpc-java types in its public
+ * signature, and it does not define how a generated client stores transport
+ * state. Generated gRPC service companions own the grpc-java integration.
+ */
+trait RpcFuture[+A] {
+ /** Return a Scala `Future` view of this RPC result. */
+ def asFuture: scala.concurrent.Future[A]
+
+ /** Cancel the underlying RPC if it has not completed. */
+ def cancel(): Boolean
+
+ /** Return true when the underlying RPC was cancelled. */
+ def isCancelled: Boolean
+
+ /** Return true when the underlying RPC has completed, failed, or cancelled.
*/
+ def isDone: Boolean
+}
diff --git a/scala/src/main/scala/org/apache/fory/scala/rpc/RpcIterator.scala
b/scala/src/main/scala/org/apache/fory/scala/rpc/RpcIterator.scala
new file mode 100644
index 000000000..9031984d4
--- /dev/null
+++ b/scala/src/main/scala/org/apache/fory/scala/rpc/RpcIterator.scala
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.fory.scala.rpc
+
+/** A closeable server-streaming RPC iterator returned by generated Scala gRPC
clients.
+ *
+ * Generated gRPC service companions own the underlying transport call.
Calling
+ * `close()` or `cancel()` releases that call early when the caller stops
+ * reading before the stream completes.
+ */
+trait RpcIterator[+A] extends Iterator[A] with AutoCloseable {
+ /** Cancel or close the underlying RPC stream. */
+ def cancel(): Unit
+
+ /** Return true when the underlying RPC stream is closed. */
+ def isClosed: Boolean
+
+ final override def close(): Unit = cancel()
+}
diff --git a/scala/src/test/scala/org/apache/fory/scala/rpc/RpcHandleTest.scala
b/scala/src/test/scala/org/apache/fory/scala/rpc/RpcHandleTest.scala
new file mode 100644
index 000000000..06075d795
--- /dev/null
+++ b/scala/src/test/scala/org/apache/fory/scala/rpc/RpcHandleTest.scala
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.fory.scala.rpc
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import scala.concurrent.Future
+
+class RpcHandleTest extends AnyFunSuite {
+ test("rpc future keeps cancellation separate from future view") {
+ final class TestFuture extends RpcFuture[Int] {
+ private var cancelled = false
+
+ override def asFuture: Future[Int] = Future.successful(1)
+
+ override def cancel(): Boolean = {
+ cancelled = true
+ true
+ }
+
+ override def isCancelled: Boolean = cancelled
+
+ override def isDone: Boolean = cancelled
+ }
+
+ val handle = new TestFuture
+ assert(handle.asFuture.isCompleted)
+ assert(!handle.isCancelled)
+ assert(handle.cancel())
+ assert(handle.isCancelled)
+ assert(handle.isDone)
+ }
+
+ test("rpc iterator close delegates to cancel") {
+ final class TestIterator extends RpcIterator[Int] {
+ private var closed = false
+
+ override def hasNext: Boolean = false
+
+ override def next(): Int = throw new NoSuchElementException
+
+ override def cancel(): Unit = {
+ closed = true
+ }
+
+ override def isClosed: Boolean = closed
+ }
+
+ val iterator = new TestIterator
+ assert(!iterator.isClosed)
+ iterator.close()
+ assert(iterator.isClosed)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
