Add counts and add_new_item methods, without tests

go/WorkQueue.go

@@ -68,6 +68,8 @@ import (
 	"context"
 	"time"
 
+	//"fmt"
+
 	"github.com/google/uuid"
 	"github.com/redis/go-redis/v9"
 )
@@ -120,6 +122,77 @@ func (workQueue *WorkQueue) AddItem(ctx context.Context, db *redis.Client, item
 	return err
 }
 
+// **************-This method can be used only when database is locked-**************
+func (workQueue *WorkQueue) AddAtomicItem(ctx context.Context, db *redis.Client, item Item) error {
+	txf := func(tx *redis.Tx) error {
+		pipe := db.Pipeline()
+
+		mainQueueLenCmd := pipe.LLen(ctx, workQueue.mainQueueKey)
+		processingQueueLenCmd := pipe.LLen(ctx, workQueue.processingKey)
+
+		_, err := pipe.Exec(ctx)
+
+		mainQueueLen, err := mainQueueLenCmd.Result()
+
+		processingQueueLen, err := processingQueueLenCmd.Result()
+
+		pipeNew := db.Pipeline()
+		processingItemsInQueue, err := pipeNew.LPos(ctx, workQueue.processingKey, item.ID, redis.LPosArgs{
+			Rank:   1,
+			MaxLen: processingQueueLen,
+		}).Result()
+
+		workingItemsInQueue, err := pipeNew.LPos(ctx, workQueue.mainQueueKey, item.ID, redis.LPosArgs{
+			Rank:   1,
+			MaxLen: mainQueueLen,
+		}).Result()
+
+		pipeNew.Exec(ctx)
+
+		if workingItemsInQueue == 0 || processingItemsInQueue == 0 {
+			return nil
+		}
+
+		pipeEx := tx.Pipeline()
+		workQueue.AddItemToPipeline(ctx, pipeEx, item)
+		_, err = pipeEx.Exec(ctx)
+		if err != nil {
+			return err
+		}
+		return nil
+	}
+
+	err := db.Watch(ctx, txf, workQueue.mainQueueKey, workQueue.processingKey)
+	//fmt.Println(err)
+	for err == redis.TxFailedErr {
+		if err == nil {
+			// Success.
+			return nil
+		} else if err == redis.TxFailedErr {
+			err = db.Watch(ctx, txf, workQueue.mainQueueKey, workQueue.processingKey)
+		}
+	}
+	return nil
+}
+
+func (workQueue *WorkQueue) GetQueueLengths(ctx context.Context, db *redis.Client) ([]int64, error) {
+	tx := db.TxPipeline()
+
+	queueLen := tx.LLen(ctx, workQueue.mainQueueKey)
+	processingLen := tx.LLen(ctx, workQueue.processingKey)
+
+	_, err := tx.Exec(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	result := make([]int64, 2)
+	result[0], _ = queueLen.Result()
+	result[1], _ = processingLen.Result()
+
+	return result, nil
+}
+
 // 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 +277,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,35 @@ 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.
+   *
+   * @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 | null> {
+    return new Promise<boolean | null>((resolve, reject) => {
+      db.multi()
+        .lrange(this.mainQueueKey, 0, -1)
+        .lrange(this.processingKey, 0, -1)
+        .exec(async (err, results) => {
+          if (
+            results &&
+            ((results[0][1] as string[]).includes(item.id) ||
+              (results[1][1] as string[]).includes(item.id))
+          ) {
+            resolve(null);
+          } else {
+            const pipeline = db.pipeline();
+            this.addItemToPipeline(pipeline, item);
+            await pipeline.exec();
+            resolve(true);
+          }
+        });
+    });
+  }
+
   /**
    * Add an item to the work queue.
    * This creates a pipeline and executes it on the database.
@@ -70,6 +99,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,24 @@ def add_item(self, db: Redis, item: Item) -> None:
         pipeline = db.pipeline()
         self.add_item_to_pipeline(pipeline, item)
         pipeline.execute()
+
+    def add__atomic_item(self, db: Redis, item: Item) -> None:
+        """Add an item to the work queue in a atomic way.
+
+        This creates a pipeline and executes it on the database if the item you are trying to add is not in either the main or the processing queue.
+        """
+        pipeline = db.pipeline(transaction=True) 
+
+        main_items = pipeline.lrange(self._main_queue_key, 0, -1)
+        processing_items = pipeline.lrange(self._processing_key, 0, -1)
+        item_id = item.id().encode('utf-8')
+
+        if item_id in main_items or item_id in processing_items:
+            # print("Same Item tried being added twice.")
+            return None
+
+        self.add_item_to_pipeline(pipeline, item)
+        pipeline.execute()  
 
     def queue_len(self, db: Redis) -> int:
         """Return the length of the work queue (not including items being processed, see
@@ -47,6 +65,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/main.go

@@ -6,12 +6,15 @@ import (
 	"fmt"
 	"os"
 	"time"
+	"strconv"
+	"math/rand"
 
 	"github.com/redis/go-redis/v9"
 
 	workqueue "github.com/mevitae/redis-work-queue/go"
 )
 
+
 type SharedJobData struct {
 	A int `json:"a"`
 	B int `json:"b"`
@@ -34,6 +37,82 @@ func main() {
 	})
 	ctx := context.Background()
 
+	keyPrefix := workqueue.KeyPrefix("Duplicate-Items-Test")
+	workQueue := workqueue.NewWorkQueue(keyPrefix)
+
+	passed := 0
+	notPassed := 0
+	numberOfTests := 150
+	numberOfRandomPossibleItems := 232
+	items := make([]workqueue.Item, 0)
+
+	for i := 1; i <= numberOfTests; i++ {
+		itemData1 := strconv.Itoa(rand.Intn(numberOfRandomPossibleItems))
+		itemData2 := strconv.Itoa(rand.Intn(numberOfRandomPossibleItems))
+		item1 := workqueue.Item{
+			Data: []byte(itemData1),
+			ID:   itemData1,
+		}
+		item2 := workqueue.Item{
+			Data: []byte(itemData2),
+			ID:   itemData2,
+		}
+		items = append(items, item1, item2)
+
+		result1 := workQueue.AddAtomicItem(ctx, db, item1)
+		result2 := workQueue.AddAtomicItem(ctx, db, item2)
+
+		if result1 == nil {
+			passed++
+		} else {
+			notPassed++
+		}
+
+		if result2 == nil {
+			passed++
+		} else {
+			notPassed++
+		}
+
+		if rand.Intn(10) < 3 {
+			workQueue.Lease(ctx, db, true, time.Second*4, time.Second*1)
+		}
+
+		if rand.Intn(10) < 5 {
+			toRemove := &items[len(items)-1]
+			items = items[:len(items)-1]
+			workQueue.Complete(ctx, db, toRemove)
+		}
+	}
+
+	expectedTotal := numberOfTests * 2
+	if notPassed+passed != expectedTotal {
+		panic(fmt.Sprintf("The number of passed items (%d) with the number of not passed items (%d) should be equal to the number of tests * 2: %d", passed, notPassed, expectedTotal))
+	}
+
+	mainQueueKey := keyPrefix.Of(":queue")
+	processingKey := keyPrefix.Of(":processing")
+	mainItems := db.LRange(ctx, mainQueueKey, 0, -1).Val()
+	processingItems := db.LRange(ctx, processingKey, 0, -1).Val()
+
+	for _, item := range mainItems {
+		if sliceContainsString(processingItems, item) {
+			panic("Found duplicated item from processing queue inside main queue")
+		}
+	}
+	for _, item := range processingItems {
+		if sliceContainsString(mainItems, item) {
+			panic("Found duplicated item from processing queue inside main queue")
+		}
+	}
+
+
+
+
+
+
+
+
 	goResultsKey := workqueue.KeyPrefix("results:go:")
 	sharedResultsKey := workqueue.KeyPrefix("results:shared:")
 
@@ -46,6 +125,8 @@ func main() {
 	shared := false
 	for {
 		shared = !shared
+		fmt.Println(goQueue.GetQueueLengths(ctx,db))
+		fmt.Println(sharedQueue.GetQueueLengths(ctx,db))
 		if shared {
 			sharedJobCounter++
 
@@ -176,3 +257,12 @@ func main() {
 		}
 	}
 }
+
+func sliceContainsString(slice []string, target string) bool {
+	for _, s := range slice {
+		if s == target {
+			return true
+		}
+	}
+	return false
+}