Add job

Author: gedorinku

.env.ci

@@ -25,4 +25,6 @@ REDIS_ADDR=localhost:6379
 
 DEBUG_LOG=true
 
+JOB_INTERVAL_SEC=120
+
 HYDRA_ADMIN_URL=http://hydra:4445

.env.sample

@@ -26,4 +26,6 @@ REDIS_ADDR=redis:6379
 
 DEBUG_LOG=true
 
+JOB_INTERVAL_SEC=120
+
 HYDRA_ADMIN_URL=http://hydra:4445

app/config/config.go

@@ -11,6 +11,7 @@ type Config struct {
 	DataBaseURL     string `envconfig:"database_url" required:"true"`
 	RedisAddr       string `envconfig:"redis_addr" required:"true"`
 	DebugLog        bool   `envconfig:"debug_log"`
+	JobIntervalSec  int    `envconfig:"job_interval_sec" required:"true"`
 	HydraAdminURL   string `envconfig:"hydra_admin_url" required:"true"`
 	MinioPublicURL  string `envconfig:"minio_public_url" required:"true"`
 	MinioEndpoint   string `envconfig:"minio_endpoint" required:"true"`

app/job/job.go

@@ -0,0 +1,65 @@
+package job
+
+import (
+	"time"
+
+	"google.golang.org/grpc/grpclog"
+
+	"github.com/ProgrammingLab/prolab-accounts/app/config"
+	"github.com/ProgrammingLab/prolab-accounts/app/di"
+	"github.com/ProgrammingLab/prolab-accounts/app/util"
+)
+
+var (
+	started = false
+	stop    = make(chan struct{})
+	jobs    = []Job{}
+)
+
+// Job represents job for worker
+type Job func(store di.StoreComponent) error
+
+// Start starts the worker
+func Start(store di.StoreComponent, cfg *config.Config) {
+	if started {
+		return
+	}
+	started = true
+
+	go func() {
+		run(store, time.Duration(cfg.JobIntervalSec)*time.Second)
+	}()
+}
+
+// Close stops the worker
+func Close() {
+	grpclog.Infoln("worker is stopping(^C to force to stop)")
+	stop <- struct{}{}
+}
+
+func run(store di.StoreComponent, interval time.Duration) {
+	defer func() {
+		if err := util.ErrorFromRecover(recover()); err != nil {
+			grpclog.Errorf("job panic: %+v", err)
+			grpclog.Infoln("worker is restarting...")
+			run(store, interval)
+		}
+	}()
+
+	grpclog.Infof("worker started: interval %v", interval)
+
+	for {
+		select {
+		case <-time.After(interval):
+			for _, j := range jobs {
+				err := j(store)
+				if err != nil {
+					grpclog.Errorf("job error: %+v", err)
+				}
+			}
+		case <-stop:
+			grpclog.Infoln("worker stopped")
+			return
+		}
+	}
+}

app/run.go

@@ -9,6 +9,7 @@ import (
 	"github.com/ProgrammingLab/prolab-accounts/app/config"
 	"github.com/ProgrammingLab/prolab-accounts/app/di"
 	"github.com/ProgrammingLab/prolab-accounts/app/interceptor"
+	"github.com/ProgrammingLab/prolab-accounts/app/job"
 	"github.com/ProgrammingLab/prolab-accounts/app/server"
 )
 
@@ -56,5 +57,9 @@ func Run() error {
 			server.NewUserBlogServiceServer(store),
 		),
 	)
+
+	job.Start(store, cfg)
+	defer job.Close()
+
 	return s.Serve()
 }