$ mkdir proto/

syntax = "proto3";
package store;

service Inventory {
    // Add inserts a new Item into the inventory.
    rpc Add(Item) returns (InventoryChangeResponse);

    // Remove removes Items from the inventory.
    rpc Remove(ItemIdentifier) returns (InventoryChangeResponse);

    // Get retrieves Item information.
    rpc Get(ItemIdentifier) returns (Item);

    // UpdateQuantity increases or decreases the stock quantity of an Item.
    rpc UpdateQuantity(QuantityChangeRequest) returns (InventoryUpdateResponse);

    // UpdatePrice increases or decreases the price of an Item.
    rpc UpdatePrice(PriceChangeRequest) returns (InventoryUpdateResponse);

    // Watch streams Item updates from the inventory.
    rpc Watch(ItemIdentifier) returns (stream Item);
}

message ItemIdentifier {
    string sku = 2;
}

message ItemStock {
    float  price    = 1;
    uint32 quantity = 2;
}

message ItemInformation {
    optional string name        = 1;
    optional string description = 2;
}

message Item {
    ItemIdentifier           identifier  = 1;
    ItemStock                stock       = 2;
    optional ItemInformation information = 3;
}

message QuantityChangeRequest {
    string sku    = 1;
    int32  change = 2;
}

message PriceChangeRequest {
    string sku   = 1;
    float  price = 2;
}

message InventoryChangeResponse {
    string status = 1;
}

message InventoryUpdateResponse {
    string status   = 1;
    float price     = 2;
    uint32 quantity = 3;
}

[package]
name = "demo"
version = "0.1.0"
edition = "2021"
publish = false

[[bin]]
name = "server"
path = "src/main.rs"

[[bin]]
name = "cli"
path = "src/cli.rs"

[dependencies]
tonic = "0.8"
prost = "0.11"
tokio = { version = "1.24", features = ["macros", "rt-multi-thread"] }
tokio-stream = { version = "0.1", features = ["net"] }
futures = "0.3"
clap = { version = "4.1.4", features = ["derive"] }
tonic-reflection = "0.6.0"

[build-dependencies]
tonic-build = "0.8"

[dev-dependencies]
uuid = { version = "1.2.2", features = ["v4", "fast-rng"] }
futures-util = "0.3.25"
anyhow = "1"

use std::env;
use std::path::PathBuf;

fn main() -> Result<(), Box<dyn std::error::Error>> {
   let proto_file = "./proto/store.proto";
   let out_dir = PathBuf::from(env::var("OUT_DIR").unwrap());

   tonic_build::configure()
           .protoc_arg("--experimental_allow_proto3_optional") // for older systems
           .build_client(true)
           .build_server(true)
           .file_descriptor_set_path(out_dir.join("store_descriptor.bin"))
           .out_dir("./src")
           .compile(&[proto_file], &["proto"])?;

   Ok(())
}

$ cargo build

use futures::Stream;
use std::borrow::BorrowMut;
use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tonic::{Request, Response, Status};

use crate::store::inventory_server::Inventory;
use crate::store::{
    InventoryChangeResponse, InventoryUpdateResponse, Item, ItemIdentifier, PriceChangeRequest,
    QuantityChangeRequest,
};

const BAD_PRICE_ERR: &str = "provided PRICE was invalid";
const DUP_PRICE_ERR: &str = "item is already at this price";
const DUP_ITEM_ERR: &str = "item already exists in inventory";
const EMPTY_QUANT_ERR: &str = "invalid quantity of 0 provided";
const EMPTY_SKU_ERR: &str = "provided SKU was empty";
const NO_ID_ERR: &str = "no ID or SKU provided for item";
const NO_ITEM_ERR: &str = "the item requested was not found";
const NO_STOCK_ERR: &str = "no stock provided for item";
const UNSUFF_INV_ERR: &str = "not enough inventory for quantity change";

#[derive(Debug)]
pub struct StoreInventory {
    inventory: Arc<Mutex<HashMap<String, Item>>>,
}

impl Default for StoreInventory {
    fn default() -> Self {
        StoreInventory {
            inventory: Arc::new(Mutex::new(HashMap::<String, Item>::new())),
        }
    }
}

#[tonic::async_trait]
impl Inventory for StoreInventory {}

    async fn add(
        &self,
        request: Request<Item>,
    ) -> Result<Response<InventoryChangeResponse>, Status> {
        let item = request.into_inner();

        // validate SKU, verify that it&#039;s present and not empty
        let sku = match item.identifier.as_ref() {
            Some(id) if id.sku == "" => return Err(Status::invalid_argument(EMPTY_SKU_ERR)),
            Some(id) => id.sku.to_owned(),
            None => return Err(Status::invalid_argument(NO_ID_ERR)),
        };

        // validate stock, verify its present and price is not negative or $0.00
        match item.stock.as_ref() {
            Some(stock) if stock.price <= 0.00 => {
                return Err(Status::invalid_argument(BAD_PRICE_ERR))
            }
            Some(_) => {}
            None => return Err(Status::invalid_argument(NO_STOCK_ERR)),
        };

        // if the item is already present don&#039;t allow the duplicate
        let mut map = self.inventory.lock().await;
        if let Some(_) = map.get(&sku) {
            return Err(Status::already_exists(DUP_ITEM_ERR));
        }

        // add the item to the inventory
        map.insert(sku.into(), item);

        Ok(Response::new(InventoryChangeResponse {
            status: "success".into(),
        }))
    }

    async fn remove(
        &self,
        request: Request<ItemIdentifier>,
    ) -> Result<Response<InventoryChangeResponse>, Status> {
        let identifier = request.into_inner();

        // don&#039;t allow empty SKU
        if identifier.sku == "" {
            return Err(Status::invalid_argument(EMPTY_SKU_ERR));
        }

        // remove the item (if present)
        let mut map = self.inventory.lock().await;
        let msg = match map.remove(&identifier.sku) {
            Some(_) => "success: item was removed",
            None => "success: item didn&#039;t exist",
        };

        Ok(Response::new(InventoryChangeResponse {
            status: msg.into(),
        }))
    }

    async fn get(&self, request: Request<ItemIdentifier>) -> Result<Response<Item>, Status> {
        let identifier = request.into_inner();

        // don&#039;t allow empty SKU
        if identifier.sku == "" {
            return Err(Status::invalid_argument(EMPTY_SKU_ERR));
        }

        // retrieve the item if it exists
        let map = self.inventory.lock().await;
        let item = match map.get(&identifier.sku) {
            Some(item) => item,
            None => return Err(Status::not_found(NO_ITEM_ERR)),
        };

        Ok(Response::new(item.clone()))
    }

    async fn update_quantity(
        &self,
        request: Request<QuantityChangeRequest>,
    ) -> Result<Response<InventoryUpdateResponse>, Status> {
        let change = request.into_inner();

        // don&#039;t allow empty SKU
        if change.sku == "" {
            return Err(Status::invalid_argument(EMPTY_SKU_ERR));
        }

        // quantity changes with no actual change don&#039;t make sense, inform user
        if change.change == 0 {
            return Err(Status::invalid_argument(EMPTY_QUANT_ERR));
        }

        // retrieve the current inventory item data
        let mut map = self.inventory.lock().await;
        let item = match map.get_mut(&change.sku) {
            Some(item) => item,
            None => return Err(Status::not_found(NO_ITEM_ERR)),
        };

        // retrieve the stock mutable so we can update the quantity
        let mut stock = match item.stock.borrow_mut() {
            Some(stock) => stock,
            None => return Err(Status::internal(NO_STOCK_ERR)),
        };

        // validate and then handle the quantity change
        stock.quantity = match change.change {
            // handle negative numbers as stock reduction
            change if change < 0 => {
                if change.abs() as u32 > stock.quantity {
                    return Err(Status::resource_exhausted(UNSUFF_INV_ERR));
                }
                stock.quantity - change.abs() as u32
            }
            // handle positive numbers as stock increases
            change => stock.quantity + change as u32,
        };

        Ok(Response::new(InventoryUpdateResponse {
            status: "success".into(),
            price: stock.price,
            quantity: stock.quantity,
        }))
    }

    async fn update_price(
        &self,
        request: Request<PriceChangeRequest>,
    ) -> Result<Response<InventoryUpdateResponse>, Status> {
        let change = request.into_inner();

        // don&#039;t allow empty SKU
        if change.sku == "" {
            return Err(Status::invalid_argument(EMPTY_SKU_ERR));
        }

        // $0.00 disallowed and negatives don&#039;t make sense, inform the user
        if change.price <= 0.0 {
            return Err(Status::invalid_argument(BAD_PRICE_ERR));
        }

        // retrieve the current inventory item data
        let mut map = self.inventory.lock().await;
        let item = match map.get_mut(&change.sku) {
            Some(item) => item,
            None => return Err(Status::not_found(NO_ITEM_ERR)),
        };

        // retrieve the stock mutable so we can update the quantity
        let mut stock = match item.stock.borrow_mut() {
            Some(stock) => stock,
            None => return Err(Status::internal(NO_STOCK_ERR)),
        };

        // let the client know if they requested to change the price to the
        // price that is already currently set
        if stock.price == change.price {
            return Err(Status::invalid_argument(DUP_PRICE_ERR));
        }

        // update the item unit price
        stock.price = change.price;

        Ok(Response::new(InventoryUpdateResponse {
            status: "success".into(),
            price: stock.price,
            quantity: stock.quantity,
        }))
    }

    type WatchStream = Pin<Box<dyn Stream<Item = Result<Item, Status>> + Send>>;

    async fn watch(
        &self,
        request: Request<ItemIdentifier>,
    ) -> Result<Response<Self::WatchStream>, Status> {
        // retrieve the relevant item and get a baseline
        let id = request.into_inner();
        let mut item = self.get(Request::new(id.clone())).await?.into_inner();

        // the channel will be our stream back to the client, we&#039;ll send copies
        // of the requested item any time we notice a change to it in the
        // inventory.
        let (tx, rx) = mpsc::unbounded_channel();

        // we&#039;ll loop and poll new copies of the item until either the client
        // closes the connection, or an error occurs.
        let inventory = self.inventory.clone();
        tokio::spawn(async move {
            loop {
                // it&#039;s somewhat basic, but for this demo we&#039;ll just check the
                // item every second for any changes.
                tokio::time::sleep(std::time::Duration::from_secs(1)).await;

                // pull a fresh copy of the item in the inventory
                let map = inventory.lock().await;
                let item_refresh = match map.get(&id.sku) {
                    Some(item) => item,
                    // the item has been removed from the inventory. Let the
                    // client know, and stop the stream.
                    None => {
                        if let Err(err) = tx.send(Err(Status::not_found(NO_ITEM_ERR))) {
                            println!("ERROR: failed to update stream client: {:?}", err);
                        }
                        return;
                    }
                };

                // check to see if the item has changed since we last saw it,
                // and if it has inform the client via the stream.
                if item_refresh != &item {
                    if let Err(err) = tx.send(Ok(item_refresh.clone())) {
                        println!("ERROR: failed to update stream client: {:?}", err);
                        return;
                    }
                }

                // cache the most recent copy of the item
                item = item_refresh.clone()
            }
        });

        let stream = UnboundedReceiverStream::new(rx);
        Ok(Response::new(Box::pin(stream) as Self::WatchStream))
    }

use tonic::transport::Server;

use server::StoreInventory;
use store::inventory_server::InventoryServer;

pub mod server;
pub mod store;

mod store_proto {
   include!("store.rs");

   pub(crate) const FILE_DESCRIPTOR_SET: &[u8] =
      tonic::include_file_descriptor_set!("store_descriptor");
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
   let addr = "127.0.0.1:9001".parse()?;
   let inventory = StoreInventory::default();

   let reflection_service = tonic_reflection::server::Builder::configure()
           .register_encoded_file_descriptor_set(store_proto::FILE_DESCRIPTOR_SET)
           .build()
           .unwrap();

   Server::builder()
           .add_service(InventoryServer::new(inventory))
           .add_service(reflection_service)
           .serve(addr)
           .await?;
   Ok(())
}

$ cargo run --release --bin server

{
   "identifier": {
      "sku": "TESTSKU"
   },
   "stock": {
      "price": 1.9900000095367432,
      "quantity": 20
   },
   "information": {
      "name": "bananas",
      "description": "yellow fruit"
   }
}

pub mod store;

use clap::Parser;
use futures::StreamExt;

use store::inventory_client::InventoryClient;
use store::{
    Item, ItemIdentifier, ItemInformation, ItemStock, PriceChangeRequest, QuantityChangeRequest,
};

#[derive(Debug, Parser)]
struct Options {
    #[clap(subcommand)]
    command: Command,
}

#[derive(Debug, Parser)]
enum Command {
    Add(AddOptions),
    Remove(RemoveOptions),
    Get(GetOptions),
    UpdateQuantity(UpdateQuantityOptions),
    UpdatePrice(UpdatePriceOptions),
    Watch(GetOptions),
}

#[derive(Debug, Parser)]
struct AddOptions {
    #[clap(long)]
    sku: String,
    #[clap(long)]
    price: f32,
    #[clap(default_value = "0", long)]
    quantity: u32,
    #[clap(long)]
    name: Option<String>,
    #[clap(long)]
    description: Option<String>,
}

async fn add(opts: AddOptions) -> Result<(), Box<dyn std::error::Error>> {
    let mut client = InventoryClient::connect("http://127.0.0.1:9001").await?;

    let id = ItemIdentifier { sku: opts.sku };

    let stock = ItemStock {
        price: opts.price,
        quantity: opts.quantity,
    };

    let info = ItemInformation {
        name: opts.name,
        description: opts.description,
    };

    let item = Item {
        identifier: Some(id),
        stock: Some(stock),
        information: Some(info),
    };

    let request = tonic::Request::new(item);
    let response = client.add(request).await?;
    assert_eq!(response.into_inner().status, "success");
    println!("success: item was added to the inventory.");

    Ok(())
}

#[derive(Debug, Parser)]
struct RemoveOptions {
    #[clap(long)]
    sku: String,
}

async fn remove(opts: RemoveOptions) -> Result<(), Box<dyn std::error::Error>> {
    let mut client = InventoryClient::connect("http://127.0.0.1:9001").await?;

    let request = tonic::Request::new(ItemIdentifier { sku: opts.sku });
    let response = client.remove(request).await?;
    let msg = response.into_inner().status;
    assert!(msg.starts_with("success"));
    println!("{}", msg);

    Ok(())
}

#[derive(Debug, Parser)]
struct GetOptions {
    #[clap(long)]
    sku: String,
}

async fn get(opts: GetOptions) -> Result<(), Box<dyn std::error::Error>> {
    let mut client = InventoryClient::connect("http://127.0.0.1:9001").await?;

    let request = tonic::Request::new(ItemIdentifier { sku: opts.sku });
    let item = client.get(request).await?.into_inner();
    println!("found item: {:?}", item);

    Ok(())
}

#[derive(Debug, Parser)]
struct UpdateQuantityOptions {
    #[clap(long)]
    sku: String,
    #[clap(allow_hyphen_values = true, long)]
    change: i32,
}

async fn update_quantity(opts: UpdateQuantityOptions) -> Result<(), Box<dyn std::error::Error>> {
    let mut client = InventoryClient::connect("http://127.0.0.1:9001").await?;

    let request = tonic::Request::new(QuantityChangeRequest {
        sku: opts.sku,
        change: opts.change,
    });

    let message = client.update_quantity(request).await?.into_inner();
    assert_eq!(message.status, "success");
    println!(
        "success: quantity was updated. Quantity: {} Price: {}",
        message.quantity, message.price
    );

    Ok(())
}

#[derive(Debug, Parser)]
struct UpdatePriceOptions {
    #[clap(long)]
    sku: String,
    #[clap(long)]
    price: f32,
}

async fn update_price(opts: UpdatePriceOptions) -> Result<(), Box<dyn std::error::Error>> {
    let mut client = InventoryClient::connect("http://127.0.0.1:9001").await?;

    let request = tonic::Request::new(PriceChangeRequest {
        sku: opts.sku,
        price: opts.price,
    });

    let message = client.update_price(request).await?.into_inner();
    assert_eq!(message.status, "success");
    println!(
        "success: price was updated. Quantity: {} Price: {}",
        message.quantity, message.price
    );

    Ok(())
}

async fn watch(opts: GetOptions) -> Result<(), Box<dyn std::error::Error>> {
    let mut client = InventoryClient::connect("http://127.0.0.1:9001").await?;

    let mut stream = client
        .watch(ItemIdentifier {
            sku: opts.sku.clone(),
        })
        .await?
        .into_inner();

    println!("streaming changes to item {}", opts.sku);
    while let Some(item) = stream.next().await {
        match item {
            Ok(item) => println!("item was updated: {:?}", item),
            Err(err) => {
                if err.code() == tonic::Code::NotFound {
                    println!("watched item has been removed from the inventory.");
                    break;
                } else {
                    return Err(err.into());
                }
            }
        };
    }
    println!("stream closed");

    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let opts = Options::parse();

    use Command::*;
    match opts.command {
        Add(opts) => add(opts).await?,
        Remove(opts) => remove(opts).await?,
        Get(opts) => get(opts).await?,
        UpdateQuantity(opts) => update_quantity(opts).await?,
        UpdatePrice(opts) => update_price(opts).await?,
        Watch(opts) => watch(opts).await?,
    };

    Ok(())
}

$ cargo build --release --bin cli
$ cp target/release/cli ./

$ ./cli add --sku TESTSKU --price 1.99 --quantity 20 --name bananas --description "yellow fruit"
success: item was added to the inventory.

$ ./cli get --sku TESTSKU
found item: { sku: "TESTSKU" }, stock: { price: 1.99, quantity: 0 }, information: { name: "bananas", description: "yellow fruit" }

$ ./cli add --sku TESTSKU --price 2.99
Error: Status { code: AlreadyExists, message: "item already exists in inventory" }

$ ./cli update-quantity --sku TESTSKU --change -17
success: quantity was updated. Quantity: 3 Price: 1.99

$ ./cli update-price --sku TESTSKU --price 2.19
success: price was updated. Quantity: 3 Price: 2.19

$ ./cli watch --sku TESTSKU
streaming changes to item TESTSKU

$ ./cli update-quantity --sku TESTSKU --change +50
success: quantity was updated. Quantity: 53 Price: 2.19
$ ./cli update-price --sku TESTSKU --price 1.99
success: price was updated. Quantity: 53 Price: 1.99
$ ./cli remove --sku TESTSKU
success: item was removed

$ ./cli watch --sku TESTSKU
streaming changes to item TESTSKU
item was updated: Item { identifier: Some(ItemIdentifier { sku: "TESTSKU" }), stock: Some(ItemStock { price: 2.19, quantity: 53 }), information: Some(ItemInformation { name: Some("bananas"), description: Some("yellow fruit") }) }
item was updated: Item { identifier: Some(ItemIdentifier { sku: "TESTSKU" }), stock: Some(ItemStock { price: 1.99, quantity: 53 }), information: Some(ItemInformation { name: Some("bananas"), description: Some("yellow fruit") }) }
watched item has been removed from the inventory.
stream closed

$ ./cli --help
Usage: cli <COMMAND>

Commands:
  add
  remove
  get
  update-quantity
  update-price
  watch
  help             Print this message or the help of the given subcommand(s)

Options:
  -h, --help  Print help