Merge pull request #1 from mohamed-essam/add_time_fallback_option

Add timed queue option

crow.go

@@ -4,7 +4,8 @@ package murder
 // Interface for any storage system for the orchestrator to use
 type Crow interface {
 	QueueSize(string) int                   // Query main queue size
-	AddToQueue(string, interface{})         // Add object to queue
+	QueueTimeSinceCreation(string) int	// Query time since queue creation
+	AddToQueue(string, interface{}, bool)   // Add object to queue
 	GetQueueContents(string) []string       // Retrieve all contents of queue
 	ClearQueue(string, string) error        // Clear all queue contents
 	CreateLockKey(string, string, int) bool // Create lock key for a queue, confirm if lock acquired, and set TTL

murder.go

@@ -8,17 +8,23 @@ type Murder struct {
 	queueSize     int
 	lockTTL       int
 	workerGroupID string
+	queueAge      int
 }
 
 // Add :
 // Create a job in any queue
 func (m *Murder) Add(obj interface{}) {
 	size := m.crow.QueueSize(m.workerGroupID)
-	if size >= m.queueSize {
+	ageConfigured := m.AgeConfigured()
+	var age int
+	if (ageConfigured) {
+		age = m.crow.QueueTimeSinceCreation(m.workerGroupID)
+	}
+	if size >= m.queueSize || (ageConfigured && age > m.queueAge) {
 		queueName := newUUID()
 		m.crow.MoveToReady(m.workerGroupID, queueName)
 	}
-	m.crow.AddToQueue(m.workerGroupID, obj)
+	m.crow.AddToQueue(m.workerGroupID,  obj, ageConfigured)
 }
 
 // Lock :
@@ -79,6 +85,10 @@ func (m *Murder) Unlock(lockKey string) {
 	m.crow.RemoveLockKey(lockKey)
 }
 
+func (m *Murder) AgeConfigured() bool {
+	return (m.queueAge > 0)
+}
+
 // NewMurder :
 // Returns a new instance of murder with the given options
 func NewMurder(bulkSize, TTL int, crow Crow, groupID string) *Murder {
@@ -89,3 +99,13 @@ func NewMurder(bulkSize, TTL int, crow Crow, groupID string) *Murder {
 		workerGroupID: groupID,
 	}
 }
+
+
+// MurderWithAge:
+// Returns a new instance of murder with extra option of queue age
+
+func NewMurderWithAge(bulkSize, TTL int, crow Crow, groupID string, queueAge int) *Murder {
+	murder := NewMurder(bulkSize, TTL, crow, groupID)
+	murder.queueAge = queueAge
+	return murder
+}

redis-crow.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"log"
 	"time"
-
+	"strconv"
 	"gopkg.in/redis.v5"
 )
 
@@ -18,12 +18,26 @@ func (c *RedisCrow) QueueSize(groupName string) int {
 	return int(size)
 }
 
+func (c *RedisCrow) QueueTimeSinceCreation(groupName string) int {
+	queueAgeKey := fmt.Sprintf("murder::%s::age", c.CurrentQueue(groupName))
+	timeSinceCreation, _ := c.Redis.Get(queueAgeKey).Result()
+	timeSinceCreationInt, _ := strconv.ParseInt(timeSinceCreation, 10, 64)
+	age := time.Now().Unix() - timeSinceCreationInt
+	return int(age)
+}
+
+
 func (c *RedisCrow) CurrentQueue(groupName string) string {
 	return fmt.Sprintf("murder::%s::mainQueue", groupName)
 }
 
-func (c *RedisCrow) AddToQueue(groupName string, obj interface{}) {
+func (c *RedisCrow) AddToQueue(groupName string, obj interface{}, ageConfigured bool) {
 	marshalled, _ := json.Marshal(obj)
+	currentQueue := c.CurrentQueue(groupName)
+	if !c.Redis.Exists(currentQueue).Val() && ageConfigured {
+		queueAgeKey := fmt.Sprintf("murder::%s::age", currentQueue)
+		c.Redis.Set(queueAgeKey, time.Now().Unix(), time.Duration(0))
+	}
 	c.Redis.LPush(c.CurrentQueue(groupName), marshalled).Result()
 }
 
@@ -81,6 +95,7 @@ func (c *RedisCrow) RemoveLockKey(lockKey string) {
 func (c *RedisCrow) MoveToReady(groupName, newName string) {
 	ok, _ := c.Redis.SetNX(fmt.Sprintf("%s::lock", c.CurrentQueue(groupName)), "1", time.Duration(1)*time.Second).Result()
 	if ok {
+		c.Redis.Del(fmt.Sprintf("murder::%s::age", c.CurrentQueue(groupName)))
 		c.Redis.Rename(c.CurrentQueue(groupName), fmt.Sprintf("murder::crows::%s", newName))
 		c.Redis.SAdd(fmt.Sprintf("murder::%s::ready", groupName), newName)
 		c.Redis.Del(fmt.Sprintf("%s::lock", c.CurrentQueue(groupName)))