Hackdays

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "rust-analyzer.checkOnSave.command": "clippy"
+}

README.md

@@ -5,9 +5,54 @@
 
 
 # Zeebe Rust Client
-Rust client for Zeebe
+Rust client for Zeebe in early stages. Suitable for early adopters
 
-In very early stages.
+Features
+* CLI for all commands based on Zeebe Client 8.0
+* Support for OAuth authentication
+
+Next Steps
+1. Worker implementation
+2. Publish crates
+3. Build an application that uses this Rust client
+
+## Cli Tool
+
+> **Warning**
+> The Cli Tool might leak credentials in log messages
+
+Run `cargo run -- help` to see available commands and options.
+
+**Authentication for Camunda Cloud**
+
+First, [generate and download](https://docs.camunda.io/docs/next/components/console/manage-clusters/manage-api-clients/) client credentials from Console. Let's assume they are in a file called credentials.txt.
+
+The `credentials.txt` file contains environment variables, so let's source them:
+```shell
+$ source credentials.txt
+```
+
+Finally, run the cli tool: 
+
+```shell
+$ cargo run -- status
+TopologyResponse {
+    brokers: [
+        BrokerInfo {
+            node_id: 0,
+...
+```
+
+Alternatively, you can also provide your credentials as arguments:
+
+```shell
+$ cargo run -- --address <ADDRESS> --client-id <CLIENT_ID> --client_secret <CLIENT_SECRET> --authorization-server <AUTH_SERVER> status 
+TopologyResponse {
+    brokers: [
+        BrokerInfo {
+            node_id: 0,
+...
+```
 
 ## Prior Work/Alternatives
 
@@ -18,3 +63,13 @@ These repositories also implement Zeebe clients for Rust. Most of them are more
 
 Your best choice is probably: https://github.com/OutThereLabs/zeebe-rust
 
+
+# Developer Guide
+
+## CLI
+
+**Conventions for CLI Argument Annotations**
+* Named parameters by default
+* Use positional parameters only if there is just one parameter, or just one required parameter
+* For required parameters use short and long version `#[clap(short, long)]`
+* For optional parameters use long version only `#[clap(long, default_value_t = 1)]`

cli/Cargo.toml

@@ -7,5 +7,10 @@ edition = "2021"
 zeebe-client = { path = "../client" }
 clap = { version = "3.2", features = ["derive", "env"] }
 color-eyre = "0.6"
-tonic = "0.8"
-tokio = { version = "1", features=["full"] }
+tonic = { version = "0.8", features = ["tls", "tls-roots", "gzip"] }
+tokio = { version = "1", features=["full"] }
+tracing = { version="0.1.36", features = ["async-await", "log"] }
+tracing-subscriber = { version = "0.3.15", features = ["env-filter"] }
+tracing-error = { version = "0.2", features = ["traced-error"] }
+tracing-tree = { version = "0.2" }
+async-trait = "0.1.57"

cli/src/activate_jobs.rs

@@ -0,0 +1,53 @@
+use crate::{Debug, ExecuteZeebeCommand};
+use async_trait::async_trait;
+use clap::Args;
+use color_eyre::Result;
+
+use zeebe_client::{
+    api::{ActivateJobsRequest, ActivateJobsResponse},
+    ZeebeClient,
+};
+
+#[derive(Debug, Args)]
+pub(crate) struct ActivateJobsArgs {
+    job_type: String,
+
+    #[clap(short, long, default_value_t = 10)]
+    max_jobs_to_activate: usize,
+    #[clap(short= 't', long, default_value_t = 5 * 60 * 1000)]
+    job_timeout: u64, // todo: should be duration
+    #[clap(long, required = false, default_value = "worker")]
+    worker: String,
+    #[clap(long, required = false)]
+    variables: Vec<String>,
+}
+
+impl From<&ActivateJobsArgs> for ActivateJobsRequest {
+    fn from(args: &ActivateJobsArgs) -> Self {
+        ActivateJobsRequest {
+            r#type: args.job_type.to_owned(),
+            worker: args.worker.to_owned(),
+            timeout: args.job_timeout as i64,
+            max_jobs_to_activate: args.max_jobs_to_activate as i32,
+            fetch_variable: args.variables.to_owned(),
+            request_timeout: Default::default(),
+        }
+    }
+}
+
+#[async_trait]
+impl ExecuteZeebeCommand for ActivateJobsArgs {
+    type Output = Vec<ActivateJobsResponse>;
+
+    #[tracing::instrument(skip(client))]
+    async fn execute(self, client: &mut ZeebeClient) -> Result<Self::Output> {
+        let args = &self;
+        let request: ActivateJobsRequest = args.into();
+        let mut stream = client.activate_jobs(request).await?.into_inner();
+        let mut result = Vec::with_capacity(args.max_jobs_to_activate);
+        while let Some(response) = stream.message().await? {
+            result.push(response);
+        }
+        Ok(result)
+    }
+}

cli/src/cancel_process_instance.rs

@@ -0,0 +1,34 @@
+use async_trait::async_trait;
+use color_eyre::eyre::Result;
+
+use clap::Args;
+use zeebe_client::{
+    api::{CancelProcessInstanceRequest, CancelProcessInstanceResponse},
+    ZeebeClient,
+};
+
+use crate::ExecuteZeebeCommand;
+
+#[derive(Args)]
+pub(crate) struct CancelProcessInstanceArgs {
+    process_instance_key: i64,
+}
+
+impl From<&CancelProcessInstanceArgs> for CancelProcessInstanceRequest {
+    fn from(args: &CancelProcessInstanceArgs) -> Self {
+        CancelProcessInstanceRequest {
+            process_instance_key: args.process_instance_key,
+        }
+    }
+}
+
+#[async_trait]
+impl ExecuteZeebeCommand for CancelProcessInstanceArgs {
+    type Output = CancelProcessInstanceResponse;
+    async fn execute(self, client: &mut ZeebeClient) -> Result<Self::Output> {
+        Ok(client
+            .cancel_process_instance(CancelProcessInstanceRequest::from(&self))
+            .await?
+            .into_inner())
+    }
+}

cli/src/complete_job.rs

@@ -0,0 +1,37 @@
+use crate::{Debug, ExecuteZeebeCommand};
+use async_trait::async_trait;
+use clap::Args;
+use color_eyre::eyre::Result;
+use zeebe_client::{
+    api::{CompleteJobRequest, CompleteJobResponse},
+    ZeebeClient,
+};
+
+#[derive(Args, Clone, Debug)]
+pub(crate) struct CompleteJobArgs {
+    job_key: i64,
+
+    #[clap(long, required = false, default_value = "")]
+    variables: String,
+}
+
+impl From<&CompleteJobArgs> for CompleteJobRequest {
+    fn from(args: &CompleteJobArgs) -> Self {
+        CompleteJobRequest {
+            job_key: args.job_key,
+            variables: args.variables.clone(),
+        }
+    }
+}
+
+#[async_trait]
+impl ExecuteZeebeCommand for CompleteJobArgs {
+    type Output = CompleteJobResponse;
+
+    #[tracing::instrument(skip(client))]
+    async fn execute(self, client: &mut ZeebeClient) -> Result<Self::Output> {
+        let args = &self;
+        let request: CompleteJobRequest = args.into();
+        Ok(client.complete_job(request).await?.into_inner())
+    }
+}

cli/src/create_process_instance.rs

@@ -0,0 +1,63 @@
+use crate::{Debug, ExecuteZeebeCommand};
+use async_trait::async_trait;
+use clap::Args;
+use color_eyre::eyre::Result;
+use zeebe_client::{
+    api::{CreateProcessInstanceRequest, CreateProcessInstanceWithResultRequest},
+    ZeebeClient,
+};
+
+#[derive(Args, Clone, Debug)]
+pub(crate) struct CreateProcessInstanceArgs {
+    process_instance_key: i64,
+
+    #[clap(long, required = false)]
+    with_results: bool,
+    #[clap(long, required = false, default_value = "")]
+    variables: String,
+    #[clap(long, required = false, default_value_t = -1)]
+    version: i32,
+}
+
+impl From<&CreateProcessInstanceArgs> for CreateProcessInstanceRequest {
+    fn from(args: &CreateProcessInstanceArgs) -> Self {
+        CreateProcessInstanceRequest {
+            process_definition_key: args.process_instance_key,
+            bpmn_process_id: String::new(),
+            version: args.version,
+            variables: args.variables.clone(),
+            start_instructions: vec![],
+        }
+    }
+}
+
+#[async_trait]
+impl ExecuteZeebeCommand for CreateProcessInstanceArgs {
+    type Output = Box<dyn Debug>;
+
+    #[tracing::instrument(skip(client))]
+    async fn execute(self, client: &mut ZeebeClient) -> Result<Self::Output> {
+        handle_create_instance_command(client, &self).await
+    }
+}
+
+async fn handle_create_instance_command(
+    client: &mut ZeebeClient,
+    args: &CreateProcessInstanceArgs,
+) -> Result<Box<dyn Debug>> {
+    let request: CreateProcessInstanceRequest = args.into();
+    match args.with_results {
+        true => Ok(Box::new(
+            client
+                .create_process_instance_with_result(CreateProcessInstanceWithResultRequest {
+                    request: Some(request),
+                    ..Default::default()
+                })
+                .await?
+                .into_inner(),
+        )),
+        false => Ok(Box::new(
+            client.create_process_instance(request).await?.into_inner(),
+        )),
+    }
+}

cli/src/deploy_resource.rs

@@ -0,0 +1,50 @@
+use std::path::PathBuf;
+
+use async_trait::async_trait;
+use clap::Args;
+
+use zeebe_client::{
+    api::{DeployResourceRequest, DeployResourceResponse, Resource},
+    ZeebeClient,
+};
+
+use crate::ExecuteZeebeCommand;
+use color_eyre::Result;
+
+#[derive(Args)]
+pub(crate) struct DeployResourceArgs {
+    #[clap(required = true, value_parser, value_name = "FILE")]
+    resources: Vec<PathBuf>,
+}
+#[async_trait]
+impl ExecuteZeebeCommand for DeployResourceArgs {
+    type Output = DeployResourceResponse;
+
+    async fn execute(self, client: &mut ZeebeClient) -> Result<Self::Output> {
+        Ok(client
+            .deploy_resource(DeployResourceRequest::try_from(&self)?)
+            .await?
+            .into_inner())
+    }
+}
+
+impl TryFrom<&DeployResourceArgs> for DeployResourceRequest {
+    type Error = color_eyre::Report;
+
+    fn try_from(args: &DeployResourceArgs) -> Result<DeployResourceRequest, Self::Error> {
+        let mut resources = Vec::with_capacity(args.resources.len());
+        for path in &args.resources {
+            let resource = Resource {
+                name: path
+                    .file_name()
+                    .expect("resource path should point to a file")
+                    .to_str()
+                    .expect("file name should be UTF-8")
+                    .to_string(),
+                content: std::fs::read(path)?,
+            };
+            resources.push(resource);
+        }
+        Ok(Self { resources })
+    }
+}

cli/src/fail_job.rs

@@ -0,0 +1,50 @@
+use async_trait::async_trait;
+use color_eyre::eyre::Result;
+
+use crate::ExecuteZeebeCommand;
+use clap::Args;
+
+use zeebe_client::{
+    api::{FailJobRequest, FailJobResponse},
+    ZeebeClient,
+};
+#[derive(Args)]
+pub(crate) struct FailJobArgs {
+    // the unique job identifier, as obtained when activating the job
+    #[clap(short, long)]
+    job_key: i64,
+    // the amount of retries the job should have left
+    #[clap(short, long)]
+    retries: i32,
+    // an optional message describing why the job failed
+    // this is particularly useful if a job runs out of retries and an incident is raised,
+    // as it this message can help explain why an incident was raised
+    #[clap(long, required = false, default_value = "")]
+    error_message: String,
+    // the back off timeout for the next retry
+    #[clap(long, required = false, default_value_t = 0)]
+    retry_back_off: i64,
+}
+
+impl From<&FailJobArgs> for FailJobRequest {
+    fn from(args: &FailJobArgs) -> Self {
+        FailJobRequest {
+            job_key: args.job_key,
+            retries: args.retries,
+            error_message: args.error_message.to_owned(),
+            retry_back_off: args.retry_back_off,
+        }
+    }
+}
+
+#[async_trait]
+impl ExecuteZeebeCommand for FailJobArgs {
+    type Output = FailJobResponse;
+
+    async fn execute(self, client: &mut ZeebeClient) -> Result<Self::Output> {
+        Ok(client
+            .fail_job(FailJobRequest::from(&self))
+            .await?
+            .into_inner())
+    }
+}