Merge branch 'master' of https://github.com/cgrates/cgrates

engine/action_timing.go

@@ -22,16 +22,14 @@ import (
 	"fmt"
 	"sort"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/cgrates/cgrates/utils"
 )
 
 const (
-	FORMAT     = "2006-1-2 15:04:05 MST"
-	ASAP       = "*asap"
-	ASAP_DELAY = "1m"
+	FORMAT = "2006-1-2 15:04:05 MST"
+	ASAP   = "*asap"
 )
 
 type ActionTiming struct {
@@ -257,25 +255,8 @@ func (at *ActionTiming) Execute() (err error) {
 	return
 }
 
-// checks for *asap string as start time and replaces it wit an actual time in the newar future
-// returns true if the *asap string was found
-func (at *ActionTiming) CheckForASAP() bool {
-	if at.Timing.Timing.StartTime == ASAP {
-		delay, _ := time.ParseDuration(ASAP_DELAY)
-		timeTokens := strings.Split(time.Now().Add(delay).Format(time.Stamp), " ")
-		at.Timing.Timing.StartTime = timeTokens[len(timeTokens)-1]
-		return true
-	}
-	return false
-}
-
-// returns true if only the starting time was is filled in the Timing field
-func (at *ActionTiming) IsOneTimeRun() bool {
-	return len(at.Timing.Timing.Years) == 0 &&
-		len(at.Timing.Timing.Months) == 0 &&
-		len(at.Timing.Timing.MonthDays) == 0 &&
-		len(at.Timing.Timing.WeekDays) == 0 &&
-		len(at.Timing.Timing.StartTime) != 0
+func (at *ActionTiming) IsASAP() bool {
+	return at.Timing.Timing.StartTime == ASAP
 }
 
 // Structure to store actions according to weight

engine/actions_test.go

@@ -212,6 +212,15 @@ func TestActionTimingOnlyYears(t *testing.T) {
 	}
 }
 
+func TestActionTimingPast(t *testing.T) {
+	at := &ActionTiming{Timing: &RateInterval{Timing: &RITiming{Years: utils.Years{2013}}}}
+	st := at.GetNextStartTime(referenceDate)
+	expected := time.Date(2013, 1, 1, 0, 0, 0, 0, time.Local)
+	if !st.Equal(expected) {
+		t.Errorf("Expected %v was %v", expected, st)
+	}
+}
+
 func TestActionTimingHourYears(t *testing.T) {
 
 	y, m, d := now.Date()
@@ -334,30 +343,11 @@ func TestActionTimingFirstMonthOfTheYear(t *testing.T) {
 
 func TestActionTimingCheckForASAP(t *testing.T) {
 	at := &ActionTiming{Timing: &RateInterval{Timing: &RITiming{StartTime: ASAP}}}
-	if !at.CheckForASAP() {
+	if !at.IsASAP() {
 		t.Errorf("%v should be asap!", at)
 	}
 }
 
-func TestActionTimingIsOneTimeRun(t *testing.T) {
-	at := &ActionTiming{Timing: &RateInterval{Timing: &RITiming{StartTime: ASAP}}}
-	if !at.CheckForASAP() {
-		t.Errorf("%v should be asap!", at)
-	}
-	if !at.IsOneTimeRun() {
-		t.Errorf("%v should be one time run!", at)
-	}
-}
-
-func TestActionTimingOneTimeRun(t *testing.T) {
-	at := &ActionTiming{Timing: &RateInterval{Timing: &RITiming{StartTime: ASAP}}}
-	at.CheckForASAP()
-	nextRun := at.GetNextStartTime(referenceDate)
-	if nextRun.IsZero() {
-		t.Error("next time failed for asap")
-	}
-}
-
 func TestActionTimingLogFunction(t *testing.T) {
 	a := &Action{
 		ActionType:  "*log",

scheduler/scheduler.go

@@ -49,9 +49,13 @@ func (s *Scheduler) Loop() {
 		if a0.GetNextStartTime(now).Equal(now) || a0.GetNextStartTime(now).Before(now) {
 			engine.Logger.Debug(fmt.Sprintf("%v - %v", a0.Id, a0.Timing))
 			go a0.Execute()
-			s.queue = append(s.queue, a0)
-			s.queue = s.queue[1:]
-			sort.Sort(s.queue)
+			// if after execute the next start time is in the past then
+			// do not add it to the queue
+			if !a0.GetNextStartTime(now).Before(now) {
+				s.queue = append(s.queue, a0)
+				s.queue = s.queue[1:]
+				sort.Sort(s.queue)
+			}
 			s.Unlock()
 		} else {
 			s.Unlock()
@@ -82,13 +86,18 @@ func (s *Scheduler) LoadActionTimings(storage engine.AccountingStorage) {
 		isAsap := false
 		newAts := make([]*engine.ActionTiming, 0) // will remove the one time runs from the database
 		for _, at := range ats {
-			isAsap = at.CheckForASAP()
+			isAsap = at.IsASAP()
 			toBeSaved = toBeSaved || isAsap
-			if at.IsOneTimeRun() {
+			if isAsap {
 				engine.Logger.Info(fmt.Sprintf("Time for one time action on %v", key))
 				go at.Execute()
 				// do not append it to the newAts list to be saved
 			} else {
+				now := time.Now()
+				if at.GetNextStartTime(now).Before(now) {
+					// the task is obsolete, do not add it to the queue
+					continue
+				}
 				s.queue = append(s.queue, at)
 				newAts = append(newAts, at)
 			}