Redis broker impl, simple example with Redis

Simple implementation of using Redis as a message broker. Together with
a simple example application using the Redis broker.

Author: magnunm

Cargo.lock

@@ -16,3 +16,4 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 ulid = "1.0"
 anyhow = "1.0.71"
+redis = "0.23"

Cargo.toml

@@ -2,3 +2,13 @@
 
 A Rust library for distributed, asynchronous computing using a task queue.
 Inspired by [Celery](https://docs.celeryq.dev/en/stable/).
+
+## Examples
+
+### `simple_redis_queue`
+
+Spin up a Redis instance and pass the connection URL as the first argument.
+
+```sh
+cargo r --example simple_redis_queue 'redis://localhost:6379/1'
+```

README.md

@@ -0,0 +1,46 @@
+use anyhow;
+use parsnip::{self, brokers::redis::RedisBroker, task::Signature, task::Task, worker::Worker};
+use std::env;
+
+struct HelloWorldTask {
+    called_with_signature: Signature<Self>,
+}
+
+impl Task for HelloWorldTask {
+    type ArgumentType = ();
+    type ReturnType = ();
+
+    const ID: &'static str = "HelloWorldTask";
+
+    fn from_signature(signature: Signature<Self>) -> Self {
+        Self {
+            called_with_signature: signature,
+        }
+    }
+
+    fn run(_: &Self::ArgumentType) -> Self::ReturnType {
+        println!("Hello, World!");
+    }
+
+    fn signature(&self) -> &Signature<Self> {
+        &self.called_with_signature
+    }
+}
+
+fn main() -> Result<(), anyhow::Error> {
+    let connect_url: String = env::args()
+        .nth(1)
+        .expect("No Redis connect URL passed the first an argument");
+
+    let mut broker = RedisBroker::new(&connect_url)?;
+
+    let mut app = parsnip::App::new(&mut broker);
+    app.register_task::<HelloWorldTask>();
+
+    let worker = Worker::new(&app);
+
+    app.queue_task::<HelloWorldTask>(())?;
+    worker.take_first_task_in_queue()?;
+
+    Ok(())
+}

examples/simple_redis_queue.rs

@@ -0,0 +1 @@
+pub mod redis;

src/brokers/mod.rs

@@ -0,0 +1,50 @@
+use crate::broker::Broker;
+
+use anyhow::Result;
+use redis::{self, Commands};
+use serde_json;
+
+pub struct RedisBroker {
+    redis_client: redis::Client,
+    queue: String,
+    result_hash_map: String,
+}
+
+impl RedisBroker {
+    pub fn new(connect_url: &str) -> Result<Self> {
+        let redis_client = redis::Client::open(connect_url)?;
+
+        Ok(Self {
+            redis_client,
+            queue: "parsnip_queue".to_string(),
+            result_hash_map: "parsnip_task_results".to_string(),
+        })
+    }
+}
+
+impl Broker for RedisBroker {
+    fn push_message(&self, message: &crate::messages::Message) -> Result<()> {
+        let mut con = self.redis_client.get_connection()?;
+        con.lpush(&self.queue, serde_json::to_string(&message)?)?;
+        Ok(())
+    }
+
+    fn pop_message(&self) -> Result<crate::messages::Message> {
+        let mut con = self.redis_client.get_connection()?;
+        let serialized_message: Option<String> = con.rpop(&self.queue, None)?;
+        match serialized_message {
+            Some(v) => Ok(serde_json::from_str(&v)?),
+            None => Err(anyhow::anyhow!("Task queue is empty")),
+        }
+    }
+
+    fn store_result(&self, result_message: crate::messages::ResultMessage) -> Result<()> {
+        let mut con = self.redis_client.get_connection()?;
+        con.hset(
+            &self.result_hash_map,
+            &result_message.signature_id,
+            serde_json::to_string(&result_message)?,
+        )?;
+        Ok(())
+    }
+}

src/brokers/redis.rs

@@ -1,4 +1,5 @@
 pub mod broker;
+pub mod brokers;
 pub mod messages;
 mod runner;
 pub mod task;