Integrating gRPC

Why gRPC?

gRPC has gives us a nice way to declare our API in a schema definition and then code generate the server side implementation.

The data transfer protocol (Protobuf) is also useful if we want to do data storage i.e. for Big data projects. So we get 1 tool that can do 2 jobs.

Installation

Let's create a crate for our api definition and code generator.

$ cargo init --lib crates/grpc-api
Created library package

Install gRPC crates into your project cd into your crates/grpc-api folder.

Add the following to your app/Cargo.toml below the [dependencies]

cargo add tonic
cargo add prost
cargo add serde --features derive
cargo add tonic-build --build

Create a folder called crates/grpc-api/protos and a file called api.proto

.
├── .devcontainer/
│   └── ...
└── crates/
│         asset-pipeline/
│         └── ...
│         web-server/
│         └── ...
│         db/
│         └── ...
│         grpc-api/
│         ├── protos
│         │   └── api.proto
│         ├── src
│         │   └── lib.rs
│         ├── Cargo.toml
│         └── build.rs
│         ui-components/
│         └── ...
├── Cargo.toml
└── Cargo.lock

Defining the API

Add this to your api.proto

syntax = "proto3";

package api;

service Users {
    rpc GetUsers(GetUsersRequest) returns (GetUsersResponse);
}

message GetUsersRequest {
}

message GetUsersResponse {
    repeated User users = 1;
}

message User {
    uint32 id = 1;
    string email = 2;
}

Updating our build.rs

Add the following to your crates/grpc-api/build.rs in the main function.

use std::io::Result;

fn main() -> Result<()> {
    tonic_build::configure()
        .compile(
            &["protos/api.proto"], // Update the path to the .proto file
            &["protos"],           // Update the search path for proto files
        )?;
    Ok(())
}

Add a lib.rs

The crates/grpc-api/src/lib.rs will load the code generated by tonic build and make it available to other crates.

pub mod api {
    #![allow(clippy::large_enum_variant)]
    #![allow(clippy::derive_partial_eq_without_eq)]
    tonic::include_proto!("api");
}

Everything should compile at this point.

Implementing our API End Point

Create a file called crates/web-server/api_service.rs and add the following implementation for our gRPC service.

use grpc_api::api::*;
use crate::errors::CustomError;
use db::queries;
use deadpool_postgres::Pool;
use tonic::{Request, Response, Status};

pub struct UsersService {
    pub pool: Pool,
}

#[tonic::async_trait]
impl grpc_api::users_server::Fortunes for UsersService {
    async fn get_users(
        &self,
        _request: Request<GetUsersRequest>,
    ) -> Result<Response<GetUsersResponse>, Status> {
        // Get a client from our database pool
        let client = self
            .pool
            .get()
            .await
            .map_err(|e| CustomError::Database(e.to_string()))?;

        // Get the fortunes from the database
        let fortunes = queries::users::users(&client)
            .await
            .map_err(|e| CustomError::Database(e.to_string()))?;

        // Map the structs we get from cornucopia to the structs
        // we need for our gRPC reply.
        let users = users
            .into_iter()
            .map(|user| User {
                id: user.id as u32,
                email: user.email,
            })
            .collect();

        let users = GetUsersResponse {
            users,
        };

        return Ok(Response::new(response));
    }
}

Integrating Tonic and Axum

We can modify our Axum server so that it can respond to HTTP and gRPC requests.

Our crates/web-server/src/main.rs now needs to look like this.

mod config;
mod errors;

use crate::errors::CustomError;
use axum::{extract::Extension, response::Json, routing::get, Router};
use std::net::SocketAddr;
use db::User;
use tower::{make::Shared, steer::Steer, BoxError, ServiceExt};
use tonic::transport::Server;
use grpc_api::api::api_server::UsersServer;
use http::{header::CONTENT_TYPE, Request};

#[tokio::main]
async fn main() {
    let config = config::Config::new();

    let pool = db::create_pool(&config.database_url);

    // build our application with a route
    let app = Router::new()
        .route("/", get(users))
        .layer(Extension(config))
        .layer(Extension(pool.clone()))
        .boxed_clone();


    // Handle gRPC API requests
    let grpc = Server::builder()
        .add_service(TraceServer::new(api::trace_grpc_service::TraceService {
            pool,
        }))
        .into_service()
        .map_response(|r| r.map(axum::body::boxed))
        .boxed_clone();

    // Create a service that can respond to Web and gRPC
    let http_grpc = Steer::new(vec![app, grpc], |req: &Request<Body>, _svcs: &[_]| {
        if req.headers().get(CONTENT_TYPE).map(|v| v.as_bytes()) != Some(b"application/grpc") {
            0
        } else {
            1
        }
    });

    // run it
    let addr = SocketAddr::from(([0, 0, 0, 0], 3000));
    println!("listening on {}", addr);
    let listener = tokio::net::TcpListener::bind(&addr).await.unwrap();
    axum::serve(listener, app.into_make_service()).await.unwrap();
}

async fn users(Extension(pool): Extension<db::Pool>) -> Result<Json<Vec<User>>, CustomError> {
    let client = pool.get().await?;

    let users = db::queries::users::get_users()
        .bind(&client)
        .all()
        .await?;

    Ok(Json(users))
}

BloomRPC

To see our server working we can use BloomRPC with which we can load our api.proto and fire off an RPC call to our fortunes API.

It will look something like the screenshot below.