Further simplify cron starting

core/tasks/analytics/cron.go

@@ -2,22 +2,17 @@ package analytics
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"github.com/nyaruka/librato"
 	"github.com/nyaruka/mailroom"
 	"github.com/nyaruka/mailroom/core/queue"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/sirupsen/logrus"
 )
 
 func init() {
-	mailroom.AddInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
-		cron.Start(rt, wg, "stats", time.Second*60, true, reportAnalytics, time.Minute*5, quit)
-		return nil
-	})
+	mailroom.RegisterCron("analytics", time.Second*60, true, reportAnalytics)
 }
 
 var (

core/tasks/campaigns/cron.go

@@ -3,7 +3,6 @@ package campaigns
 import (
 	"context"
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/gomodule/redigo/redis"
@@ -13,7 +12,6 @@ import (
 	"github.com/nyaruka/mailroom/core/models"
 	"github.com/nyaruka/mailroom/core/queue"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/nyaruka/redisx"
 
 	"github.com/pkg/errors"
@@ -27,10 +25,7 @@ const (
 var campaignsMarker = redisx.NewIntervalSet("campaign_event", time.Hour*24, 2)
 
 func init() {
-	mailroom.AddInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
-		cron.Start(rt, wg, "campaign_event", time.Second*60, false, QueueEventFires, time.Minute*5, quit)
-		return nil
-	})
+	mailroom.RegisterCron("campaign_event", time.Second*60, false, QueueEventFires)
 }
 
 // QueueEventFires looks for all due campaign event fires and queues them to be started

core/tasks/expirations/cron.go

@@ -3,15 +3,13 @@ package expirations
 import (
 	"context"
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/nyaruka/mailroom"
 	"github.com/nyaruka/mailroom/core/ivr"
 	"github.com/nyaruka/mailroom/core/models"
 	"github.com/nyaruka/mailroom/core/tasks/handler"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/nyaruka/redisx"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -24,11 +22,8 @@ const (
 var expirationsMarker = redisx.NewIntervalSet("run_expirations", time.Hour*24, 2)
 
 func init() {
-	mailroom.AddInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
-		cron.Start(rt, wg, "run_expirations", time.Minute, false, HandleWaitExpirations, time.Minute*5, quit)
-		cron.Start(rt, wg, "expire_ivr_calls", time.Minute, false, ExpireVoiceSessions, time.Minute*5, quit)
-		return nil
-	})
+	mailroom.RegisterCron("run_expirations", time.Minute, false, HandleWaitExpirations)
+	mailroom.RegisterCron("expire_ivr_calls", time.Minute, false, ExpireVoiceSessions)
 }
 
 // HandleWaitExpirations handles waiting messaging sessions whose waits have expired, resuming those that can be resumed,

core/tasks/handler/cron.go

@@ -4,27 +4,21 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/nyaruka/mailroom"
 	"github.com/nyaruka/mailroom/core/models"
 	"github.com/nyaruka/mailroom/core/queue"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/nyaruka/redisx"
-
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
 var retriedMsgs = redisx.NewIntervalSet("retried_msgs", time.Hour*24, 2)
 
 func init() {
-	mailroom.AddInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
-		cron.Start(rt, wg, "retry_msgs", time.Minute*5, false, RetryPendingMsgs, time.Minute*5, quit)
-		return nil
-	})
+	mailroom.RegisterCron("retry_msgs", time.Minute*5, false, RetryPendingMsgs)
 }
 
 // RetryPendingMsgs looks for any pending msgs older than five minutes and queues them to be handled again

core/tasks/incidents/end_incidents.go

@@ -3,24 +3,19 @@ package incidents
 import (
 	"context"
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/gomodule/redigo/redis"
 	"github.com/nyaruka/goflow/flows"
 	"github.com/nyaruka/mailroom"
 	"github.com/nyaruka/mailroom/core/models"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
 func init() {
-	mailroom.AddInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
-		cron.Start(rt, wg, "end_incidents", time.Minute*3, false, EndIncidents, time.Minute*5, quit)
-		return nil
-	})
+	mailroom.RegisterCron("end_incidents", time.Minute*3, false, EndIncidents)
 }
 
 // EndIncidents checks open incidents and end any that no longer apply

core/tasks/ivr/cron.go

@@ -2,23 +2,18 @@ package ivr
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"github.com/nyaruka/mailroom"
 	"github.com/nyaruka/mailroom/core/ivr"
 	"github.com/nyaruka/mailroom/core/models"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
 func init() {
-	mailroom.AddInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
-		cron.Start(rt, wg, "retry_ivr_calls", time.Minute, false, RetryCalls, time.Minute*5, quit)
-		return nil
-	})
+	mailroom.RegisterCron("retry_ivr_calls", time.Minute, false, RetryCalls)
 }
 
 // RetryCalls looks for calls that need to be retried and retries them

core/tasks/msgs/retries.go

@@ -2,23 +2,18 @@ package msgs
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"github.com/nyaruka/mailroom"
 	"github.com/nyaruka/mailroom/core/models"
 	"github.com/nyaruka/mailroom/core/msgio"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
 func init() {
-	mailroom.AddInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
-		cron.Start(rt, wg, "retry_errored_messages", time.Second*60, false, RetryErroredMessages, time.Minute*5, quit)
-		return nil
-	})
+	mailroom.RegisterCron("retry_errored_messages", time.Second*60, false, RetryErroredMessages)
 }
 
 func RetryErroredMessages(ctx context.Context, rt *runtime.Runtime) error {

core/tasks/schedules/cron.go

@@ -2,23 +2,18 @@ package schedules
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"github.com/nyaruka/mailroom"
 	"github.com/nyaruka/mailroom/core/models"
 	"github.com/nyaruka/mailroom/core/queue"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
 func init() {
-	mailroom.AddInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
-		cron.Start(rt, wg, "fire_schedules", time.Minute*1, false, checkSchedules, time.Minute*5, quit)
-		return nil
-	})
+	mailroom.RegisterCron("fire_schedules", time.Minute*1, false, checkSchedules)
 }
 
 // checkSchedules looks up any expired schedules and fires them, setting the next fire as needed

core/tasks/timeouts/cron.go

@@ -3,27 +3,21 @@ package timeouts
 import (
 	"context"
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/nyaruka/mailroom"
 	"github.com/nyaruka/mailroom/core/models"
 	"github.com/nyaruka/mailroom/core/tasks/handler"
 	"github.com/nyaruka/mailroom/runtime"
-	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/nyaruka/redisx"
-
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
 var marker = redisx.NewIntervalSet("session_timeouts", time.Hour*24, 2)
 
 func init() {
-	mailroom.AddInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
-		cron.Start(rt, wg, "sessions_timeouts", time.Second*60, false, timeoutSessions, time.Minute*5, quit)
-		return nil
-	})
+	mailroom.RegisterCron("sessions_timeouts", time.Second*60, false, timeoutSessions)
 }
 
 // timeoutRuns looks for any runs that have timed out and schedules for them to continue

mailroom.go

@@ -10,6 +10,7 @@ import (
 	"github.com/nyaruka/gocommon/storage"
 	"github.com/nyaruka/mailroom/core/queue"
 	"github.com/nyaruka/mailroom/runtime"
+	"github.com/nyaruka/mailroom/utils/cron"
 	"github.com/nyaruka/mailroom/web"
 	"github.com/pkg/errors"
 
@@ -21,15 +22,22 @@ import (
 )
 
 // InitFunction is a function that will be called when mailroom starts
-type InitFunction func(runtime *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error
+type InitFunction func(*runtime.Runtime, *sync.WaitGroup, chan bool) error
 
 var initFunctions = make([]InitFunction, 0)
 
-// AddInitFunction adds an init function that will be called on startup
-func AddInitFunction(initFunc InitFunction) {
+func addInitFunction(initFunc InitFunction) {
 	initFunctions = append(initFunctions, initFunc)
 }
 
+// RegisterCron registers a new cron function to run every interval
+func RegisterCron(name string, interval time.Duration, allInstances bool, fn cron.Function) {
+	addInitFunction(func(rt *runtime.Runtime, wg *sync.WaitGroup, quit chan bool) error {
+		cron.Start(rt, wg, name, interval, allInstances, fn, time.Minute*5, quit)
+		return nil
+	})
+}
+
 // TaskFunction is the function that will be called for a type of task
 type TaskFunction func(ctx context.Context, rt *runtime.Runtime, task *queue.Task) error