Rename 'runner.Runner' to 'manager.Manager'

admin.go

@@ -30,7 +30,7 @@ func handleGetConfigScript(c echo.Context) error {
 			RootURL:    app.Constants.RootURL,
 			UploadURI:  app.Constants.UploadURI,
 			FromEmail:  app.Constants.FromEmail,
-			Messengers: app.Runner.GetMessengerNames(),
+			Messengers: app.Manager.GetMessengerNames(),
 		}
 
 		b = bytes.Buffer{}

campaigns.go

@@ -127,13 +127,13 @@ func handlePreviewCampaign(c echo.Context) error {
 		camp.Body = body
 	}
 
-	if err := camp.CompileTemplate(app.Runner.TemplateFuncs(camp)); err != nil {
+	if err := camp.CompileTemplate(app.Manager.TemplateFuncs(camp)); err != nil {
 		return echo.NewHTTPError(http.StatusBadRequest,
 			fmt.Sprintf("Error compiling template: %v", err))
 	}
 
 	// Render the message body.
-	m := app.Runner.NewMessage(camp, &sub)
+	m := app.Manager.NewMessage(camp, &sub)
 	if err := m.Render(); err != nil {
 		return echo.NewHTTPError(http.StatusBadRequest,
 			fmt.Sprintf("Error rendering message: %v", err))
@@ -159,7 +159,7 @@ func handleCreateCampaign(c echo.Context) error {
 		return echo.NewHTTPError(http.StatusBadRequest, err.Error())
 	}
 
-	if !app.Runner.HasMessenger(o.MessengerID) {
+	if !app.Manager.HasMessenger(o.MessengerID) {
 		return echo.NewHTTPError(http.StatusBadRequest,
 			fmt.Sprintf("Unknown messenger %s", o.MessengerID))
 	}
@@ -466,12 +466,12 @@ func handleTestCampaign(c echo.Context) error {
 
 // sendTestMessage takes a campaign and a subsriber and sends out a sample campain message.
 func sendTestMessage(sub *models.Subscriber, camp *models.Campaign, app *App) error {
-	if err := camp.CompileTemplate(app.Runner.TemplateFuncs(camp)); err != nil {
+	if err := camp.CompileTemplate(app.Manager.TemplateFuncs(camp)); err != nil {
 		return fmt.Errorf("Error compiling template: %v", err)
 	}
 
 	// Render the message body.
-	m := app.Runner.NewMessage(camp, sub)
+	m := app.Manager.NewMessage(camp, sub)
 	if err := m.Render(); err != nil {
 		return echo.NewHTTPError(http.StatusBadRequest,
 			fmt.Sprintf("Error rendering message: %v", err))
@@ -511,7 +511,7 @@ func validateCampaignFields(c campaignReq, app *App) error {
 	}
 
 	camp := models.Campaign{Body: c.Body, TemplateBody: tplTag}
-	if err := c.CompileTemplate(app.Runner.TemplateFuncs(&camp)); err != nil {
+	if err := c.CompileTemplate(app.Manager.TemplateFuncs(&camp)); err != nil {
 		return fmt.Errorf("Error compiling campaign body: %v", err)
 	}
 

main.go

@@ -12,8 +12,8 @@ import (
 	_ "github.com/jinzhu/gorm/dialects/postgres"
 	"github.com/jmoiron/sqlx"
 	"github.com/knadh/goyesql"
+	"github.com/knadh/listmonk/manager"
 	"github.com/knadh/listmonk/messenger"
-	"github.com/knadh/listmonk/runner"
 	"github.com/knadh/listmonk/subimporter"
 	"github.com/labstack/echo"
 	flag "github.com/spf13/pflag"
@@ -36,7 +36,7 @@ type App struct {
 	DB        *sqlx.DB
 	Queries   *Queries
 	Importer  *subimporter.Importer
-	Runner    *runner.Runner
+	Manager   *manager.Manager
 	Logger    *log.Logger
 	NotifTpls *template.Template
 	Messenger messenger.Messenger
@@ -158,7 +158,7 @@ func registerHandlers(e *echo.Echo) {
 }
 
 // initMessengers initializes various messaging backends.
-func initMessengers(r *runner.Runner) messenger.Messenger {
+func initMessengers(r *manager.Manager) messenger.Messenger {
 	// Load SMTP configurations for the default e-mail Messenger.
 	var srv []messenger.Server
 	for name := range viper.GetStringMapString("smtp") {
@@ -251,7 +251,7 @@ func main() {
 	campNotifCB := func(subject string, data map[string]interface{}) error {
 		return sendNotification(notifTplCampaign, subject, data, app)
 	}
-	r := runner.New(runner.Config{
+	m := manager.New(manager.Config{
 		Concurrency:   viper.GetInt("app.concurrency"),
 		MaxSendErrors: viper.GetInt("app.max_send_errors"),
 		FromEmail:     app.Constants.FromEmail,
@@ -264,14 +264,14 @@ func main() {
 
 		// url.com/campaign/{campaign_uuid}/{subscriber_uuid}/px.png
 		ViewTrackURL: fmt.Sprintf("%s/campaign/%%s/%%s/px.png", app.Constants.RootURL),
-	}, newRunnerDB(q), campNotifCB, logger)
-	app.Runner = r
+	}, newManagerDB(q), campNotifCB, logger)
+	app.Manager = m
 
 	// Add messengers.
-	app.Messenger = initMessengers(app.Runner)
+	app.Messenger = initMessengers(app.Manager)
 
-	go r.Run(time.Duration(time.Second * 5))
-	r.SpawnWorkers()
+	go m.Run(time.Duration(time.Second * 5))
+	m.SpawnWorkers()
 
 	// Initialize the server.
 	var srv = echo.New()

runner/runner.go → manager/manager.go

@@ -1,4 +1,4 @@
-package runner
+package manager
 
 import (
 	"bytes"
@@ -33,9 +33,9 @@ type DataSource interface {
 	CreateLink(url string) (string, error)
 }
 
-// Runner handles the scheduling, processing, and queuing of campaigns
+// Manager handles the scheduling, processing, and queuing of campaigns
 // and message pushes.
-type Runner struct {
+type Manager struct {
 	cfg        Config
 	src        DataSource
 	messengers map[string]messenger.Messenger
@@ -67,7 +67,7 @@ type Message struct {
 	to             string
 }
 
-// Config has parameters for configuring the runner.
+// Config has parameters for configuring the manager.
 type Config struct {
 	Concurrency    int
 	MaxSendErrors  int
@@ -84,8 +84,8 @@ type msgError struct {
 }
 
 // New returns a new instance of Mailer.
-func New(cfg Config, src DataSource, notifCB models.AdminNotifCallback, l *log.Logger) *Runner {
-	r := Runner{
+func New(cfg Config, src DataSource, notifCB models.AdminNotifCallback, l *log.Logger) *Manager {
+	return &Manager{
 		cfg:            cfg,
 		src:            src,
 		notifCB:        notifCB,
@@ -98,37 +98,35 @@ func New(cfg Config, src DataSource, notifCB models.AdminNotifCallback, l *log.L
 		msgErrorQueue:  make(chan msgError, cfg.MaxSendErrors),
 		msgErrorCounts: make(map[int]int),
 	}
-
-	return &r
 }
 
 // NewMessage creates and returns a Message that is made available
 // to message templates while they're compiled.
-func (r *Runner) NewMessage(c *models.Campaign, s *models.Subscriber) *Message {
+func (m *Manager) NewMessage(c *models.Campaign, s *models.Subscriber) *Message {
 	return &Message{
 		from:           c.FromEmail,
 		to:             s.Email,
 		Campaign:       c,
 		Subscriber:     s,
-		UnsubscribeURL: fmt.Sprintf(r.cfg.UnsubscribeURL, c.UUID, s.UUID),
+		UnsubscribeURL: fmt.Sprintf(m.cfg.UnsubscribeURL, c.UUID, s.UUID),
 	}
 }
 
-// AddMessenger adds a Messenger messaging backend to the runner process.
-func (r *Runner) AddMessenger(msg messenger.Messenger) error {
+// AddMessenger adds a Messenger messaging backend to the manager.
+func (m *Manager) AddMessenger(msg messenger.Messenger) error {
 	id := msg.Name()
-	if _, ok := r.messengers[id]; ok {
+	if _, ok := m.messengers[id]; ok {
 		return fmt.Errorf("messenger '%s' is already loaded", id)
 	}
-	r.messengers[id] = msg
+	m.messengers[id] = msg
 
 	return nil
 }
 
 // GetMessengerNames returns the list of registered messengers.
-func (r *Runner) GetMessengerNames() []string {
+func (m *Manager) GetMessengerNames() []string {
 	var names []string
-	for n := range r.messengers {
+	for n := range m.messengers {
 		names = append(names, n)
 	}
 
@@ -136,8 +134,8 @@ func (r *Runner) GetMessengerNames() []string {
 }
 
 // HasMessenger checks if a given messenger is registered.
-func (r *Runner) HasMessenger(id string) bool {
-	_, ok := r.messengers[id]
+func (m *Manager) HasMessenger(id string) bool {
+	_, ok := m.messengers[id]
 	return ok
 }
 
@@ -147,55 +145,55 @@ func (r *Runner) HasMessenger(id string) bool {
 // subscribers and pushes messages to them for each queued campaign
 // until all subscribers are exhausted, at which point, a campaign is marked
 // as "finished".
-func (r *Runner) Run(tick time.Duration) {
+func (m *Manager) Run(tick time.Duration) {
 	go func() {
 		t := time.NewTicker(tick)
 		for {
 			select {
 			// Periodically scan the data source for campaigns to process.
 			case <-t.C:
-				campaigns, err := r.src.NextCampaigns(r.getPendingCampaignIDs())
+				campaigns, err := m.src.NextCampaigns(m.getPendingCampaignIDs())
 				if err != nil {
-					r.logger.Printf("error fetching campaigns: %v", err)
+					m.logger.Printf("error fetching campaigns: %v", err)
 					continue
 				}
 
 				for _, c := range campaigns {
-					if err := r.addCampaign(c); err != nil {
-						r.logger.Printf("error processing campaign (%s): %v", c.Name, err)
+					if err := m.addCampaign(c); err != nil {
+						m.logger.Printf("error processing campaign (%s): %v", c.Name, err)
 						continue
 					}
-					r.logger.Printf("start processing campaign (%s)", c.Name)
+					m.logger.Printf("start processing campaign (%s)", c.Name)
 
 					// If subscriber processing is busy, move on. Blocking and waiting
 					// can end up in a race condition where the waiting campaign's
 					// state in the data source has changed.
 					select {
-					case r.subFetchQueue <- c:
+					case m.subFetchQueue <- c:
 					default:
 					}
 				}
 
 				// Aggregate errors from sending messages to check against the error threshold
 				// after which a campaign is paused.
-			case e := <-r.msgErrorQueue:
-				if r.cfg.MaxSendErrors < 1 {
+			case e := <-m.msgErrorQueue:
+				if m.cfg.MaxSendErrors < 1 {
 					continue
 				}
 
 				// If the error threshold is met, pause the campaign.
-				r.msgErrorCounts[e.camp.ID]++
-				if r.msgErrorCounts[e.camp.ID] >= r.cfg.MaxSendErrors {
-					r.logger.Printf("error counted exceeded %d. pausing campaign %s",
-						r.cfg.MaxSendErrors, e.camp.Name)
+				m.msgErrorCounts[e.camp.ID]++
+				if m.msgErrorCounts[e.camp.ID] >= m.cfg.MaxSendErrors {
+					m.logger.Printf("error counted exceeded %d. pausing campaign %s",
+						m.cfg.MaxSendErrors, e.camp.Name)
 
-					if r.isCampaignProcessing(e.camp.ID) {
-						r.exhaustCampaign(e.camp, models.CampaignStatusPaused)
+					if m.isCampaignProcessing(e.camp.ID) {
+						m.exhaustCampaign(e.camp, models.CampaignStatusPaused)
 					}
-					delete(r.msgErrorCounts, e.camp.ID)
+					delete(m.msgErrorCounts, e.camp.ID)
 
 					// Notify admins.
-					r.sendNotif(e.camp,
+					m.sendNotif(e.camp,
 						models.CampaignStatusPaused,
 						"Too many errors")
 				}
@@ -204,50 +202,50 @@ func (r *Runner) Run(tick time.Duration) {
 	}()
 
 	// Fetch the next set of subscribers for a campaign and process them.
-	for c := range r.subFetchQueue {
-		has, err := r.nextSubscribers(c, batchSize)
+	for c := range m.subFetchQueue {
+		has, err := m.nextSubscribers(c, batchSize)
 		if err != nil {
-			r.logger.Printf("error processing campaign batch (%s): %v", c.Name, err)
+			m.logger.Printf("error processing campaign batch (%s): %v", c.Name, err)
 			continue
 		}
 
 		if has {
 			// There are more subscribers to fetch.
-			r.subFetchQueue <- c
-		} else if r.isCampaignProcessing(c.ID) {
+			m.subFetchQueue <- c
+		} else if m.isCampaignProcessing(c.ID) {
 			// There are no more subscribers. Either the campaign status
 			// has changed or all subscribers have been processed.
-			newC, err := r.exhaustCampaign(c, "")
+			newC, err := m.exhaustCampaign(c, "")
 			if err != nil {
-				r.logger.Printf("error exhausting campaign (%s): %v", c.Name, err)
+				m.logger.Printf("error exhausting campaign (%s): %v", c.Name, err)
 				continue
 			}
-			r.sendNotif(newC, newC.Status, "")
+			m.sendNotif(newC, newC.Status, "")
 		}
 	}
 
 }
 
 // SpawnWorkers spawns workers goroutines that push out messages.
-func (r *Runner) SpawnWorkers() {
-	for i := 0; i < r.cfg.Concurrency; i++ {
+func (m *Manager) SpawnWorkers() {
+	for i := 0; i < m.cfg.Concurrency; i++ {
 		go func() {
-			for m := range r.msgQueue {
-				if !r.isCampaignProcessing(m.Campaign.ID) {
+			for msg := range m.msgQueue {
+				if !m.isCampaignProcessing(msg.Campaign.ID) {
 					continue
 				}
 
-				err := r.messengers[m.Campaign.MessengerID].Push(
-					m.from,
-					[]string{m.to},
-					m.Campaign.Subject,
-					m.Body)
+				err := m.messengers[msg.Campaign.MessengerID].Push(
+					msg.from,
+					[]string{msg.to},
+					msg.Campaign.Subject,
+					msg.Body)
 				if err != nil {
-					r.logger.Printf("error sending message in campaign %s: %v",
-						m.Campaign.Name, err)
+					m.logger.Printf("error sending message in campaign %s: %v",
+						msg.Campaign.Name, err)
 
 					select {
-					case r.msgErrorQueue <- msgError{camp: m.Campaign, err: err}:
+					case m.msgErrorQueue <- msgError{camp: msg.Campaign, err: err}:
 					default:
 					}
 				}
@@ -257,28 +255,28 @@ func (r *Runner) SpawnWorkers() {
 }
 
 // addCampaign adds a campaign to the process queue.
-func (r *Runner) addCampaign(c *models.Campaign) error {
+func (m *Manager) addCampaign(c *models.Campaign) error {
 	// Validate messenger.
-	if _, ok := r.messengers[c.MessengerID]; !ok {
-		r.src.UpdateCampaignStatus(c.ID, models.CampaignStatusCancelled)
+	if _, ok := m.messengers[c.MessengerID]; !ok {
+		m.src.UpdateCampaignStatus(c.ID, models.CampaignStatusCancelled)
 		return fmt.Errorf("unknown messenger %s on campaign %s", c.MessengerID, c.Name)
 	}
 
 	// Load the template.
-	if err := c.CompileTemplate(r.TemplateFuncs(c)); err != nil {
+	if err := c.CompileTemplate(m.TemplateFuncs(c)); err != nil {
 		return err
 	}
 
 	// Add the campaign to the active map.
-	r.camps[c.ID] = c
+	m.camps[c.ID] = c
 	return nil
 }
 
 // getPendingCampaignIDs returns the IDs of campaigns currently being processed.
-func (r *Runner) getPendingCampaignIDs() []int64 {
+func (m *Manager) getPendingCampaignIDs() []int64 {
 	// Needs to return an empty slice in case there are no campaigns.
 	ids := make([]int64, 0)
-	for _, c := range r.camps {
+	for _, c := range m.camps {
 		ids = append(ids, int64(c.ID))
 	}
 
@@ -289,9 +287,9 @@ func (r *Runner) getPendingCampaignIDs() []int64 {
 // If returns a bool indicating whether there any subscribers were processed
 // in the current batch or not. This can happen when all the subscribers
 // have been processed, or if a campaign has been paused or cancelled abruptly.
-func (r *Runner) nextSubscribers(c *models.Campaign, batchSize int) (bool, error) {
+func (m *Manager) nextSubscribers(c *models.Campaign, batchSize int) (bool, error) {
 	// Fetch a batch of subscribers.
-	subs, err := r.src.NextSubscribers(c.ID, batchSize)
+	subs, err := m.src.NextSubscribers(c.ID, batchSize)
 	if err != nil {
 		return false, fmt.Errorf("error fetching campaign subscribers (%s): %v", c.Name, err)
 	}
@@ -303,42 +301,42 @@ func (r *Runner) nextSubscribers(c *models.Campaign, batchSize int) (bool, error
 
 	// Push messages.
 	for _, s := range subs {
-		m := r.NewMessage(c, s)
-		if err := m.Render(); err != nil {
-			r.logger.Printf("error rendering message (%s) (%s): %v", c.Name, s.Email, err)
+		msg := m.NewMessage(c, s)
+		if err := msg.Render(); err != nil {
+			m.logger.Printf("error rendering message (%s) (%s): %v", c.Name, s.Email, err)
 			continue
 		}
 
 		// Push the message to the queue while blocking and waiting until
 		// the queue is drained.
-		r.msgQueue <- m
+		m.msgQueue <- msg
 	}
 
 	return true, nil
 }
 
 // isCampaignProcessing checks if the campaign is bing processed.
-func (r *Runner) isCampaignProcessing(id int) bool {
-	_, ok := r.camps[id]
+func (m *Manager) isCampaignProcessing(id int) bool {
+	_, ok := m.camps[id]
 	return ok
 }
 
-func (r *Runner) exhaustCampaign(c *models.Campaign, status string) (*models.Campaign, error) {
-	delete(r.camps, c.ID)
+func (m *Manager) exhaustCampaign(c *models.Campaign, status string) (*models.Campaign, error) {
+	delete(m.camps, c.ID)
 
 	// A status has been passed. Change the campaign's status
 	// without further checks.
 	if status != "" {
-		if err := r.src.UpdateCampaignStatus(c.ID, status); err != nil {
-			r.logger.Printf("error updating campaign (%s) status to %s: %v", c.Name, status, err)
+		if err := m.src.UpdateCampaignStatus(c.ID, status); err != nil {
+			m.logger.Printf("error updating campaign (%s) status to %s: %v", c.Name, status, err)
 		} else {
-			r.logger.Printf("set campaign (%s) to %s", c.Name, status)
+			m.logger.Printf("set campaign (%s) to %s", c.Name, status)
 		}
 		return c, nil
 	}
 
 	// Fetch the up-to-date campaign status from the source.
-	cm, err := r.src.GetCampaign(c.ID)
+	cm, err := m.src.GetCampaign(c.ID)
 	if err != nil {
 		return nil, err
 	}
@@ -346,13 +344,13 @@ func (r *Runner) exhaustCampaign(c *models.Campaign, status string) (*models.Cam
 	// If a running campaign has exhausted subscribers, it's finished.
 	if cm.Status == models.CampaignStatusRunning {
 		cm.Status = models.CampaignStatusFinished
-		if err := r.src.UpdateCampaignStatus(c.ID, models.CampaignStatusFinished); err != nil {
-			r.logger.Printf("error finishing campaign (%s): %v", c.Name, err)
+		if err := m.src.UpdateCampaignStatus(c.ID, models.CampaignStatusFinished); err != nil {
+			m.logger.Printf("error finishing campaign (%s): %v", c.Name, err)
 		} else {
-			r.logger.Printf("campaign (%s) finished", c.Name)
+			m.logger.Printf("campaign (%s) finished", c.Name)
 		}
 	} else {
-		r.logger.Printf("stop processing campaign (%s)", c.Name)
+		m.logger.Printf("stop processing campaign (%s)", c.Name)
 	}
 
 	return cm, nil
@@ -371,32 +369,32 @@ func (m *Message) Render() error {
 
 // trackLink register a URL and return its UUID to be used in message templates
 // for tracking links.
-func (r *Runner) trackLink(url, campUUID, subUUID string) string {
-	r.linksMutex.RLock()
-	if uu, ok := r.links[url]; ok {
-		r.linksMutex.RUnlock()
-		return fmt.Sprintf(r.cfg.LinkTrackURL, uu, campUUID, subUUID)
+func (m *Manager) trackLink(url, campUUID, subUUID string) string {
+	m.linksMutex.RLock()
+	if uu, ok := m.links[url]; ok {
+		m.linksMutex.RUnlock()
+		return fmt.Sprintf(m.cfg.LinkTrackURL, uu, campUUID, subUUID)
 	}
-	r.linksMutex.RUnlock()
+	m.linksMutex.RUnlock()
 
 	// Register link.
-	uu, err := r.src.CreateLink(url)
+	uu, err := m.src.CreateLink(url)
 	if err != nil {
-		r.logger.Printf("error registering tracking for link '%s': %v", url, err)
+		m.logger.Printf("error registering tracking for link '%s': %v", url, err)
 
 		// If the registration fails, fail over to the original URL.
 		return url
 	}
 
-	r.linksMutex.Lock()
-	r.links[url] = uu
-	r.linksMutex.Unlock()
+	m.linksMutex.Lock()
+	m.links[url] = uu
+	m.linksMutex.Unlock()
 
-	return fmt.Sprintf(r.cfg.LinkTrackURL, uu, campUUID, subUUID)
+	return fmt.Sprintf(m.cfg.LinkTrackURL, uu, campUUID, subUUID)
 }
 
 // sendNotif sends a notification to registered admin e-mails.
-func (r *Runner) sendNotif(c *models.Campaign, status, reason string) error {
+func (m *Manager) sendNotif(c *models.Campaign, status, reason string) error {
 	var (
 		subject = fmt.Sprintf("%s: %s", strings.Title(status), c.Name)
 		data    = map[string]interface{}{
@@ -409,19 +407,19 @@ func (r *Runner) sendNotif(c *models.Campaign, status, reason string) error {
 		}
 	)
 
-	return r.notifCB(subject, data)
+	return m.notifCB(subject, data)
 }
 
 // TemplateFuncs returns the template functions to be applied into
 // compiled campaign templates.
-func (r *Runner) TemplateFuncs(c *models.Campaign) template.FuncMap {
+func (m *Manager) TemplateFuncs(c *models.Campaign) template.FuncMap {
 	return template.FuncMap{
 		"TrackLink": func(url, campUUID, subUUID string) string {
-			return r.trackLink(url, campUUID, subUUID)
+			return m.trackLink(url, campUUID, subUUID)
 		},
 		"TrackView": func(campUUID, subUUID string) template.HTML {
 			return template.HTML(fmt.Sprintf(`<img src="%s" alt="" />`,
-				fmt.Sprintf(r.cfg.ViewTrackURL, campUUID, subUUID)))
+				fmt.Sprintf(m.cfg.ViewTrackURL, campUUID, subUUID)))
 		},
 		"Date": func(layout string) string {
 			if layout == "" {

runner_db.go → manager_db.go

@@ -12,7 +12,7 @@ type runnerDB struct {
 	queries *Queries
 }
 
-func newRunnerDB(q *Queries) *runnerDB {
+func newManagerDB(q *Queries) *runnerDB {
 	return &runnerDB{
 		queries: q,
 	}

templates.go

@@ -111,12 +111,12 @@ func handlePreviewTemplate(c echo.Context) error {
 		Body:         dummyTpl,
 	}
 
-	if err := camp.CompileTemplate(app.Runner.TemplateFuncs(&camp)); err != nil {
+	if err := camp.CompileTemplate(app.Manager.TemplateFuncs(&camp)); err != nil {
 		return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Error compiling template: %v", err))
 	}
 
 	// Render the message body.
-	m := app.Runner.NewMessage(&camp, &dummySubscriber)
+	m := app.Manager.NewMessage(&camp, &dummySubscriber)
 	if err := m.Render(); err != nil {
 		return echo.NewHTTPError(http.StatusBadRequest,
 			fmt.Sprintf("Error rendering message: %v", err))