gRPC

Support for gRPC is implemented using

• tonic for client/service communication
• prost-reflect for handling dynamic protobufs and serialization/deserialization

Use the grpc feature flag of crossflow to enable this support:

# Cargo.toml

[dependencies]
crossflow = { version = "*", features = ["diagram", "grpc"] }

Enabling gRPC

Use registry.enable_grpc(rt) to register premade node builders that can build gRPC clients into a workflow:

use crossflow::prelude::*;
use tokio::runtime::Runtime;

// Initialize a regular registry
let mut registry = DiagramElementRegistry::new();

// Use a tokio runtime to enable gRPC in the registry. Tonic is only compatible
// with tokio, so a tokio runtime is necessary to execute it.
//
// This will add gRPC node builders to the registry.
let rt = Arc::new(Runtime::new().unwrap());
registry.enable_grpc(Arc::clone(&rt));

// Run the tokio runtime on a separate thread.
//
// Bevy uses smol-rs for its async runtime, which is not directly compatible with
// tokio. We let tokio run on a separate thread and use channels to pass data
// between the runtimes.
let (exit_notifier, exit_receiver) = tokio::sync::oneshot::channel::<()>();
std::thread::spawn(move || {
    let _ = rt.block_on(exit_receiver);
});

Descriptor Pool

Additionally you will need to use decode_global_file_descriptor_set to add your proto definitions to the global descriptor pool. The node builders will look for service and protobuf definitions in the global descriptor pool. If a description isn’t present in the pool, the node builders will return a NodeBuildingError when you attempt to build a workflow from the diagram.

We leave it open-ended how descriptors are sent to the executor. There are many viable ways to do this using various middlewares or compiling the definitions into the executor itself. Users are welcome to come up with reusable third-party libraries to implement specific approaches and share them with the ecosystem.

Configuration

Regardless of client type, all gRPC nodes are configured using GrpcConfig. That config allows you to specify the service and method to call, as well as the URI of where the service should be found. You can optionally specify a timeout that will have the client quit if the response does not arrive by then.

As mentioned above whatever service type that you configure your node to use needs to be present within the executor’s global descriptor pool before you attempt to build a workflow from the diagram.

Unary and Server-Streaming Requests

Unary and server-streaming requests are both covered by the same node builder: "grcp_request".

Request — The grpc_request nodes take in a single JsonMessage which will be used as the single request message sent to the server. Each JsonMessage sent in activates a new gRPC client←→server session where the server will receive this single request. The gRPC client itself is instantiated when the workflow is originally built, and will be reused for all requests sent into this node. Separate sessions of the workflow will also use the same gRPC client whenever the same node is activated.

“out” — The response messages received from the gRPC server will be sent out of the stream named "out". This is the case whether the service is unary or server-streaming. This allows the node to have a single consistent structure regardless of whether the server ends up sending zero, one, or arbitrarily many response messages.

Response — The final response of the node will be a Result whose value is Ok after the server has ended the connection with no errors, or Err if an error came up during the request. The Err will contain a string rendering of a Status, based on gRPC status codes.

“canceller” — In case you want to cancel the gRPC request, you can capture this UnboundedSender, either storing it in a buffer or passing it to a node that can make a decision about when to cancel. Passing in a Some(String) will have the string included inside the string-rendered Status of the final response’s Err.

Bidirectional and Client-Streaming Requests

Bidirectional and client-streaming requests are both covered by the "grpc_client" node builder. Technically this node builder can also support unary and server-streaming requests, but the “grpc_request” node builder is more ergonomic for those.

What makes "grpc_client" different is it takes in an UnboundedReceiver of JsonMessage instead of just a single JsonMessage. This allows any number of messages to be streamed from the client (inside the workflow) out to the server within a single gRPC client←→server session.

Each UnboundedReceiver sent into this node will active a new independent gRPC client←→server session using the same client. Every message sent through the UnboundedSender associated with that receiver will be streamed out over the same gRPC client←→server session.