Add counts and add_new_item methods, without tests

go/WorkQueue.go

@@ -66,6 +66,7 @@ package workqueue
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/google/uuid"
@@ -120,6 +121,72 @@ func (workQueue *WorkQueue) AddItem(ctx context.Context, db *redis.Client, item
 	return err
 }
 
+// AddItemAtomically to the work Queue.
+//
+// This creates a pipeline and executes it on the database if there is no other same item inside eitther main or processing queue.
+//
+// This method can be used only when database is locked
+func (workQueue *WorkQueue) AddItemAtomically(ctx context.Context, db *redis.Client, item Item) error {
+	txf := func(tx *redis.Tx) error {
+
+		pipeNew := db.Pipeline()
+
+		processingItemsInQueueCmd := pipeNew.LPos(ctx, workQueue.processingKey, item.ID, redis.LPosArgs{
+			Rank:   0,
+			MaxLen: 0,
+		})
+
+		workingItemsInQueueCmd := pipeNew.LPos(ctx, workQueue.mainQueueKey, item.ID, redis.LPosArgs{
+			Rank:   0,
+			MaxLen: 0,
+		})
+
+		_, err := pipeNew.Exec(ctx)
+
+		_, ProcessingQueueCheck := processingItemsInQueueCmd.Result()
+		_, WorkingQueueCheck := workingItemsInQueueCmd.Result()
+		if ProcessingQueueCheck == redis.Nil && WorkingQueueCheck == redis.Nil {
+
+			pipeEx := tx.Pipeline()
+			workQueue.AddItemToPipeline(ctx, pipeEx, item)
+			_, err = pipeEx.Exec(ctx)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	for {
+		err := db.Watch(ctx, txf, workQueue.mainQueueKey, workQueue.processingKey)
+		// Retry on transaction failure, return on anything else.
+		if err != redis.TxFailedErr {
+			return err
+		}
+	}
+}
+
+func (workQueue *WorkQueue) Lengths(ctx context.Context, db *redis.Client) (queueLen, processingLen int64, err error) {
+	tx := db.TxPipeline()
+
+	queueLen, err = tx.LLen(ctx, workQueue.mainQueueKey).Result()
+	if err != nil {
+		return 0, 0, err
+	}
+
+	processingLen, err = tx.LLen(ctx, workQueue.processingKey).Result()
+	if err != nil {
+		return 0, 0, err
+	}
+
+	_, err = tx.Exec(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+
+	return queueLen, processingLen, err
+}
+
 // Return the length of the work queue (not including items being processed, see
 // [WorkQueue.Processing]).
 func (workQueue *WorkQueue) QueueLen(ctx context.Context, db *redis.Client) (int64, error) {
@@ -204,3 +271,12 @@ func (workQueue *WorkQueue) Complete(ctx context.Context, db *redis.Client, item
 	})
 	return true, err
 }
+
+func sliceContainsString(slice []string, target string) bool {
+	for _, s := range slice {
+		if s == target {
+			return true
+		}
+	}
+	return false
+}

node/src/WorkQueue.ts

@@ -47,6 +47,64 @@ export class WorkQueue {
     pipeline.lpush(this.mainQueueKey, item.id);
   }
 
+  /**
+   * Add an item to the work queue in a atomic way.
+   * This creates a pipeline and executes it on the database if the item id is not already in either the main queue or the procesing queue.
+   * 
+   * This method can be used only when database is locked
+   * 
+   * @param {Redis} db The Redis Connection.
+   * @param item The item that will be executed using the method addItemToPipeline.
+   */
+  async addAtomicItem(db: Redis, item: Item): Promise<boolean> {
+    const result = async (): Promise<boolean> => {
+      let transactionError = false;
+      do {
+        try {
+          await db.watch(this.mainQueueKey, this.processingKey);
+
+          const pipeNew = db.pipeline();
+
+          pipeNew.lpos(
+            this.processingKey,
+            item.id
+          );
+
+          pipeNew.lpos(
+            this.mainQueueKey,
+            item.id
+          );
+
+          const pipeResult = await pipeNew.exec();
+
+          console.log(pipeResult);
+          if (!pipeResult || pipeResult[0][1] !== null || pipeResult[1][1] !== null) {
+            return false; // Item already exists in either mainQueue or processingKey
+          }
+
+          const pipeEx = db.pipeline();
+          this.addItemToPipeline(pipeEx, item);
+          await pipeEx.exec();
+          return true;
+        } catch (error) {
+          transactionError = true;
+        } finally {
+          db.unwatch();
+        }
+      } while (transactionError);
+
+      return false; 
+    };
+
+    return await result(); // Invoke the async function to get the result
+  }
+
+
+
+
+
+
+
   /**
    * Add an item to the work queue.
    * This creates a pipeline and executes it on the database.
@@ -70,6 +128,23 @@ export class WorkQueue {
     return db.llen(this.mainQueueKey);
   }
 
+  /**
+   * This is used to ge the lenght of the Queues in a atomic way.
+   *
+   * @param {Redis} db The Redis Connection.
+   * @returns {Promise<[number, number]>} Return the length of main queue and processing queue.
+   */
+  async getQueueLengths(db: Redis): Promise<[number, number]> {
+    const multi = db.multi();
+    multi.llen(this.mainQueueKey);
+    multi.llen(this.processingKey);
+
+    const result = (await multi.exec()) as Array<[Error | null, number]>;
+    const queueLength = result[0][1];
+    const processingLength = result[1][1];
+    return [queueLength, processingLength];
+  }
+
   /**
    * This is used to get the lenght of the Processing Queue.
    *

python/redis_work_queue/workqueue.py

@@ -37,6 +37,34 @@ def add_item(self, db: Redis, item: Item) -> None:
         pipeline = db.pipeline()
         self.add_item_to_pipeline(pipeline, item)
         pipeline.execute()
+
+    def add_item_atomically(self, db: Redis, item: Item) -> None:
+        """Add an item to the work queue in an atomic way.
+
+        This checks for the existence of the item in either the main or the processing queue, and adds it to the main queue atomically.
+        """
+        pipeline = db.pipeline(transaction=True)
+        while True:
+            try:
+                pipeline.watch(self._main_queue_key, self._processing_key)
+
+                main_pos = pipeline.lpos(self._main_queue_key, item.id())
+                processing_pos = pipeline.lpos(self._processing_key, item.id())
+                if main_pos is not None or processing_pos is not None:
+                    break  
+
+                pipeline.multi()
+                self.add_item_to_pipeline(pipeline, item)
+                pipeline.execute()
+                break
+            except WatchError:
+                continue
+
+        pipeline.unwatch()
+
+
+
+
 
     def queue_len(self, db: Redis) -> int:
         """Return the length of the work queue (not including items being processed, see
@@ -47,6 +75,16 @@ def processing(self, db: Redis) -> int:
         """Return the number of items being processed."""
         return db.llen(self._processing_key)
 
+
+    def get_queue_lengths(self, db):
+        """Return the length of the work queue and processing queue"""
+        pipeline = db.pipeline(transaction=True)
+        pipeline.llen(self._main_queue_key)
+        pipeline.llen(self._processing_key)
+        return pipeline.execute()
+
+
+
     def light_clean(self, db: Redis) -> None:
         processing: list[bytes | str] = db.lrange(
             self._processing_key, 0, -1)

rust/src/lib.rs

@@ -280,12 +280,43 @@ impl WorkQueue {
     /// Add an item to the work queue.
     ///
     /// This creates a pipeline and executes it on the database.
+    pub async fn add_atomic_item <'a,C: AsyncCommands>(
+        &'a self,
+        connection: &mut C,
+        item: &Item,
+    ) -> Result<Option<bool>, redis::RedisError> {
+
+        let main_queue_key = self.main_queue_key.clone();
+        let processing_key = self.main_queue_key.clone();
+
+        let mut pipe = redis::pipe();
+
+        pipe
+            .atomic()
+            .cmd("LRANGE").arg(&main_queue_key).arg(0).arg(-1)
+            .cmd("LRANGE").arg(&processing_key).arg(0).arg(-1);
+
+        let result: Vec<Vec<String>> = pipe.query_async(connection).await?;
+
+        let main_items = &result[0];
+        let processing_items = &result[1];
+
+        if main_items.contains(&item.id) || processing_items.contains(&item.id) {
+            return Ok(None);
+        } else {
+            let mut pipeline = redis::pipe();
+            self.add_item_to_pipeline(&mut pipeline, &item);
+            pipeline.query_async(connection).await?;
+            return Ok(Some(true));
+        }
+    }
+
     pub async fn add_item<C: AsyncCommands>(&self, db: &mut C, item: &Item) -> RedisResult<()> {
         let mut pipeline = Box::new(redis::pipe());
         self.add_item_to_pipeline(&mut pipeline, item);
         pipeline.query_async(db).await
     }
-
+    
     /// Return the length of the work queue (not including items being processed, see
     /// [`WorkQueue::processing`]).
     pub fn queue_len<'a, C: AsyncCommands>(
@@ -303,6 +334,20 @@ impl WorkQueue {
         db.llen(&self.processing_key)
     }
 
+    pub async fn get_queue_lengths<'a, C: AsyncCommands>(
+        &'a self,
+        db: &mut C,
+    ) -> RedisResult<Vec<i64>> {
+        let (queue_length, processing_length): (i64, i64) = redis::pipe()
+            .atomic()
+            .llen(&self.main_queue_key)
+            .llen(&self.processing_key)
+            .query_async(db)
+            .await?;
+
+        Ok(vec![queue_length, processing_length])
+    }
+
     /// Request a work lease the work queue. This should be called by a worker to get work to
     /// complete. When completed, the `complete` method should be called.
     ///

tests/go/go.mod

@@ -6,10 +6,11 @@ go 1.20
 
 require (
 	github.com/mevitae/redis-work-queue/go v0.1.0
-	github.com/redis/go-redis/v9 v9.0.2
+	github.com/redis/go-redis/v9 v9.0.3
 )
 
 require (
+	github.com/bsm/redislock v0.9.3 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/google/uuid v1.3.0 // indirect

tests/go/go.sum

@@ -1,5 +1,9 @@
 github.com/bsm/ginkgo/v2 v2.5.0 h1:aOAnND1T40wEdAtkGSkvSICWeQ8L3UASX7YVCqQx+eQ=
+github.com/bsm/ginkgo/v2 v2.7.0 h1:ItPMPH90RbmZJt5GtkcNvIRuGEdwlBItdNVoyzaNQao=
 github.com/bsm/gomega v1.20.0 h1:JhAwLmtRzXFTx2AkALSLa8ijZafntmhSoU63Ok18Uq8=
+github.com/bsm/gomega v1.26.0 h1:LhQm+AFcgV2M0WyKroMASzAzCAJVpAxQXv4SaI9a69Y=
+github.com/bsm/redislock v0.9.3 h1:osmvugkXGiLDEhzUPdM0EUtKpTEgLLuli4Ky2Z4vx38=
+github.com/bsm/redislock v0.9.3/go.mod h1:Epf7AJLiSFwLCiZcfi6pWFO/8eAYrYpQXFxEDPoDeAk=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -10,5 +14,7 @@ github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/redis/go-redis/v9 v9.0.2 h1:BA426Zqe/7r56kCcvxYLWe1mkaz71LKF77GwgFzSxfE=
 github.com/redis/go-redis/v9 v9.0.2/go.mod h1:/xDTe9EF1LM61hek62Poq2nzQSGj0xSrEtEHbBQevps=
+github.com/redis/go-redis/v9 v9.0.3 h1:+7mmR26M0IvyLxGZUHxu4GiBkJkVDid0Un+j4ScYu4k=
+github.com/redis/go-redis/v9 v9.0.3/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk=
 github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=