Streaming in gRPC

Streaming RPCs let you transfer sequences of messages instead of a single one. gRPC Fundamentals & Contracts lists the three types: server streaming, client streaming and bidirectional. This guide shows how to implement them in a Sword controller.

Server streaming

In server streaming the client sends one request and the server responds with a stream of messages. The trait generated by tonic defines an associated type per method (for example type StreamUsersStream) that you assign to GrpcStream<T>, and you must return GrpcResult<Self::XxxStream>.

You can build the stream with async_stream::try_stream and wrap it with GrpcResponse::stream():

use async_stream::try_stream;
use std::sync::Arc;
use tokio::time::{self, Duration};

use sword::grpc::*;
use sword::prelude::*;

#[controller(kind = Controller::Grpc, service = UserServiceServer)]
pub struct UsersController {
    users: Arc<UserRepository>,
}

#[sword::grpc::async_trait]
impl UserService for UsersController {
    type StreamUsersStream = GrpcStream<UserItem>;

    async fn stream_users(
        &self,
        _: Request<StreamUsersRequest>,
    ) -> GrpcResult<Self::StreamUsersStream> {
        let users = self.users.find_all().await;

        let output = try_stream! {
            for user in users {
                yield UserItem::from(&user);

                time::sleep(Duration::from_secs(1)).await;
            }
        };

        Ok(GrpcResponse::stream(output))
    }
}

GrpcResponse::stream() boxes and wraps the stream in a Response<GrpcStream<T>>, ready for the server to send to the client.

Client streaming

In client streaming the client sends a stream of messages and the server responds with a single message. The method receives Request<tonic::Streaming<T>>; with into_inner() you get the stream and iterate over the messages:

use tonic::Streaming;

use sword::grpc::*;
use sword::prelude::*;

#[sword::grpc::async_trait]
impl UserService for UsersController {
    async fn upload_users(&self, req: Request<Streaming<UserItem>>) -> GrpcResult<UploadReply> {
        let mut stream = req.into_inner();

        let mut count = 0u32;
        while let Some(item) = stream.message().await? {
            // ... process each item ...
            count += 1;
        }

        Ok(GrpcResponse::message(UploadReply { count }))
    }
}

stream.message().await returns Result<Option<T>, tonic::Status>: Some for each received message and None when the client closes the stream.

Bidirectional streaming

In bidi streaming both sides send and receive streams concurrently. You receive Request<tonic::Streaming<T>> and return GrpcResult<Self::XxxStream>, so you can combine iterating over the incoming stream with async_stream for the response:

use async_stream::try_stream;
use tonic::Streaming;

use sword::grpc::*;
use sword::prelude::*;

#[sword::grpc::async_trait]
impl UserService for UsersController {
    type ChatStream = GrpcStream<ChatReply>;

    async fn chat(&self, req: Request<Streaming<ChatMessage>>) -> GrpcResult<Self::ChatStream> {
        let mut incoming = req.into_inner();

        let output = try_stream! {
            while let Some(message) = incoming.message().await? {
                // ... build one response per message ...
                yield ChatReply { text: message.text };
            }
        };

        Ok(GrpcResponse::stream(output))
    }
}

Base types

gRPC steaming types

The GrpcStream<T> Type

pub type GrpcStream<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send + 'static>>;

Returns

• A boxed stream of T messages or Status errors.

When to use it

• As the type of the associated stream in server streaming methods (type StreamUsersStream = GrpcStream<T>) and as the result of GrpcResponse::stream().

The GrpcResponse::stream() Method

pub fn stream<T, S>(stream: S) -> tonic::Response<GrpcStream<T>>
where
    S: Stream<Item = Result<T, Status>> + Send + 'static,

Returns

• A Response<GrpcStream<T>> wrapping a stream.

When to use it

• In server and bidirectional streaming methods to return a sequence of messages.

Notes

• GrpcResponse is a stateless struct: its methods are static constructors.
• The associated type defined by the generated trait must match GrpcStream<T>; otherwise the implementation will not compile.
• async_stream is an optional but convenient dependency for building streams; you can also implement Stream by hand if you prefer.