No tarball for gmail

.dockerignore

@@ -0,0 +1 @@
+target/

.gitignore

@@ -0,0 +1 @@
+/target

.rustfmt.toml

@@ -0,0 +1 @@
+edition = "2021"

Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "pubsub-rs"
+version = "0.1.0"
+edition = "2021"
+
+[[bin]]
+name = "client"
+path = "src/client.rs"
+
+[dependencies]
+tonic = "0.7"
+prost = "0.10"
+tokio = { version = "1.0", features = ["macros", "rt-multi-thread"] }
+clap = { version = "3.2.14", features = ["derive"] }
+
+[build-dependencies]
+tonic-build = "0.7"

Dockerfile

@@ -0,0 +1,14 @@
+FROM rust as builder
+
+WORKDIR /opt/pubsub-client
+
+COPY . .
+
+RUN apt update
+RUN apt install -y cmake
+RUN cargo build --release
+
+FROM rust
+WORKDIR /opt/
+COPY --from=builder /opt/pubsub-client/target/release/client .
+ENTRYPOINT ["/opt/client"]

README.md

@@ -0,0 +1,77 @@
+# PubSub client
+
+## Quick and easy
+1. run `sudo docker build -t client .`
+2. run `sudo docker run --rm --network=host -it client publish topic message`
+2. run `sudo docker run --rm --network=host -it client consume topic`
+
+## With Cargo
+1. run `cargo build --release`
+2. run `./target/release/client publish topic message`
+2. run `./target/release/client consume topic`
+
+
+## CLI
+Note that there are a few changes in the CLI interface as opposed to what's in
+the email. Please see `--help` for details.
+
+Examples (binary named client):
+Publish a message
+`client publish mytopic mymessage`
+
+Consume a message (default timeout of 10s)
+`client consume mytopic`
+
+Consume a message with given timeout (seconds)
+`client consume mytopic -t 15`
+
+
+```
+$ ./target/release/client --help
+pubsub-rs 0.1.0
+
+USAGE:
+    client [OPTIONS] <SUBCOMMAND>
+
+OPTIONS:
+    -e, --endpoint <ENDPOINT>    [default: localhost]
+    -h, --help                   Print help information
+    -p, --port <PORT>            [default: 50051]
+        --protocol <PROTOCOL>    [default: tcp] [possible values: http, tcp]
+    -V, --version                Print version information
+
+SUBCOMMANDS:
+    consume
+    help       Print this message or the help of the given subcommand(s)
+    publish
+```
+
+Publish:
+```
+client-publish
+
+USAGE:
+    client publish <TOPIC> <MESSAGE>
+
+ARGS:
+    <TOPIC>
+    <MESSAGE>
+
+OPTIONS:
+    -h, --help    Print help information
+```
+
+Consume:
+```
+client-consume
+
+USAGE:
+    client consume [OPTIONS] <TOPIC>
+
+ARGS:
+    <TOPIC>
+
+OPTIONS:
+    -h, --help                 Print help information
+    -t, --timeout <TIMEOUT>    [default: 10]
+```

build.rs

@@ -0,0 +1,4 @@
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tonic_build::compile_protos("proto/pubsub.proto")?;
+    Ok(())
+}

proto/pubsub.proto

@@ -0,0 +1,28 @@
+syntax = "proto3";
+package pubsub;
+
+service PubSub {
+
+  rpc Publish(PublishMessage) returns (PublishResponse) {}
+
+  rpc Consume(Consumer) returns (Message) {}
+
+}
+
+message PublishResponse {
+}
+
+message Consumer {
+    string topic = 1;
+    uint64 timeout_ms = 2;
+}
+
+message PublishMessage {
+  string topic = 1;
+  string message = 2;
+}
+
+message Message {
+    string message = 1;
+}
+

src/client.rs

@@ -0,0 +1,84 @@
+use clap::Parser;
+use pubsub_client::pub_sub_client::PubSubClient;
+use pubsub_client::PublishMessage;
+use pubsub_client::Consumer;
+use std::fmt;
+use std::time::Duration;
+
+pub mod pubsub_client {
+    tonic::include_proto!("pubsub");
+}
+
+#[derive(clap::ValueEnum, Clone, Debug)]
+enum Protocol {
+    Http,
+    Tcp,
+}
+
+#[derive(clap::Subcommand, Debug)]
+enum Command {
+    Publish { topic: String, message: String },
+    Consume { 
+        topic: String, 
+        #[clap(short, long, default_value_t = 10)]
+        timeout: u64},
+}
+
+impl fmt::Display for Protocol {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            Protocol::Http => write!(f, "http"),
+            Protocol::Tcp => write!(f, "tcp"),
+        }
+    }
+}
+
+#[derive(Parser, Debug)]
+#[clap(author, version, about, long_about = None)]
+struct Cli {
+    #[clap(short, long, value_parser, default_value_t = 50051)]
+    port: u16,
+
+    #[clap(long, value_enum, default_value_t = Protocol::Tcp)]
+    protocol: Protocol,
+
+    #[clap(short, long, value_parser, default_value = "localhost")]
+    endpoint: String,
+
+    #[clap(subcommand)]
+    command: Command,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let args = Cli::parse();
+
+    let mut client = PubSubClient::connect(format!(
+        "{}://{}:{}",
+        args.protocol, args.endpoint, args.port
+    ))
+    .await?;
+
+    match args.command {
+        Command::Consume { topic, timeout } => {
+            let mut request = tonic::Request::new(Consumer { topic: topic.clone(), timeout_ms: timeout * 1000 });
+            //Add 1s to the client-side timeout to let 
+            request.set_timeout(Duration::from_secs(timeout + 1));
+            match client.consume(request).await {
+                Err(e) => eprintln!("Error consuming {}\n{}", topic, e),
+                Ok(msg) => println!("{}", msg.get_ref().message)
+            }
+        }
+        Command::Publish { topic, message } => {
+            let request = tonic::Request::new(PublishMessage {
+                topic: topic.clone(),
+                message: message.clone(),
+            });
+            if let Err(e) = client.publish(request).await {
+                eprintln!("Error publishing {}:{}\n{}", topic, message, e);
+            }
+        }
+    }
+
+    Ok(())
+}